a49db192f4
- Renamed all stacks to English with emojis (Backlog, Planning, In Progress, Complete)
- Updated sync script to use new stack names
- Created all 3 initiative cards (IDs 189-191)
- Enhanced error handling with detailed debug output
- Updated documentation with API limitations and troubleshooting
- Fixed stack fallback from 'eingang' to '📥 Backlog'
Changes:
- scripts/sync-roadmap-to-deck.py: Updated STATUS_TO_STACK mapping, added verbose logging
- docs/NEXTCLOUD_DECK_SYNC.md: Updated stack table, added Known Limitations section, enhanced troubleshooting
Note: 6 duplicate/test cards (184-188, 192) must be deleted manually from Nextcloud UI
due to API limitations (DELETE returns 405)
320 lines
11 KiB
Python
Executable File
320 lines
11 KiB
Python
Executable File
#!/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()
|
||
tasks = []
|
||
|
||
# Parse initiatives (main sections) - handle different emoji prefixes
|
||
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("<EFBFBD> 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()
|