#!/usr/bin/env python3 """ Nextcloud Deck <-> Trading Bot Roadmap Sync Syncs roadmap markdown files with Nextcloud Deck kanban board """ import json import re import sys import os import requests from pathlib import Path from typing import Dict, List, Optional from datetime import datetime, timedelta import argparse # Configuration DECK_CONFIG_FILE = '/tmp/deck-config.json' ROADMAP_DIR = Path('/home/icke/traderv4') # Load Nextcloud Deck config with open(DECK_CONFIG_FILE, 'r') as f: DECK_CONFIG = json.load(f) NEXTCLOUD_URL = DECK_CONFIG['url'] NEXTCLOUD_USER = DECK_CONFIG['user'] NEXTCLOUD_PASSWORD = DECK_CONFIG['password'] BOARD_ID = DECK_CONFIG['board_id'] # Stack mapping STACKS = {stack['name']: stack['id'] for stack in DECK_CONFIG['stacks']} # Status to stack mapping STATUS_TO_STACK = { 'IN PROGRESS': '🚀 In Progress', '🔄 IN PROGRESS': '🚀 In Progress', 'COMPLETE': '✅ Complete', '✅ COMPLETE': '✅ Complete', 'PENDING': '📋 Planning', '🔜 NEXT': '📋 Planning', 'FUTURE': '📥 Backlog', '🎯 FUTURE': '📥 Backlog', } # Roadmap files to parse ROADMAP_FILES = [ 'OPTIMIZATION_MASTER_ROADMAP.md', 'SIGNAL_QUALITY_OPTIMIZATION_ROADMAP.md', 'POSITION_SCALING_ROADMAP.md', 'ATR_BASED_TP_ROADMAP.md', ] class DeckAPI: """Nextcloud Deck API wrapper""" def __init__(self): self.base_url = f"{NEXTCLOUD_URL}/index.php/apps/deck/api/v1.0" self.auth = (NEXTCLOUD_USER, NEXTCLOUD_PASSWORD) self.headers = { 'OCS-APIRequest': 'true', 'Content-Type': 'application/json' } def get_board_details(self) -> Dict: """Get full board details including stacks and cards""" url = f"{self.base_url}/boards/{BOARD_ID}" response = requests.get(url, auth=self.auth, headers=self.headers, verify=False) response.raise_for_status() return response.json() def get_cards_in_stack(self, stack_id: int) -> List[Dict]: """Get all cards in a specific stack""" url = f"{self.base_url}/stacks/{stack_id}/cards" response = requests.get(url, auth=self.auth, headers=self.headers, verify=False) response.raise_for_status() return response.json() def create_card(self, stack_id: int, title: str, description: str, due_date: Optional[str] = None, labels: Optional[List[str]] = None) -> Dict: """Create a new card""" url = f"{self.base_url}/boards/{BOARD_ID}/stacks/{stack_id}/cards" payload = { 'title': title, 'type': 'plain', 'order': 999, 'description': description, } if due_date: payload['duedate'] = due_date response = requests.post(url, auth=self.auth, headers=self.headers, json=payload, verify=False) response.raise_for_status() return response.json() def update_card(self, card_id: int, **kwargs) -> Dict: """Update an existing card""" url = f"{self.base_url}/boards/{BOARD_ID}/cards/{card_id}" response = requests.put(url, auth=self.auth, headers=self.headers, json=kwargs, verify=False) response.raise_for_status() return response.json() def get_all_cards(self) -> List[Dict]: """Get all cards from all stacks""" all_cards = [] for stack_name, stack_id in STACKS.items(): try: cards = self.get_cards_in_stack(stack_id) for card in cards: card['stack_name'] = stack_name card['stack_id'] = stack_id all_cards.extend(cards) except requests.exceptions.HTTPError as e: # Stack might be empty, continue print(f"⚠️ Warning: Could not get cards from stack '{stack_name}': {e}") continue return all_cards class RoadmapParser: """Parse roadmap markdown files""" @staticmethod def parse_file(filepath: Path) -> List[Dict]: """Parse a roadmap file and extract phases/tasks""" if not filepath.exists(): return [] content = filepath.read_text() filename = filepath.name tasks = [] # Master roadmap: Parse initiatives (high-level overview only) if filename == 'OPTIMIZATION_MASTER_ROADMAP.md': initiative_pattern = r'## (?:🎯|📐|📊) Initiative (\d+): (.+?)(?=\n##|\Z)' initiatives = re.finditer(initiative_pattern, content, re.DOTALL) for initiative in initiatives: initiative_num = initiative.group(1) initiative_title = initiative.group(2).strip() initiative_content = initiative.group(0) # Extract file reference file_match = re.search(r'\*\*File:\*\* \[`(.+?)`\]', initiative_content) file_ref = file_match.group(1) if file_match else '' # Extract purpose purpose_match = re.search(r'### Purpose\n(.+?)(?=\n###|\Z)', initiative_content, re.DOTALL) purpose = purpose_match.group(1).strip() if purpose_match else '' # Extract current status status_match = re.search(r'### Current Status\n(.+?)(?=\n###|\Z)', initiative_content, re.DOTALL) status = status_match.group(1).strip() if status_match else '' # Determine phase status phase_status = 'FUTURE' if '✅ COMPLETE' in status or 'COMPLETE' in status: phase_status = 'COMPLETE' elif '🔄 IN PROGRESS' in status or 'IN PROGRESS' in status: phase_status = 'IN PROGRESS' elif '🔜 NEXT' in status: phase_status = 'PENDING' task = { 'title': f"Initiative {initiative_num}: {initiative_title}", 'description': f"""**File:** [{file_ref}](./{file_ref}) {purpose} **Status:** {status} """, 'status': phase_status, 'progress': '', 'due_date': None, 'labels': [f'initiative-{initiative_num}'], } tasks.append(task) # Detailed roadmaps: Parse phases (use ## or ### pattern depending on file) else: # Try ## Phase pattern first (SIGNAL_QUALITY, POSITION_SCALING) phase_pattern = r'## Phase (\d+): (.+?)(?=\n(?:##|\Z))' phases = list(re.finditer(phase_pattern, content, re.DOTALL)) # If no ## Phase found, try ### Phase pattern (ATR_BASED_TP) if not phases: phase_pattern = r'### Phase (\d+): (.+?)(?=\n(?:###|##|\Z))' phases = list(re.finditer(phase_pattern, content, re.DOTALL)) for phase in phases: phase_num = phase.group(1) phase_title_raw = phase.group(2).strip() # Clean up phase title (remove status emojis and extra text in parens) phase_title = re.sub(r'\s*\(.*?\)\s*', ' ', phase_title_raw) phase_title = re.sub(r'[🔄✅🔜🎯⏳🤖📏📊🔮]+', '', phase_title).strip() # Extract full phase content for description phase_start = phase.end() next_phase_match = re.search(r'\n(?:##|###) Phase \d+:', content[phase_start:]) phase_end = phase_start + next_phase_match.start() if next_phase_match else len(content) phase_content = content[phase_start:phase_end].strip() # Remove code blocks FIRST (they cause 400/500 errors) phase_content_clean = re.sub(r'```[\s\S]*?```', '[Code block omitted - see roadmap file]', phase_content) # Then limit description to first 400 chars description_preview = phase_content_clean[:400] if len(phase_content_clean) > 400: description_preview += '...' # Determine status from title phase_status = 'FUTURE' title_upper = phase_title_raw.upper() # Check for status indicators in the raw title if ('CURRENT' in title_upper or '✅' in phase_title_raw or 'DEPLOYED' in title_upper) and 'FUTURE' not in title_upper: phase_status = 'IN PROGRESS' elif '🔜' in phase_title_raw or ('NEXT' in title_upper and 'FUTURE' not in title_upper): phase_status = 'PENDING' elif 'COMPLETE' in title_upper: phase_status = 'COMPLETE' # Otherwise stays FUTURE # Extract progress if present progress_match = re.search(r'(\d+)/(\d+)', phase_content[:200]) progress = f"{progress_match.group(1)}/{progress_match.group(2)}" if progress_match else '' task = { 'title': f"Phase {phase_num}: {phase_title}", 'description': f"""**Initiative:** {filepath.stem.replace('_', ' ').title().replace('Roadmap', '')} **File:** [{filename}](./{filename}) {description_preview} """, 'status': phase_status, 'progress': progress, 'due_date': None, 'labels': [filename.replace('.md', '').lower(), f'phase-{phase_num}'], } tasks.append(task) return tasks def sync_roadmap_to_deck(dry_run: bool = False): """Sync roadmap files to Nextcloud Deck""" print("🔄 Syncing roadmap to Nextcloud Deck...") print() api = DeckAPI() parser = RoadmapParser() # Note: Nextcloud Deck API doesn't support GET cards by stack in this version # We'll create cards and rely on manual deduplication for now print("� Creating cards from roadmap (manual deduplication required)") print() # Parse all roadmap files all_tasks = [] for filename in ROADMAP_FILES: filepath = ROADMAP_DIR / filename if filepath.exists(): print(f"📖 Parsing {filename}...") tasks = parser.parse_file(filepath) all_tasks.extend(tasks) print(f" Found {len(tasks)} tasks") print() print(f"📝 Total tasks from roadmaps: {len(all_tasks)}") print() # Create/update cards created = 0 skipped = 0 for task in all_tasks: title = task['title'] # Create new card stack_name = STATUS_TO_STACK.get(task['status'], '📥 Backlog') # Debug output print(f"✨ Create: {title}") print(f" Status from roadmap: {task['status']!r}") print(f" Mapped to stack: {stack_name!r}") if stack_name not in STACKS: print(f" ❌ ERROR: Stack {stack_name!r} not found in STACKS") print(f" Available stacks: {list(STACKS.keys())}") skipped += 1 continue stack_id = STACKS[stack_name] print(f" Stack ID: {stack_id}") if task['progress']: print(f" Progress: {task['progress']}") if not dry_run: try: result = api.create_card( stack_id=stack_id, title=title, description=task['description'], due_date=task['due_date'], ) print(f" ✅ Created card ID: {result['id']}") created += 1 except Exception as e: print(f" ❌ Error: {e}") import traceback traceback.print_exc() skipped += 1 else: created += 1 print() print("=" * 60) print(f"✅ Sync complete!") print(f" Created: {created}") print(f" Skipped: {skipped}") if dry_run: print() print(" (Dry run - no changes made)") def main(): parser = argparse.ArgumentParser(description='Sync roadmap with Nextcloud Deck') parser.add_argument('--init', action='store_true', help='Initialize: create cards from roadmap') parser.add_argument('--sync', action='store_true', help='Sync: update existing cards') parser.add_argument('--dry-run', action='store_true', help='Show what would be done without making changes') args = parser.parse_args() if not args.init and not args.sync: parser.print_help() sys.exit(1) try: sync_roadmap_to_deck(dry_run=args.dry_run) except Exception as e: print(f"❌ Error: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == '__main__': main()