#!/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 arbeit', '🔄 IN PROGRESS': 'in arbeit', 'COMPLETE': 'erledigt', '✅ COMPLETE': 'erledigt', 'PENDING': 'in planung', '🔜 NEXT': 'in planung', 'FUTURE': 'eingang', '🎯 FUTURE': 'eingang', } # 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() tasks = [] # Parse initiatives (main sections) 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 '' # Extract timeline timeline_match = re.search(r'### Timeline\n(.+?)(?=\n###|\Z)', initiative_content, re.DOTALL) timeline = timeline_match.group(1).strip() if timeline_match else '' # Extract success metrics metrics_match = re.search(r'### Success Metrics\n(.+?)(?=\n###|\Z)', initiative_content, re.DOTALL) metrics = metrics_match.group(1).strip() if metrics_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 or 'Phase 2' in status: phase_status = 'PENDING' # Extract progress progress_match = re.search(r'Progress:.*?(\d+)/(\d+)', status) progress = f"{progress_match.group(1)}/{progress_match.group(2)}" if progress_match else '' # Calculate due date from timeline due_date = None if 'weeks' in timeline.lower(): weeks = re.search(r'(\d+)-?(\d+)?\s*weeks', timeline.lower()) if weeks: max_weeks = int(weeks.group(2) if weeks.group(2) else weeks.group(1)) due_date = (datetime.now() + timedelta(weeks=max_weeks)).strftime('%Y-%m-%d') task = { 'title': f"Initiative {initiative_num}: {initiative_title}", 'description': f"""**Initiative {initiative_num}: {initiative_title}** {purpose} **Current Status:** {status} **Timeline:** {timeline} **Success Metrics:** {metrics} **File:** {file_ref} """, 'status': phase_status, 'initiative_num': initiative_num, 'progress': progress, 'due_date': due_date, 'labels': [f'initiative-{initiative_num}', 'optimization'], } 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'], 'eingang') stack_id = STACKS[stack_name] print(f"✨ Create: {title}") print(f" Stack: {stack_name}") if task['progress']: print(f" Progress: {task['progress']}") if not dry_run: try: api.create_card( stack_id=stack_id, title=title, description=task['description'], due_date=task['due_date'], ) created += 1 except Exception as e: print(f" ❌ Error: {e}") 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()