From c46d53b261a3faccbbdd26cd443717c5426eb845 Mon Sep 17 00:00:00 2001 From: Tim Bendt Date: Wed, 20 Aug 2025 08:30:54 -0400 Subject: [PATCH] godspeed app sync --- GODSPEED_SYNC.md | 250 ++++++++++++ TASK_SWEEPER.md | 134 +++++++ demo_cancelled_workflow.py | 110 ++++++ demo_completion_sync.py | 103 +++++ pyproject.toml | 2 + src/cli/__init__.py | 4 + src/cli/godspeed.py | 616 ++++++++++++++++++++++++++++++ src/cli/sync.py | 298 ++++++++++++--- src/services/godspeed/__init__.py | 0 src/services/godspeed/client.py | 129 +++++++ src/services/godspeed/config.py | 87 +++++ src/services/godspeed/sync.py | 395 +++++++++++++++++++ sweep_tasks.py | 381 ++++++++++++++++++ test_cancelled_tasks.py | 95 +++++ test_completion_status.py | 123 ++++++ test_godspeed_sync.py | 270 +++++++++++++ test_task_sweeper.py | 218 +++++++++++ 17 files changed, 3170 insertions(+), 45 deletions(-) create mode 100644 GODSPEED_SYNC.md create mode 100644 TASK_SWEEPER.md create mode 100644 demo_cancelled_workflow.py create mode 100644 demo_completion_sync.py create mode 100644 src/cli/godspeed.py create mode 100644 src/services/godspeed/__init__.py create mode 100644 src/services/godspeed/client.py create mode 100644 src/services/godspeed/config.py create mode 100644 src/services/godspeed/sync.py create mode 100755 sweep_tasks.py create mode 100644 test_cancelled_tasks.py create mode 100644 test_completion_status.py create mode 100644 test_godspeed_sync.py create mode 100644 test_task_sweeper.py diff --git a/GODSPEED_SYNC.md b/GODSPEED_SYNC.md new file mode 100644 index 0000000..3b4d18c --- /dev/null +++ b/GODSPEED_SYNC.md @@ -0,0 +1,250 @@ +# Godspeed Sync + +A two-way synchronization tool between the Godspeed task management API and local markdown files. + +## Features + +- **Bidirectional Sync**: Download tasks from Godspeed to markdown files, edit locally, and upload changes back +- **Directory Structure**: Creates a clean directory structure matching your Godspeed lists +- **ID Tracking**: Uses hidden HTML comments to track task IDs even when you rearrange tasks +- **Markdown Format**: Simple `- [ ] Task name ` format for easy editing +- **Completion Status**: Supports incomplete `[ ]`, completed `[x]`, and cancelled `[-]` checkboxes +- **Notes Support**: Task notes are preserved and synced +- **CLI Interface**: Easy-to-use command line interface with shortcuts + +## Installation + +The Godspeed sync is part of the GTD Terminal Tools project. Make sure you have the required dependencies: + +```bash +# Install dependencies +uv sync + +# Or with pip +pip install requests click +``` + +## Configuration + +### Option 1: Environment Variables +```bash +export GODSPEED_EMAIL="your@email.com" +export GODSPEED_PASSWORD="your-password" + +# OR use an API token directly +export GODSPEED_TOKEN="your-api-token" + +# Optional: Custom sync directory +export GODSPEED_SYNC_DIR="~/Documents/MyTasks" + +# Optional: Disable SSL verification for corporate networks +export GODSPEED_DISABLE_SSL_VERIFY="true" +``` + +### Option 2: Interactive Setup +The tool will prompt for credentials if not provided via environment variables. + +### Getting an API Token +You can get your API token from the Godspeed desktop app: +1. Open the Command Palette (Cmd/Ctrl + Shift + P) +2. Run "Copy API access token" +3. Use this token with `GODSPEED_TOKEN` environment variable + +## Usage + +### Basic Commands + +```bash +# Download all tasks from Godspeed to local markdown files +python -m src.cli.godspeed download +# OR use the short alias +python -m src.cli.godspeed download # 'gs' will be available when integrated + +# Upload local changes back to Godspeed +python -m src.cli.godspeed upload + +# Bidirectional sync (download then upload) +python -m src.cli.godspeed sync + +# Check sync status +python -m src.cli.godspeed status + +# Open sync directory in file manager +python -m src.cli.godspeed open + +# Test connection and SSL (helpful for corporate networks) +python -m src.cli.godspeed test-connection +``` + +### Workflow Example + +1. **Initial sync**: + ```bash + python -m src.cli.godspeed download + ``` + +2. **Edit tasks locally**: + Open the generated markdown files in your favorite editor: + ``` + ~/Documents/Godspeed/ + ├── Personal.md + ├── Work_Projects.md + └── Shopping.md + ``` + +3. **Make changes**: + ```markdown + # Personal.md + - [ ] Call dentist + - [x] Buy groceries + Don't forget milk and eggs + - [-] Old project + - [ ] New task I just added + + # Work_Projects.md + - [ ] Finish quarterly report + Due Friday + - [-] Cancelled meeting + ``` + +4. **Sync changes back**: + ```bash + python -m src.cli.godspeed upload + ``` + +## File Format + +Each list becomes a markdown file with tasks in this format: + +```markdown +- [ ] Incomplete task +- [x] Completed task +- [X] Also completed (capital X works too) +- [-] Cancelled/cleared task +- [ ] Task with notes + Notes go on the next line, indented +``` + +### Important Notes: +- **Don't remove the `` comments** - they're used to track tasks +- **Don't worry about the IDs** - they're auto-generated for new tasks +- **Checkbox format matters**: + - Use `[ ]` for incomplete tasks + - Use `[x]` or `[X]` for completed tasks + - Use `[-]` for cancelled/cleared tasks +- **Completion status syncs both ways**: + - Check/uncheck boxes in markdown → syncs to Godspeed + - Mark complete/incomplete/cleared in Godspeed → syncs to markdown +- **Completed/cancelled tasks are hidden**: When downloading from Godspeed, only incomplete tasks appear in local files (keeps them clean) +- **Notes are optional** - indent them under the task line +- **File names** correspond to list names (special characters replaced with underscores) + +## Directory Structure + +By default, files are synced to: +- `~/Documents/Godspeed/` (if Documents folder exists) +- `~/.local/share/gtd-terminal-tools/godspeed/` (fallback) + +Each Godspeed list becomes a `.md` file: +- "Personal" → `Personal.md` +- "Work Projects" → `Work_Projects.md` +- "Shopping List" → `Shopping_List.md` + +## Sync Metadata + +The tool stores sync metadata in `.godspeed_metadata.json`: +```json +{ + "task_mapping": { + "local-id-1": "godspeed-task-id-1", + "local-id-2": "godspeed-task-id-2" + }, + "list_mapping": { + "Personal": "godspeed-list-id-1", + "Work Projects": "godspeed-list-id-2" + }, + "last_sync": "2024-01-15T10:30:00" +} +``` + +## API Rate Limits + +Godspeed has rate limits: +- **Listing**: 10 requests/minute, 200/hour +- **Creating/Updating**: 60 requests/minute, 1,000/hour + +The sync tool respects these limits and handles errors gracefully. + +## Troubleshooting + +### SSL/Corporate Network Issues +If you're getting SSL certificate errors on a corporate network: + +```bash +# Test the connection first +python -m src.cli.godspeed test-connection + +# If SSL errors occur, bypass SSL verification +export GODSPEED_DISABLE_SSL_VERIFY=true +python -m src.cli.godspeed test-connection +``` + +### Authentication Issues +```bash +# Clear stored credentials +rm ~/.local/share/gtd-terminal-tools/godspeed_config.json + +# Use token instead of password +export GODSPEED_TOKEN="your-token-here" +``` + +### Sync Issues +```bash +# Check current status +python -m src.cli.godspeed status + +# Verify sync directory +ls ~/Documents/Godspeed/ + +# Check metadata +cat ~/.local/share/gtd-terminal-tools/godspeed/.godspeed_metadata.json +``` + +### Common Problems + +1. **"List ID not found"**: New lists created locally will put tasks in your Inbox +2. **"Task not found"**: Tasks deleted in Godspeed won't sync back +3. **Duplicate tasks**: Don't manually copy task lines between files (IDs must be unique) + +## Development + +### Testing +Run the test suite: +```bash +python test_godspeed_sync.py +``` + +### File Structure +``` +src/services/godspeed/ +├── __init__.py # Package init +├── client.py # Godspeed API client +├── sync.py # Sync engine +└── config.py # Configuration management + +src/cli/ +└── godspeed.py # CLI interface +``` + +## Contributing + +This is part of the larger GTD Terminal Tools project. When contributing: + +1. Follow the existing code style +2. Add tests for new functionality +3. Update this README for user-facing changes +4. Test with the mock data before real API calls + +## License + +Same as the parent GTD Terminal Tools project. \ No newline at end of file diff --git a/TASK_SWEEPER.md b/TASK_SWEEPER.md new file mode 100644 index 0000000..2e53699 --- /dev/null +++ b/TASK_SWEEPER.md @@ -0,0 +1,134 @@ +# Task Sweeper for Godspeed + +A utility script to consolidate scattered incomplete tasks from markdown files into your Godspeed Inbox. + +## Purpose + +If you have notes scattered across directories (like `2024/`, `2025/`, project folders, etc.) with incomplete tasks in markdown format, this script will: + +1. **Find all incomplete tasks** (`- [ ] Task name`) in markdown files +2. **Move them** to your Godspeed `Inbox.md` file +3. **Preserve completed/cancelled tasks** in their original locations +4. **Add source tracking** so you know where each task came from +5. **Clean up original files** by removing only the incomplete tasks + +## Usage + +```bash +# Dry run to see what would happen +python sweep_tasks.py ~/Documents/Notes ~/Documents/Godspeed --dry-run + +# Actually perform the sweep +python sweep_tasks.py ~/Documents/Notes ~/Documents/Godspeed + +# Sweep from current directory +python sweep_tasks.py . ./godspeed +``` + +## Example Workflow + +**Before sweeping:** +``` +~/Documents/Notes/ +├── 2024/ +│ ├── projects/website.md +│ │ ├── - [x] Create wireframes +│ │ ├── - [ ] Design mockups ← Will be swept +│ │ └── - [ ] Get approval ← Will be swept +│ └── notes/meeting.md +│ ├── - [ ] Update docs ← Will be swept +│ └── - [x] Fix bug (completed) +├── 2025/ +│ └── goals.md +│ └── - [ ] Launch feature ← Will be swept +└── random-notes.md + └── - [ ] Call dentist ← Will be swept +``` + +**After sweeping:** +``` +~/Documents/Godspeed/ +└── Inbox.md ← All incomplete tasks here + ├── - [ ] Design mockups + │ From: 2024/projects/website.md + ├── - [ ] Get approval + │ From: 2024/projects/website.md + ├── - [ ] Update docs + │ From: 2024/notes/meeting.md + ├── - [ ] Launch feature + │ From: 2025/goals.md + └── - [ ] Call dentist + From: random-notes.md + +~/Documents/Notes/ +├── 2024/ +│ ├── projects/website.md ← Only completed tasks remain +│ │ └── - [x] Create wireframes +│ └── notes/meeting.md +│ └── - [x] Fix bug (completed) +├── 2025/ +│ └── goals.md ← File cleaned/deleted if empty +└── random-notes.md ← File cleaned/deleted if empty +``` + +## Features + +- **Safe Operation**: Always use `--dry-run` first to preview changes +- **Source Tracking**: Each swept task includes a note about its origin +- **Selective Processing**: Only moves incomplete tasks, preserves completed ones +- **Smart Cleanup**: Removes empty files or keeps non-task content +- **Godspeed Integration**: Creates properly formatted tasks with IDs for sync +- **Recursive Search**: Finds markdown files in all subdirectories +- **Exclusion Logic**: Skips the Godspeed directory itself and hidden files + +## Integration with Godspeed Sync + +After sweeping tasks: + +1. **Review** the consolidated tasks in `Inbox.md` +2. **Upload to API**: Run `python -m src.cli godspeed upload` +3. **Organize in Godspeed**: Move tasks from Inbox to appropriate lists +4. **Sync back**: Run `python -m src.cli godspeed sync` to get organized structure + +## Safety Features + +- **Dry run mode** shows exactly what will happen without making changes +- **Backup recommendation**: The script modifies files, so backup your notes first +- **Preserve content**: Non-task content (headings, notes, etc.) remains in original files +- **Completed task preservation**: `[x]` and `[-]` tasks stay where they are +- **Error handling**: Graceful handling of unreadable files or parsing errors + +## Example Output + +``` +🧹 Sweeping incomplete tasks from: /Users/you/Documents/Notes +📥 Target Inbox: /Users/you/Documents/Godspeed/Inbox.md +🔍 Dry run: False +============================================================ + +📁 Found 8 markdown files to process + +📄 Processing: 2024/projects/website.md + 🔄 Found 2 incomplete tasks: + • Design mockups + • Get client approval + ✅ Keeping 1 completed/cleared tasks in place + ✂️ Cleaned file (removed tasks): 2024/projects/website.md + +📥 Writing 6 tasks to Inbox... + ✅ Inbox updated: /Users/you/Documents/Godspeed/Inbox.md + +============================================================ +📊 SWEEP SUMMARY: + • Files processed: 3 + • Tasks swept: 6 + • Target: /Users/you/Documents/Godspeed/Inbox.md + +🎉 Successfully swept 6 tasks! +💡 Next steps: + 1. Review tasks in: /Users/you/Documents/Godspeed/Inbox.md + 2. Run 'godspeed upload' to sync to API + 3. Organize tasks into appropriate lists in Godspeed app +``` + +This tool is perfect for periodic "note cleanup" sessions where you consolidate scattered tasks into your main GTD system. \ No newline at end of file diff --git a/demo_cancelled_workflow.py b/demo_cancelled_workflow.py new file mode 100644 index 0000000..013addf --- /dev/null +++ b/demo_cancelled_workflow.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +""" +Demo showing cancelled task workflow with Godspeed sync. +""" + +import tempfile +from pathlib import Path + + +def demo_cancelled_workflow(): + print("=== Godspeed Cancelled Task Workflow Demo ===\n") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + print("📝 Scenario: Managing a project with tasks that get cancelled") + print("=" * 65) + + # Initial tasks + print("\n1. Initial project tasks in markdown:") + initial_tasks = [ + ("task1", "incomplete", "Design new feature", ""), + ("task2", "incomplete", "Get approval from stakeholders", ""), + ("task3", "incomplete", "Implement feature", ""), + ("task4", "incomplete", "Write documentation", ""), + ("task5", "incomplete", "Deploy to production", ""), + ] + + project_file = sync_dir / "New_Feature_Project.md" + sync_engine._write_list_file(project_file, initial_tasks) + + with open(project_file, "r") as f: + print(f.read()) + + print("2. Project update - some tasks completed, one cancelled:") + print("-" * 58) + + # Simulate project evolution + updated_content = """- [x] Design new feature +- [-] Get approval from stakeholders + Stakeholders decided to cancel this feature +- [-] Implement feature + No longer needed since feature was cancelled +- [-] Write documentation + Documentation not needed for cancelled feature +- [-] Deploy to production + Cannot deploy cancelled feature +- [ ] Archive project files + New cleanup task +""" + + with open(project_file, "w") as f: + f.write(updated_content) + + print(updated_content) + + # Parse the changes + updated_tasks = sync_engine._read_list_file(project_file) + + print("3. What would sync to Godspeed API:") + print("-" * 36) + + api_calls = [] + for local_id, status, title, notes in updated_tasks: + if status == "complete": + api_calls.append( + f"PATCH /tasks/{local_id} {{'is_complete': True, 'is_cleared': False}}" + ) + print(f" ✅ COMPLETE: {title}") + elif status == "cleared": + api_calls.append( + f"PATCH /tasks/{local_id} {{'is_complete': True, 'is_cleared': True}}" + ) + print(f" ❌ CANCEL: {title}") + if notes: + print(f" Reason: {notes}") + elif local_id == "task6": # New task + api_calls.append( + f"POST /tasks {{'title': '{title}', 'list_id': 'project-list'}}" + ) + print(f" ➕ NEW: {title}") + else: + print(f" ⏳ INCOMPLETE: {title}") + + print(f"\n4. API calls that would be made ({len(api_calls)} total):") + print("-" * 49) + for call in api_calls: + print(f" {call}") + + print("\n5. Next sync download behavior:") + print("-" * 32) + print(" When downloading from Godspeed API:") + print(" • Only incomplete tasks appear in local files") + print(" • Completed and cancelled tasks are hidden") + print(" • This keeps your local markdown files clean") + print(f" • Current file would only show: 'Archive project files'") + + print("\n✨ Benefits of this workflow:") + print(" • Clear visual distinction: [-] for cancelled vs [x] for completed") + print(" • Cancelled tasks sync to Godspeed's 'cleared' status") + print(" • Completed/cancelled tasks auto-hide on next download") + print(" • Notes explain why tasks were cancelled") + print(" • Clean local files focused on active work") + + +if __name__ == "__main__": + demo_cancelled_workflow() diff --git a/demo_completion_sync.py b/demo_completion_sync.py new file mode 100644 index 0000000..ec84040 --- /dev/null +++ b/demo_completion_sync.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +Demo script showing how Godspeed completion status sync works. +This creates sample markdown files and shows the sync behavior. +""" + +import tempfile +from pathlib import Path + + +def demo_completion_sync(): + print("=== Godspeed Completion Status Sync Demo ===\n") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + print("1. Creating sample markdown file with mixed completion states:") + print("-" * 60) + + # Create sample tasks + sample_tasks = [ + ("task001", False, "Buy groceries", "Don't forget milk"), + ("task002", True, "Call dentist", ""), + ("task003", False, "Finish project", "Due next Friday"), + ("task004", True, "Exercise today", "Went for a 30min run"), + ] + + # Write to markdown file + demo_file = sync_dir / "Personal.md" + sync_engine._write_list_file(demo_file, sample_tasks) + + # Show the generated markdown + with open(demo_file, "r") as f: + content = f.read() + + print(content) + print("-" * 60) + + print("\n2. What this represents in Godspeed:") + for task_id, is_complete, title, notes in sample_tasks: + status = "✅ COMPLETED" if is_complete else "⏳ INCOMPLETE" + print(f" {status}: {title}") + if notes: + print(f" Notes: {notes}") + + print("\n3. Now let's modify the markdown file (simulate user editing):") + print("-" * 60) + + # Simulate user changes - flip some completion states + modified_content = content.replace( + "- [ ] Buy groceries", + "- [x] Buy groceries", # Mark as complete + ).replace( + "- [x] Call dentist", + "- [ ] Call dentist", # Mark as incomplete + ) + + # Add a new task + modified_content += "- [ ] New task from markdown \n" + + print(modified_content) + print("-" * 60) + + # Write the modified content + with open(demo_file, "w") as f: + f.write(modified_content) + + # Parse the changes + updated_tasks = sync_engine._read_list_file(demo_file) + + print("\n4. Changes that would sync to Godspeed:") + print("-" * 40) + + for i, (task_id, is_complete, title, notes) in enumerate(updated_tasks): + if i < len(sample_tasks): + old_complete = sample_tasks[i][1] + if old_complete != is_complete: + action = "MARK COMPLETE" if is_complete else "MARK INCOMPLETE" + print(f" 🔄 {action}: {title}") + else: + status = "✅" if is_complete else "⏳" + print(f" {status} No change: {title}") + else: + print(f" ➕ CREATE NEW: {title}") + + print("\n5. API calls that would be made:") + print("-" * 35) + print(" PATCH /tasks/task001 {'is_complete': True}") + print(" PATCH /tasks/task002 {'is_complete': False}") + print(" POST /tasks {'title': 'New task from markdown'}") + + print("\n✨ Summary:") + print(" • Checking [x] or [X] in markdown marks task complete in Godspeed") + print(" • Unchecking [ ] in markdown marks task incomplete in Godspeed") + print(" • Adding new tasks in markdown creates them in Godspeed") + print(" • Changes sync both directions during 'godspeed sync'") + + +if __name__ == "__main__": + demo_completion_sync() diff --git a/pyproject.toml b/pyproject.toml index f29207e..addc2c4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ requires-python = ">=3.12" dependencies = [ "aiohttp>=3.11.18", "certifi>=2025.4.26", + "click>=8.1.0", "html2text>=2025.4.15", "mammoth>=1.9.0", "markitdown[all]>=0.1.1", @@ -16,6 +17,7 @@ dependencies = [ "pillow>=11.2.1", "python-dateutil>=2.9.0.post0", "python-docx>=1.1.2", + "requests>=2.31.0", "rich>=14.0.0", "textual>=3.2.0", "textual-image>=0.8.2", diff --git a/src/cli/__init__.py b/src/cli/__init__.py index 4142078..475afc1 100644 --- a/src/cli/__init__.py +++ b/src/cli/__init__.py @@ -7,6 +7,7 @@ from .drive import drive from .email import email from .calendar import calendar from .ticktick import ticktick +from .godspeed import godspeed @click.group() @@ -20,6 +21,9 @@ cli.add_command(drive) cli.add_command(email) cli.add_command(calendar) cli.add_command(ticktick) +cli.add_command(godspeed) # Add 'tt' as a short alias for ticktick cli.add_command(ticktick, name="tt") +# Add 'gs' as a short alias for godspeed +cli.add_command(godspeed, name="gs") diff --git a/src/cli/godspeed.py b/src/cli/godspeed.py new file mode 100644 index 0000000..0a960b8 --- /dev/null +++ b/src/cli/godspeed.py @@ -0,0 +1,616 @@ +"""CLI interface for Godspeed sync functionality.""" + +import click +import getpass +import os +import sys +from pathlib import Path +from datetime import datetime + +from ..services.godspeed.client import GodspeedClient +from ..services.godspeed.sync import GodspeedSync + + +def get_credentials(): + """Get Godspeed credentials from environment or user input.""" + email = os.getenv("GODSPEED_EMAIL") + password = os.getenv("GODSPEED_PASSWORD") + token = os.getenv("GODSPEED_TOKEN") + + if token: + return None, None, token + + if not email: + email = click.prompt("Godspeed email") + + if not password: + password = click.prompt("Godspeed password", hide_input=True) + + return email, password, None + + +def get_sync_directory(): + """Get sync directory from environment or default.""" + sync_dir = os.getenv("GODSPEED_SYNC_DIR") + if sync_dir: + return Path(sync_dir) + + # Default to ~/Documents/Godspeed or ~/.local/share/gtd-terminal-tools/godspeed + home = Path.home() + + # Try Documents first + docs_dir = home / "Documents" / "Godspeed" + if docs_dir.parent.exists(): + return docs_dir + + # Fall back to data directory + data_dir = home / ".local" / "share" / "gtd-terminal-tools" / "godspeed" + return data_dir + + +@click.group() +def godspeed(): + """Godspeed sync tool - bidirectional sync between Godspeed API and markdown files.""" + pass + + +@godspeed.command() +def download(): + """Download tasks from Godspeed API to local files.""" + email, password, token = get_credentials() + sync_dir = get_sync_directory() + + try: + client = GodspeedClient(email=email, password=password, token=token) + sync_engine = GodspeedSync(client, sync_dir) + sync_engine.download_from_api() + + click.echo(f"\nTasks downloaded to: {sync_dir}") + click.echo( + "You can now edit the markdown files and run 'godspeed upload' to sync changes back." + ) + + except Exception as e: + click.echo(f"Error during download: {e}", err=True) + sys.exit(1) + + +@godspeed.command() +def upload(): + """Upload local markdown files to Godspeed API.""" + email, password, token = get_credentials() + sync_dir = get_sync_directory() + + if not sync_dir.exists(): + click.echo(f"Sync directory does not exist: {sync_dir}", err=True) + click.echo("Run 'godspeed download' first to initialize the sync directory.") + sys.exit(1) + + try: + client = GodspeedClient(email=email, password=password, token=token) + sync_engine = GodspeedSync(client, sync_dir) + sync_engine.upload_to_api() + + click.echo("Local changes uploaded successfully.") + + except Exception as e: + click.echo(f"Error during upload: {e}", err=True) + sys.exit(1) + + +@godspeed.command() +def sync(): + """Perform bidirectional sync between local files and Godspeed API.""" + email, password, token = get_credentials() + sync_dir = get_sync_directory() + + try: + client = GodspeedClient(email=email, password=password, token=token) + sync_engine = GodspeedSync(client, sync_dir) + sync_engine.sync_bidirectional() + + click.echo(f"\nSync complete. Files are in: {sync_dir}") + + except Exception as e: + click.echo(f"Error during sync: {e}", err=True) + sys.exit(1) + + +@godspeed.command() +def status(): + """Show sync status and directory information.""" + sync_dir = get_sync_directory() + + if not sync_dir.exists(): + click.echo(f"Sync directory does not exist: {sync_dir}") + click.echo("Run 'godspeed download' or 'godspeed sync' to initialize.") + return + + # Create a minimal sync engine for status (no API client needed) + sync_engine = GodspeedSync(None, sync_dir) + status_info = sync_engine.get_sync_status() + + click.echo(f"Sync Directory: {status_info['sync_directory']}") + click.echo(f"Local Files: {status_info['local_files']}") + click.echo(f"Total Local Tasks: {status_info['total_local_tasks']}") + click.echo(f"Tracked Tasks: {status_info['tracked_tasks']}") + click.echo(f"Tracked Lists: {status_info['tracked_lists']}") + + if status_info["last_sync"]: + click.echo(f"Last Sync: {status_info['last_sync']}") + else: + click.echo("Last Sync: Never") + + click.echo("\nMarkdown Files:") + for file_path in sync_engine.list_local_files(): + tasks = sync_engine._read_list_file(file_path) + completed = sum( + 1 for _, status, _, _ in tasks if status in ["complete", "cleared"] + ) + total = len(tasks) + click.echo(f" {file_path.name}: {completed}/{total} completed") + + +@godspeed.command() +def test_connection(): + """Test connection to Godspeed API with SSL diagnostics.""" + import requests + import ssl + import socket + + click.echo("Testing connection to Godspeed API...") + + # Check if SSL bypass is enabled first + disable_ssl = os.getenv("GODSPEED_DISABLE_SSL_VERIFY", "").lower() == "true" + if disable_ssl: + click.echo("⚠️ SSL verification is disabled (GODSPEED_DISABLE_SSL_VERIFY=true)") + + # Test basic connectivity + ssl_error_occurred = False + try: + response = requests.get("https://api.godspeedapp.com", timeout=10) + click.echo("✓ Basic HTTPS connection successful") + except requests.exceptions.SSLError as e: + ssl_error_occurred = True + click.echo(f"✗ SSL Error: {e}") + if not disable_ssl: + click.echo("\n💡 Try setting: export GODSPEED_DISABLE_SSL_VERIFY=true") + except requests.exceptions.ConnectionError as e: + click.echo(f"✗ Connection Error: {e}") + return + except Exception as e: + click.echo(f"✗ Unexpected Error: {e}") + return + + # Test with SSL bypass if enabled and there was an SSL error + if disable_ssl and ssl_error_occurred: + try: + response = requests.get( + "https://api.godspeedapp.com", verify=False, timeout=10 + ) + click.echo("✓ Connection successful with SSL bypass") + except Exception as e: + click.echo(f"✗ Connection failed even with SSL bypass: {e}") + return + + # Test authentication if credentials available + email, password, token = get_credentials() + if token or (email and password): + try: + client = GodspeedClient(email=email, password=password, token=token) + lists = client.get_lists() + click.echo(f"✓ Authentication successful, found {len(lists)} lists") + except Exception as e: + click.echo(f"✗ Authentication failed: {e}") + else: + click.echo("ℹ️ No credentials provided for authentication test") + + click.echo("\nConnection test complete!") + + +@godspeed.command() +def open(): + """Open the sync directory in the default file manager.""" + sync_dir = get_sync_directory() + + if not sync_dir.exists(): + click.echo(f"Sync directory does not exist: {sync_dir}", err=True) + click.echo("Run 'godspeed download' or 'godspeed sync' to initialize.") + return + + import subprocess + import platform + + system = platform.system() + try: + if system == "Darwin": # macOS + subprocess.run(["open", str(sync_dir)]) + elif system == "Windows": + subprocess.run(["explorer", str(sync_dir)]) + else: # Linux + subprocess.run(["xdg-open", str(sync_dir)]) + + click.echo(f"Opened sync directory: {sync_dir}") + except Exception as e: + click.echo(f"Could not open directory: {e}", err=True) + click.echo(f"Sync directory is: {sync_dir}") + + +class TaskSweeper: + """Sweeps incomplete tasks from markdown files into Godspeed Inbox.""" + + def __init__(self, notes_dir: Path, godspeed_dir: Path, dry_run: bool = False): + self.notes_dir = Path(notes_dir) + self.godspeed_dir = Path(godspeed_dir) + self.dry_run = dry_run + self.inbox_file = self.godspeed_dir / "Inbox.md" + + # Try to use the sync engine for consistent ID generation and formatting + try: + self.sync_engine = GodspeedSync(None, str(godspeed_dir)) + except Exception: + # Fallback parsing if sync engine fails + self.sync_engine = None + + def _parse_task_line_fallback(self, line: str): + """Fallback task parsing if sync engine not available.""" + import re + import uuid + + # Match patterns like: - [ ] Task title + task_pattern = ( + r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*)?\s*$" + ) + match = re.match(task_pattern, line.strip()) + + if not match: + return None + + checkbox, title_and_notes, local_id = match.groups() + + # Determine status + if checkbox.lower() == "x": + status = "complete" + elif checkbox == "-": + status = "cleared" + else: + status = "incomplete" + + # Extract title (remove any inline notes after " + if notes: + formatted += f"\n {notes}" + + new_task_lines.append(formatted) + + # Combine with existing content + if existing_content.strip(): + new_content = ( + existing_content.rstrip() + "\n\n" + "\n".join(new_task_lines) + "\n" + ) + else: + new_content = "\n".join(new_task_lines) + "\n" + + with builtins.open(str(file_path), "w", encoding="utf-8") as f: + f.write(new_content) + + def _clean_file(self, file_path: Path, non_task_lines): + """Remove tasks from original file, keeping only non-task content.""" + import builtins + + if not non_task_lines or all(not line.strip() for line in non_task_lines): + # File would be empty, delete it + if not self.dry_run: + file_path.unlink() + click.echo(f" 🗑️ Would delete empty file: {file_path}") + else: + # Write back non-task content + cleaned_content = "\n".join(non_task_lines).strip() + if cleaned_content: + cleaned_content += "\n" + + if not self.dry_run: + with builtins.open(str(file_path), "w", encoding="utf-8") as f: + f.write(cleaned_content) + click.echo(f" ✂️ Cleaned file (removed tasks): {file_path}") + + def find_markdown_files(self): + """Find all markdown files in the notes directory, excluding Godspeed directory.""" + markdown_files = [] + + for md_file in self.notes_dir.rglob("*.md"): + # Skip files in the Godspeed directory + if ( + self.godspeed_dir in md_file.parents + or md_file.parent == self.godspeed_dir + ): + continue + + # Skip hidden files and directories + if any(part.startswith(".") for part in md_file.parts): + continue + + markdown_files.append(md_file) + + return sorted(markdown_files) + + def sweep_tasks(self): + """Sweep incomplete tasks from all markdown files into Inbox.""" + click.echo(f"🧹 Sweeping incomplete tasks from: {self.notes_dir}") + click.echo(f"📥 Target Inbox: {self.inbox_file}") + click.echo(f"🔍 Dry run: {self.dry_run}") + click.echo("=" * 60) + + markdown_files = self.find_markdown_files() + click.echo(f"\n📁 Found {len(markdown_files)} markdown files to process") + + swept_tasks = [] + processed_files = [] + + for file_path in markdown_files: + try: + rel_path = file_path.relative_to(self.notes_dir) + rel_path_str = str(rel_path) + except Exception as e: + click.echo(f"Error getting relative path for {file_path}: {e}") + rel_path_str = str(file_path.name) + + click.echo(f"\n📄 Processing: {rel_path_str}") + + tasks, non_task_lines = self._parse_markdown_file(file_path) + + if not tasks: + click.echo(f" ℹ️ No tasks found") + continue + if not tasks: + click.echo(f" ℹ️ No tasks found") + continue + + # Separate incomplete tasks from completed/cleared ones + incomplete_tasks = [] + complete_tasks = [] + + for task in tasks: + local_id, status, title, notes = task + if status == "incomplete": + incomplete_tasks.append(task) + else: + complete_tasks.append(task) + + if incomplete_tasks: + click.echo(f" 🔄 Found {len(incomplete_tasks)} incomplete tasks:") + for _, status, title, notes in incomplete_tasks: + click.echo(f" • {title}") + if notes: + click.echo(f" Notes: {notes}") + + # Add source file annotation with clean task IDs + annotated_tasks = [] + for local_id, status, title, notes in incomplete_tasks: + # Generate a fresh ID for swept tasks to avoid conflicts + if self.sync_engine: + fresh_id = self.sync_engine._generate_local_id() + else: + import uuid + + fresh_id = str(uuid.uuid4())[:8] + + # Add source info to notes + source_notes = f"From: {rel_path_str}" + if notes: + combined_notes = f"{notes}\n{source_notes}" + else: + combined_notes = source_notes + annotated_tasks.append((fresh_id, status, title, combined_notes)) + + swept_tasks.extend(annotated_tasks) + processed_files.append(str(rel_path)) + + if complete_tasks: + click.echo( + f" ✅ Keeping {len(complete_tasks)} completed/cleared tasks in place" + ) + + # Reconstruct remaining content (non-tasks + completed tasks) + remaining_content = non_task_lines.copy() + + # Add completed/cleared tasks back to remaining content + if complete_tasks: + remaining_content.append("") # Empty line before tasks + for task in complete_tasks: + if self.sync_engine: + formatted = self.sync_engine._format_task_line(*task) + else: + local_id, status, title, notes = task + checkbox = { + "incomplete": "[ ]", + "complete": "[x]", + "cleared": "[-]", + }[status] + formatted = f"- {checkbox} {title} " + if notes: + formatted += f"\n {notes}" + remaining_content.append(formatted) + + # Clean the original file + if incomplete_tasks: + self._clean_file(file_path, remaining_content) + + # Write swept tasks to Inbox + if swept_tasks: + click.echo(f"\n📥 Writing {len(swept_tasks)} tasks to Inbox...") + if not self.dry_run: + self._write_tasks_to_file(self.inbox_file, swept_tasks) + click.echo(f" ✅ Inbox updated: {self.inbox_file}") + + # Summary + click.echo(f"\n" + "=" * 60) + click.echo(f"📊 SWEEP SUMMARY:") + click.echo(f" • Files processed: {len(processed_files)}") + click.echo(f" • Tasks swept: {len(swept_tasks)}") + click.echo(f" • Target: {self.inbox_file}") + + if self.dry_run: + click.echo(f"\n⚠️ DRY RUN - No files were actually modified") + click.echo(f" Run without --dry-run to perform the sweep") + + return { + "swept_tasks": len(swept_tasks), + "processed_files": processed_files, + "inbox_file": str(self.inbox_file), + } + + +@godspeed.command() +@click.argument( + "notes_dir", + type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path), + required=False, +) +@click.argument( + "godspeed_dir", + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + required=False, +) +@click.option( + "--dry-run", is_flag=True, help="Show what would be done without making changes" +) +def sweep(notes_dir, godspeed_dir, dry_run): + """Sweep incomplete tasks from markdown files into Godspeed Inbox. + + NOTES_DIR: Directory containing markdown files with tasks to sweep (optional, defaults to $NOTES_DIR) + GODSPEED_DIR: Godspeed sync directory (optional, defaults to sync directory) + """ + # Handle notes_dir default from environment + if notes_dir is None: + notes_dir_env = os.getenv("NOTES_DIR") + if not notes_dir_env: + click.echo( + "❌ No notes directory specified and $NOTES_DIR environment variable not set", + err=True, + ) + click.echo("Usage: godspeed sweep [godspeed_dir]", err=True) + click.echo( + " or: export NOTES_DIR=/path/to/notes && godspeed sweep", err=True + ) + sys.exit(1) + notes_dir = Path(notes_dir_env) + if not notes_dir.exists(): + click.echo( + f"❌ Notes directory from $NOTES_DIR does not exist: {notes_dir}", + err=True, + ) + sys.exit(1) + if not notes_dir.is_dir(): + click.echo( + f"❌ Notes path from $NOTES_DIR is not a directory: {notes_dir}", + err=True, + ) + sys.exit(1) + + if godspeed_dir is None: + godspeed_dir = get_sync_directory() + + # Ensure we have Path objects + notes_dir = Path(notes_dir) + godspeed_dir = Path(godspeed_dir) + + try: + sweeper = TaskSweeper(notes_dir, godspeed_dir, dry_run) + result = sweeper.sweep_tasks() + + if result["swept_tasks"] > 0: + click.echo(f"\n🎉 Successfully swept {result['swept_tasks']} tasks!") + if not dry_run: + click.echo(f"💡 Next steps:") + click.echo(f" 1. Review tasks in: {result['inbox_file']}") + click.echo(f" 2. Run 'godspeed upload' to sync to API") + click.echo( + f" 3. Organize tasks into appropriate lists in Godspeed app" + ) + else: + click.echo(f"\n✨ No incomplete tasks found to sweep.") + + except Exception as e: + click.echo(f"❌ Error during sweep: {e}", err=True) + import traceback + + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + godspeed() diff --git a/src/cli/sync.py b/src/cli/sync.py index 714d981..ee8cc94 100644 --- a/src/cli/sync.py +++ b/src/cli/sync.py @@ -1,8 +1,11 @@ import click import asyncio import os -from rich.progress import Progress, SpinnerColumn, MofNCompleteColumn +import json +import time from datetime import datetime, timedelta +from pathlib import Path +from rich.progress import Progress, SpinnerColumn, MofNCompleteColumn from src.utils.mail_utils.helpers import ensure_directory_exists from src.utils.calendar_utils import save_events_to_vdir, save_events_to_file @@ -21,6 +24,180 @@ from src.services.microsoft_graph.mail import ( process_outbox_async, ) from src.services.microsoft_graph.auth import get_access_token +from src.services.godspeed.client import GodspeedClient +from src.services.godspeed.sync import GodspeedSync + + +# Timing state management +def get_sync_state_file(): + """Get the path to the sync state file.""" + return os.path.expanduser("~/.local/share/gtd-terminal-tools/sync_state.json") + + +def load_sync_state(): + """Load the sync state from file.""" + state_file = get_sync_state_file() + if os.path.exists(state_file): + try: + with open(state_file, "r") as f: + return json.load(f) + except Exception: + pass + return { + "last_godspeed_sync": 0, + "last_sweep_date": None, + "sweep_completed_today": False, + } + + +def save_sync_state(state): + """Save the sync state to file.""" + state_file = get_sync_state_file() + os.makedirs(os.path.dirname(state_file), exist_ok=True) + with open(state_file, "w") as f: + json.dump(state, f, indent=2) + + +def should_run_godspeed_sync(): + """Check if Godspeed sync should run (every 15 minutes).""" + state = load_sync_state() + current_time = time.time() + last_sync = state.get("last_godspeed_sync", 0) + return current_time - last_sync >= 900 # 15 minutes in seconds + + +def should_run_sweep(): + """Check if sweep should run (once after 6pm each day).""" + state = load_sync_state() + current_time = datetime.now() + + # Check if it's after 6 PM + if current_time.hour < 18: + return False + + # Check if we've already swept today + today_str = current_time.strftime("%Y-%m-%d") + last_sweep_date = state.get("last_sweep_date") + + return last_sweep_date != today_str + + +def get_godspeed_sync_directory(): + """Get Godspeed sync directory from environment or default.""" + sync_dir = os.getenv("GODSPEED_SYNC_DIR") + if sync_dir: + return Path(sync_dir) + + # Default to ~/Documents/Godspeed or ~/.local/share/gtd-terminal-tools/godspeed + home = Path.home() + + # Try Documents first + docs_dir = home / "Documents" / "Godspeed" + if docs_dir.parent.exists(): + return docs_dir + + # Fall back to data directory + data_dir = home / ".local" / "share" / "gtd-terminal-tools" / "godspeed" + return data_dir + + +def get_godspeed_credentials(): + """Get Godspeed credentials from environment.""" + email = os.getenv("GODSPEED_EMAIL") + password = os.getenv("GODSPEED_PASSWORD") + token = os.getenv("GODSPEED_TOKEN") + return email, password, token + + +async def run_godspeed_sync(progress=None): + """Run Godspeed bidirectional sync.""" + try: + email, password, token = get_godspeed_credentials() + if not (token or (email and password)): + if progress: + progress.console.print( + "[yellow]⚠️ Skipping Godspeed sync: No credentials configured[/yellow]" + ) + return False + + sync_dir = get_godspeed_sync_directory() + + if progress: + progress.console.print( + f"[cyan]🔄 Running Godspeed sync to {sync_dir}...[/cyan]" + ) + + client = GodspeedClient(email=email, password=password, token=token) + sync_engine = GodspeedSync(client, sync_dir) + sync_engine.sync_bidirectional() + + # Update sync state + state = load_sync_state() + state["last_godspeed_sync"] = time.time() + save_sync_state(state) + + if progress: + progress.console.print("[green]✅ Godspeed sync completed[/green]") + return True + + except Exception as e: + if progress: + progress.console.print(f"[red]❌ Godspeed sync failed: {e}[/red]") + return False + + +async def run_task_sweep(progress=None): + """Run task sweep from notes directory to Godspeed inbox.""" + try: + from src.cli.godspeed import TaskSweeper + + notes_dir_env = os.getenv("NOTES_DIR") + if not notes_dir_env: + if progress: + progress.console.print( + "[yellow]⚠️ Skipping task sweep: $NOTES_DIR not configured[/yellow]" + ) + return False + + notes_dir = Path(notes_dir_env) + if not notes_dir.exists(): + if progress: + progress.console.print( + f"[yellow]⚠️ Skipping task sweep: Notes directory does not exist: {notes_dir}[/yellow]" + ) + return False + + godspeed_dir = get_godspeed_sync_directory() + + if progress: + progress.console.print( + f"[cyan]🧹 Running task sweep from {notes_dir} to {godspeed_dir}...[/cyan]" + ) + + sweeper = TaskSweeper(notes_dir, godspeed_dir, dry_run=False) + result = sweeper.sweep_tasks() + + # Update sweep state + state = load_sync_state() + state["last_sweep_date"] = datetime.now().strftime("%Y-%m-%d") + save_sync_state(state) + + if result["swept_tasks"] > 0: + if progress: + progress.console.print( + f"[green]✅ Task sweep completed: {result['swept_tasks']} tasks swept[/green]" + ) + else: + if progress: + progress.console.print( + "[green]✅ Task sweep completed: No tasks to sweep[/green]" + ) + return True + + except Exception as e: + if progress: + progress.console.print(f"[red]❌ Task sweep failed: {e}[/red]") + return False # Function to create Maildir structure @@ -362,6 +539,36 @@ async def _sync_outlook_data( notify_new_emails(new_message_count, org) progress.console.print("[bold green]Step 2: New data fetched.[/bold green]") + + # Stage 3: Run Godspeed operations based on timing + progress.console.print( + "\n[bold cyan]Step 3: Running Godspeed operations...[/bold cyan]" + ) + + # Check if Godspeed sync should run (every 15 minutes) + if should_run_godspeed_sync(): + await run_godspeed_sync(progress) + else: + progress.console.print("[dim]⏭️ Skipping Godspeed sync (not due yet)[/dim]") + + # Check if task sweep should run (once after 6pm daily) + if should_run_sweep(): + await run_task_sweep(progress) + else: + current_hour = datetime.now().hour + if current_hour < 18: + progress.console.print( + "[dim]⏭️ Skipping task sweep (before 6 PM)[/dim]" + ) + else: + progress.console.print( + "[dim]⏭️ Skipping task sweep (already completed today)[/dim]" + ) + + progress.console.print( + "[bold green]Step 3: Godspeed operations completed.[/bold green]" + ) + click.echo("Sync complete.") @@ -656,59 +863,43 @@ async def daemon_mode( pending_email_count = len(pending_emails) outbox_changes = pending_email_count > 0 + # Check Godspeed operations + godspeed_sync_due = should_run_godspeed_sync() + sweep_due = should_run_sweep() + # Determine what changed and show appropriate status - if mail_changes and calendar_changes and outbox_changes: - console.print( - create_status_display( - f"Changes detected! Mail: Remote {remote_message_count}, Local {local_message_count} | Calendar: {calendar_change_desc} | Outbox: {pending_email_count} pending. Starting sync...", - "yellow", + changes_detected = ( + mail_changes + or calendar_changes + or outbox_changes + or godspeed_sync_due + or sweep_due + ) + + if changes_detected: + change_parts = [] + if mail_changes: + change_parts.append( + f"Mail: Remote {remote_message_count}, Local {local_message_count}" ) - ) - elif mail_changes and calendar_changes: + if calendar_changes: + change_parts.append(f"Calendar: {calendar_change_desc}") + if outbox_changes: + change_parts.append(f"Outbox: {pending_email_count} pending") + if godspeed_sync_due: + change_parts.append("Godspeed sync due") + if sweep_due: + change_parts.append("Task sweep due") + console.print( create_status_display( - f"Changes detected! Mail: Remote {remote_message_count}, Local {local_message_count} | Calendar: {calendar_change_desc}. Starting sync...", - "yellow", - ) - ) - elif mail_changes and outbox_changes: - console.print( - create_status_display( - f"Changes detected! Mail: Remote {remote_message_count}, Local {local_message_count} | Outbox: {pending_email_count} pending. Starting sync...", - "yellow", - ) - ) - elif calendar_changes and outbox_changes: - console.print( - create_status_display( - f"Changes detected! Calendar: {calendar_change_desc} | Outbox: {pending_email_count} pending. Starting sync...", - "yellow", - ) - ) - elif mail_changes: - console.print( - create_status_display( - f"New messages detected! Remote: {remote_message_count}, Local: {local_message_count}. Starting sync...", - "yellow", - ) - ) - elif calendar_changes: - console.print( - create_status_display( - f"Calendar changes detected! {calendar_change_desc}. Starting sync...", - "yellow", - ) - ) - elif outbox_changes: - console.print( - create_status_display( - f"Outbound emails detected! {pending_email_count} emails pending. Starting sync...", + f"Changes detected! {' | '.join(change_parts)}. Starting sync...", "yellow", ) ) # Sync if any changes detected - if mail_changes or calendar_changes or outbox_changes: + if changes_detected: await _sync_outlook_data( dry_run, vdir, @@ -732,6 +923,23 @@ async def daemon_mode( status_parts.append(f"Outbox: {pending_email_count} pending") + # Add Godspeed status + state = load_sync_state() + last_godspeed = state.get("last_godspeed_sync", 0) + minutes_since_godspeed = int((time.time() - last_godspeed) / 60) + status_parts.append(f"Godspeed: {minutes_since_godspeed}m ago") + + last_sweep = state.get("last_sweep_date") + if last_sweep == datetime.now().strftime("%Y-%m-%d"): + status_parts.append("Sweep: done today") + else: + current_hour = datetime.now().hour + if current_hour >= 18: + status_parts.append("Sweep: due") + else: + hours_until_sweep = 18 - current_hour + status_parts.append(f"Sweep: in {hours_until_sweep}h") + console.print( create_status_display( f"No changes detected ({', '.join(status_parts)})", diff --git a/src/services/godspeed/__init__.py b/src/services/godspeed/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/services/godspeed/client.py b/src/services/godspeed/client.py new file mode 100644 index 0000000..ac906ab --- /dev/null +++ b/src/services/godspeed/client.py @@ -0,0 +1,129 @@ +"""Godspeed API client for task and list management.""" + +import json +import os +import re +import requests +from pathlib import Path +from typing import Dict, List, Optional, Any +from datetime import datetime +import urllib3 + + +class GodspeedClient: + """Client for interacting with the Godspeed API.""" + + BASE_URL = "https://api.godspeedapp.com" + + def __init__(self, email: str = None, password: str = None, token: str = None): + self.email = email + self.password = password + self.token = token + self.session = requests.Session() + + # Handle SSL verification bypass for corporate networks + disable_ssl = os.getenv("GODSPEED_DISABLE_SSL_VERIFY", "").lower() == "true" + if disable_ssl: + self.session.verify = False + # Suppress only the specific warning about unverified HTTPS requests + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + print("⚠️ SSL verification disabled for Godspeed API") + + if token: + self.session.headers.update({"Authorization": f"Bearer {token}"}) + elif email and password: + self._authenticate() + + def _authenticate(self) -> str: + """Authenticate and get access token.""" + if not self.email or not self.password: + raise ValueError("Email and password required for authentication") + + response = self.session.post( + f"{self.BASE_URL}/sessions/sign_in", + json={"email": self.email, "password": self.password}, + headers={"Content-Type": "application/json"}, + ) + response.raise_for_status() + + data = response.json() + if not data.get("success"): + raise Exception("Authentication failed") + + self.token = data["token"] + self.session.headers.update({"Authorization": f"Bearer {self.token}"}) + return self.token + + def get_lists(self) -> List[Dict[str, Any]]: + """Get all lists.""" + response = self.session.get(f"{self.BASE_URL}/lists") + response.raise_for_status() + return response.json() + + def get_tasks(self, list_id: str = None, status: str = None) -> Dict[str, Any]: + """Get tasks with optional filtering.""" + params = {} + if list_id: + params["list_id"] = list_id + if status: + params["status"] = status + + response = self.session.get(f"{self.BASE_URL}/tasks", params=params) + response.raise_for_status() + return response.json() + + def get_task(self, task_id: str) -> Dict[str, Any]: + """Get a single task by ID.""" + response = self.session.get(f"{self.BASE_URL}/tasks/{task_id}") + response.raise_for_status() + return response.json() + + def create_task( + self, + title: str, + list_id: str = None, + notes: str = None, + location: str = "end", + **kwargs, + ) -> Dict[str, Any]: + """Create a new task.""" + data = {"title": title, "location": location} + + if list_id: + data["list_id"] = list_id + if notes: + data["notes"] = notes + + # Add any additional kwargs + data.update(kwargs) + + response = self.session.post( + f"{self.BASE_URL}/tasks", + json=data, + headers={"Content-Type": "application/json"}, + ) + response.raise_for_status() + return response.json() + + def update_task(self, task_id: str, **kwargs) -> Dict[str, Any]: + """Update an existing task.""" + response = self.session.patch( + f"{self.BASE_URL}/tasks/{task_id}", + json=kwargs, + headers={"Content-Type": "application/json"}, + ) + response.raise_for_status() + return response.json() + + def delete_task(self, task_id: str) -> None: + """Delete a task.""" + response = self.session.delete(f"{self.BASE_URL}/tasks/{task_id}") + response.raise_for_status() + + def complete_task(self, task_id: str) -> Dict[str, Any]: + """Mark a task as complete.""" + return self.update_task(task_id, is_complete=True) + + def incomplete_task(self, task_id: str) -> Dict[str, Any]: + """Mark a task as incomplete.""" + return self.update_task(task_id, is_complete=False) diff --git a/src/services/godspeed/config.py b/src/services/godspeed/config.py new file mode 100644 index 0000000..7512aef --- /dev/null +++ b/src/services/godspeed/config.py @@ -0,0 +1,87 @@ +"""Configuration and credential management for Godspeed sync.""" + +import json +import os +from pathlib import Path +from typing import Optional, Dict, Any + + +class GodspeedConfig: + """Manages configuration and credentials for Godspeed sync.""" + + def __init__(self, config_dir: Optional[Path] = None): + if config_dir is None: + config_dir = Path.home() / ".local" / "share" / "gtd-terminal-tools" + + self.config_dir = Path(config_dir) + self.config_file = self.config_dir / "godspeed_config.json" + self.config = self._load_config() + + def _load_config(self) -> Dict[str, Any]: + """Load configuration from file.""" + if self.config_file.exists(): + with open(self.config_file, "r") as f: + return json.load(f) + return {} + + def _save_config(self): + """Save configuration to file.""" + self.config_dir.mkdir(parents=True, exist_ok=True) + with open(self.config_file, "w") as f: + json.dump(self.config, f, indent=2) + + def get_email(self) -> Optional[str]: + """Get stored email or from environment.""" + return os.getenv("GODSPEED_EMAIL") or self.config.get("email") + + def set_email(self, email: str): + """Store email in config.""" + self.config["email"] = email + self._save_config() + + def get_token(self) -> Optional[str]: + """Get stored token or from environment.""" + return os.getenv("GODSPEED_TOKEN") or self.config.get("token") + + def set_token(self, token: str): + """Store token in config.""" + self.config["token"] = token + self._save_config() + + def get_sync_directory(self) -> Path: + """Get sync directory from config or environment.""" + sync_dir = os.getenv("GODSPEED_SYNC_DIR") or self.config.get("sync_directory") + + if sync_dir: + return Path(sync_dir) + + # Default to ~/Documents/Godspeed or ~/.local/share/gtd-terminal-tools/godspeed + home = Path.home() + + # Try Documents first + docs_dir = home / "Documents" / "Godspeed" + if docs_dir.parent.exists(): + return docs_dir + + # Fall back to data directory + return home / ".local" / "share" / "gtd-terminal-tools" / "godspeed" + + def set_sync_directory(self, sync_dir: Path): + """Store sync directory in config.""" + self.config["sync_directory"] = str(sync_dir) + self._save_config() + + def clear_credentials(self): + """Clear stored credentials.""" + self.config.pop("email", None) + self.config.pop("token", None) + self._save_config() + + def get_all_settings(self) -> Dict[str, Any]: + """Get all current settings.""" + return { + "email": self.get_email(), + "has_token": bool(self.get_token()), + "sync_directory": str(self.get_sync_directory()), + "config_file": str(self.config_file), + } diff --git a/src/services/godspeed/sync.py b/src/services/godspeed/sync.py new file mode 100644 index 0000000..1543be0 --- /dev/null +++ b/src/services/godspeed/sync.py @@ -0,0 +1,395 @@ +"""Two-way synchronization engine for Godspeed API and local markdown files.""" + +import json +import os +import re +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple +from datetime import datetime + +from .client import GodspeedClient + + +class GodspeedSync: + """Handles bidirectional sync between Godspeed API and local markdown files.""" + + def __init__(self, client: GodspeedClient, sync_dir: Path): + self.client = client + self.sync_dir = Path(sync_dir) + self.metadata_file = self.sync_dir / ".godspeed_metadata.json" + self.metadata = self._load_metadata() + + def _load_metadata(self) -> Dict: + """Load sync metadata from local file.""" + if self.metadata_file.exists(): + with open(self.metadata_file, "r") as f: + return json.load(f) + return { + "task_mapping": {}, # local_id -> godspeed_id + "list_mapping": {}, # list_name -> list_id + "last_sync": None, + } + + def _save_metadata(self): + """Save sync metadata to local file.""" + self.sync_dir.mkdir(parents=True, exist_ok=True) + with open(self.metadata_file, "w") as f: + json.dump(self.metadata, f, indent=2) + + def _sanitize_filename(self, name: str) -> str: + """Convert list name to safe filename.""" + # Replace special characters with underscores + sanitized = re.sub(r'[<>:"/\\|?*]', "_", name) + # Remove multiple underscores + sanitized = re.sub(r"_+", "_", sanitized) + # Strip leading/trailing underscores and spaces + return sanitized.strip("_ ") + + def _generate_local_id(self) -> str: + """Generate a unique local ID for tracking.""" + import uuid + + return str(uuid.uuid4())[:8] + + def _parse_task_line(self, line: str) -> Optional[Tuple[str, str, str, str]]: + """Parse a markdown task line and extract components. + + Returns: (local_id, status, title, notes) or None if invalid + status can be: 'incomplete', 'complete', or 'cleared' + """ + # Match patterns like: + # - [ ] Task title + # - [x] Completed task + # - [-] Cleared/cancelled task + # - [ ] Task with notes Some notes here + + task_pattern = r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*)?\s*(?:\n\s*(.+))?$" + match = re.match(task_pattern, line.strip(), re.MULTILINE | re.DOTALL) + + if not match: + return None + + checkbox, title_and_maybe_notes, local_id, extra_notes = match.groups() + + # Determine status from checkbox + if checkbox.lower() == "x": + status = "complete" + elif checkbox == "-": + status = "cleared" + else: + status = "incomplete" + + # Split title and inline notes if present + title_parts = title_and_maybe_notes.split("" + if notes: + line += f"\n {notes}" + return line + + def _read_list_file(self, list_path: Path) -> List[Tuple[str, str, str, str]]: + """Read and parse tasks from a markdown file.""" + if not list_path.exists(): + return [] + + tasks = [] + with open(list_path, "r", encoding="utf-8") as f: + content = f.read() + + # Split into potential task blocks + lines = content.split("\n") + current_task_lines = [] + + for line in lines: + if line.strip().startswith("- ["): + # Process previous task if exists + if current_task_lines: + task_block = "\n".join(current_task_lines) + parsed = self._parse_task_line(task_block) + if parsed: + tasks.append(parsed) + current_task_lines = [] + + current_task_lines = [line] + elif current_task_lines and line.strip(): + # Continuation of current task (notes) + current_task_lines.append(line) + elif current_task_lines: + # Empty line ends the current task + task_block = "\n".join(current_task_lines) + parsed = self._parse_task_line(task_block) + if parsed: + tasks.append(parsed) + current_task_lines = [] + + # Process last task if exists + if current_task_lines: + task_block = "\n".join(current_task_lines) + parsed = self._parse_task_line(task_block) + if parsed: + tasks.append(parsed) + + return tasks + + def _write_list_file(self, list_path: Path, tasks: List[Tuple[str, str, str, str]]): + """Write tasks to a markdown file.""" + list_path.parent.mkdir(parents=True, exist_ok=True) + + with open(list_path, "w", encoding="utf-8") as f: + for local_id, status, title, notes in tasks: + f.write(self._format_task_line(local_id, status, title, notes)) + f.write("\n") + + def download_from_api(self) -> None: + """Download all lists and tasks from Godspeed API to local files.""" + print("Downloading from Godspeed API...") + + # Get all lists + lists_data = self.client.get_lists() + lists = ( + lists_data if isinstance(lists_data, list) else lists_data.get("lists", []) + ) + + # Update list mapping + for list_item in lists: + list_name = list_item["name"] + list_id = list_item["id"] + self.metadata["list_mapping"][list_name] = list_id + + # Get only incomplete tasks (hide completed/cleared from local files) + all_tasks_data = self.client.get_tasks(status="incomplete") + tasks = all_tasks_data.get("tasks", []) + task_lists = all_tasks_data.get("lists", {}) + + # Group tasks by list + tasks_by_list = {} + for task in tasks: + list_id = task.get("list_id") + if list_id in task_lists: + list_name = task_lists[list_id]["name"] + else: + # Find list name from our mapping + list_name = None + for name, lid in self.metadata["list_mapping"].items(): + if lid == list_id: + list_name = name + break + if not list_name: + list_name = "Unknown" + + if list_name not in tasks_by_list: + tasks_by_list[list_name] = [] + tasks_by_list[list_name].append(task) + + # Create directory structure and files + for list_name, list_tasks in tasks_by_list.items(): + safe_name = self._sanitize_filename(list_name) + list_path = self.sync_dir / f"{safe_name}.md" + + # Convert API tasks to our format + local_tasks = [] + for task in list_tasks: + # Find existing local ID or create new one + godspeed_id = task["id"] + local_id = None + for lid, gid in self.metadata["task_mapping"].items(): + if gid == godspeed_id: + local_id = lid + break + + if not local_id: + local_id = self._generate_local_id() + self.metadata["task_mapping"][local_id] = godspeed_id + + # Convert API task status to our format + is_complete = task.get("is_complete", False) + is_cleared = task.get("is_cleared", False) + + if is_cleared: + status = "cleared" + elif is_complete: + status = "complete" + else: + status = "incomplete" + + title = task["title"] + notes = task.get("notes", "") + + local_tasks.append((local_id, status, title, notes)) + + self._write_list_file(list_path, local_tasks) + print(f" Downloaded {len(local_tasks)} tasks to {list_path}") + + self.metadata["last_sync"] = datetime.now().isoformat() + self._save_metadata() + print(f"Download complete. Synced {len(tasks_by_list)} lists.") + + def upload_to_api(self) -> None: + """Upload local markdown files to Godspeed API.""" + print("Uploading to Godspeed API...") + + # Find all markdown files + md_files = list(self.sync_dir.glob("*.md")) + + for md_file in md_files: + if md_file.name.startswith("."): + continue # Skip hidden files + + list_name = md_file.stem + local_tasks = self._read_list_file(md_file) + + # Get or create list ID + list_id = self.metadata["list_mapping"].get(list_name) + if not list_id: + print( + f" Warning: No list ID found for '{list_name}', tasks will go to Inbox" + ) + list_id = None + + for local_id, status, title, notes in local_tasks: + # Skip tasks with empty titles + if not title or not title.strip(): + print(f" Skipping task with empty title (id: {local_id})") + continue + + godspeed_id = self.metadata["task_mapping"].get(local_id) + + if godspeed_id: + # Update existing task + try: + update_data = {"title": title.strip()} + + # Handle status conversion to API format + if status == "complete": + update_data["is_complete"] = True + update_data["is_cleared"] = False + elif status == "cleared": + # Note: API requires task to be complete before clearing + update_data["is_complete"] = True + update_data["is_cleared"] = True + else: # incomplete + update_data["is_complete"] = False + update_data["is_cleared"] = False + + if notes and notes.strip(): + update_data["notes"] = notes.strip() + + self.client.update_task(godspeed_id, **update_data) + + action = { + "complete": "completed", + "cleared": "cleared", + "incomplete": "reopened", + }[status] + print(f" Updated task ({action}): {title}") + except Exception as e: + print(f" Error updating task '{title}': {e}") + else: + # Create new task + try: + create_data = { + "title": title.strip(), + "list_id": list_id, + } + + # Only add notes if they exist and are not empty + if notes and notes.strip(): + create_data["notes"] = notes.strip() + + print(f" Creating task: '{title}' with data: {create_data}") + response = self.client.create_task(**create_data) + print(f" API response: {response}") + + # Handle different response formats + if isinstance(response, dict): + if "id" in response: + new_godspeed_id = response["id"] + elif "task" in response and "id" in response["task"]: + new_godspeed_id = response["task"]["id"] + else: + print( + f" Warning: No ID found in response: {response}" + ) + continue + else: + print( + f" Warning: Unexpected response format: {response}" + ) + continue + + self.metadata["task_mapping"][local_id] = new_godspeed_id + + # Set status if not incomplete + if status == "complete": + self.client.update_task(new_godspeed_id, is_complete=True) + print(f" Created completed task: {title}") + elif status == "cleared": + # Mark complete first, then clear + self.client.update_task( + new_godspeed_id, is_complete=True, is_cleared=True + ) + print(f" Created cleared task: {title}") + else: + print(f" Created task: {title}") + except Exception as e: + print(f" Error creating task '{title}': {e}") + import traceback + + traceback.print_exc() + + self.metadata["last_sync"] = datetime.now().isoformat() + self._save_metadata() + print("Upload complete.") + + def sync_bidirectional(self) -> None: + """Perform a full bidirectional sync.""" + print("Starting bidirectional sync...") + + # Download first to get latest state + self.download_from_api() + + # Then upload any local changes + self.upload_to_api() + + print("Bidirectional sync complete.") + + def list_local_files(self) -> List[Path]: + """List all markdown files in sync directory.""" + if not self.sync_dir.exists(): + return [] + return list(self.sync_dir.glob("*.md")) + + def get_sync_status(self) -> Dict: + """Get current sync status and statistics.""" + local_files = self.list_local_files() + + total_local_tasks = 0 + for file_path in local_files: + tasks = self._read_list_file(file_path) + total_local_tasks += len(tasks) + + return { + "sync_directory": str(self.sync_dir), + "local_files": len(local_files), + "total_local_tasks": total_local_tasks, + "tracked_tasks": len(self.metadata["task_mapping"]), + "tracked_lists": len(self.metadata["list_mapping"]), + "last_sync": self.metadata.get("last_sync"), + } diff --git a/sweep_tasks.py b/sweep_tasks.py new file mode 100755 index 0000000..51e0e44 --- /dev/null +++ b/sweep_tasks.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python3 +""" +Godspeed Task Sweeper - Consolidate incomplete tasks from markdown files. + +This script recursively searches through directories (like 2024/, 2025/, etc.) +and moves all incomplete tasks from markdown files into the Godspeed Inbox.md file. +""" + +import argparse +import re +import shutil +from pathlib import Path +from typing import List, Tuple, Set +from datetime import datetime + + +class TaskSweeper: + """Sweeps incomplete tasks from markdown files into Godspeed Inbox.""" + + def __init__(self, notes_dir: Path, godspeed_dir: Path, dry_run: bool = False): + self.notes_dir = Path(notes_dir) + self.godspeed_dir = Path(godspeed_dir) + self.dry_run = dry_run + self.inbox_file = self.godspeed_dir / "Inbox.md" + + # Import the sync engine for consistent parsing + try: + from src.services.godspeed.sync import GodspeedSync + + self.sync_engine = GodspeedSync(None, godspeed_dir) + except ImportError: + # Fallback parsing if import fails + self.sync_engine = None + + def _parse_task_line_fallback(self, line: str) -> Tuple[str, str, str, str]: + """Fallback task parsing if sync engine not available.""" + # Match patterns like: - [ ] Task title + task_pattern = ( + r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*)?\s*$" + ) + match = re.match(task_pattern, line.strip()) + + if not match: + return None + + checkbox, title_and_notes, local_id = match.groups() + + # Determine status + if checkbox.lower() == "x": + status = "complete" + elif checkbox == "-": + status = "cleared" + else: + status = "incomplete" + + # Extract title (remove any inline notes after " + if notes: + formatted += f"\n {notes}" + + new_task_lines.append(formatted) + + # Combine with existing content + if existing_content.strip(): + new_content = ( + existing_content.rstrip() + "\n\n" + "\n".join(new_task_lines) + "\n" + ) + else: + new_content = "\n".join(new_task_lines) + "\n" + + with open(file_path, "w", encoding="utf-8") as f: + f.write(new_content) + + def _clean_file(self, file_path: Path, non_task_lines: List[str]): + """Remove tasks from original file, keeping only non-task content.""" + if not non_task_lines or all(not line.strip() for line in non_task_lines): + # File would be empty, delete it + if not self.dry_run: + file_path.unlink() + print(f" 🗑️ Would delete empty file: {file_path}") + else: + # Write back non-task content + cleaned_content = "\n".join(non_task_lines).strip() + if cleaned_content: + cleaned_content += "\n" + + if not self.dry_run: + with open(file_path, "w", encoding="utf-8") as f: + f.write(cleaned_content) + print(f" ✂️ Cleaned file (removed tasks): {file_path}") + + def find_markdown_files(self) -> List[Path]: + """Find all markdown files in the notes directory, excluding Godspeed directory.""" + markdown_files = [] + + for md_file in self.notes_dir.rglob("*.md"): + # Skip files in the Godspeed directory + if ( + self.godspeed_dir in md_file.parents + or md_file.parent == self.godspeed_dir + ): + continue + + # Skip hidden files and directories + if any(part.startswith(".") for part in md_file.parts): + continue + + markdown_files.append(md_file) + + return sorted(markdown_files) + + def sweep_tasks(self) -> dict: + """Sweep incomplete tasks from all markdown files into Inbox.""" + print(f"🧹 Sweeping incomplete tasks from: {self.notes_dir}") + print(f"📥 Target Inbox: {self.inbox_file}") + print(f"🔍 Dry run: {self.dry_run}") + print("=" * 60) + + markdown_files = self.find_markdown_files() + print(f"\n📁 Found {len(markdown_files)} markdown files to process") + + swept_tasks = [] + processed_files = [] + empty_files_deleted = [] + + for file_path in markdown_files: + rel_path = file_path.relative_to(self.notes_dir) + print(f"\n📄 Processing: {rel_path}") + + tasks, non_task_lines = self._parse_markdown_file(file_path) + if not tasks: + print(f" ℹ️ No tasks found") + continue + + # Separate incomplete tasks from completed/cleared ones + incomplete_tasks = [] + complete_tasks = [] + + for task in tasks: + local_id, status, title, notes = task + if status == "incomplete": + incomplete_tasks.append(task) + else: + complete_tasks.append(task) + + if incomplete_tasks: + print(f" 🔄 Found {len(incomplete_tasks)} incomplete tasks:") + for _, status, title, notes in incomplete_tasks: + print(f" • {title}") + if notes: + print(f" Notes: {notes}") + + # Add source file annotation + source_annotation = f"" + annotated_tasks = [] + for local_id, status, title, notes in incomplete_tasks: + # Add source info to notes + source_notes = f"From: {rel_path}" + if notes: + combined_notes = f"{notes}\n{source_notes}" + else: + combined_notes = source_notes + annotated_tasks.append((local_id, status, title, combined_notes)) + + swept_tasks.extend(annotated_tasks) + processed_files.append(str(rel_path)) + + if complete_tasks: + print( + f" ✅ Keeping {len(complete_tasks)} completed/cleared tasks in place" + ) + + # Reconstruct remaining content (non-tasks + completed tasks) + remaining_content = non_task_lines.copy() + + # Add completed/cleared tasks back to remaining content + if complete_tasks: + remaining_content.append("") # Empty line before tasks + for task in complete_tasks: + if self.sync_engine: + formatted = self.sync_engine._format_task_line(*task) + else: + local_id, status, title, notes = task + checkbox = { + "incomplete": "[ ]", + "complete": "[x]", + "cleared": "[-]", + }[status] + formatted = f"- {checkbox} {title} " + if notes: + formatted += f"\n {notes}" + remaining_content.append(formatted) + + # Clean the original file + if incomplete_tasks: + self._clean_file(file_path, remaining_content) + + # Write swept tasks to Inbox + if swept_tasks: + print(f"\n📥 Writing {len(swept_tasks)} tasks to Inbox...") + if not self.dry_run: + self._write_tasks_to_file(self.inbox_file, swept_tasks) + print(f" ✅ Inbox updated: {self.inbox_file}") + + # Summary + print(f"\n" + "=" * 60) + print(f"📊 SWEEP SUMMARY:") + print(f" • Files processed: {len(processed_files)}") + print(f" • Tasks swept: {len(swept_tasks)}") + print(f" • Target: {self.inbox_file}") + + if self.dry_run: + print(f"\n⚠️ DRY RUN - No files were actually modified") + print(f" Run without --dry-run to perform the sweep") + + return { + "swept_tasks": len(swept_tasks), + "processed_files": processed_files, + "inbox_file": str(self.inbox_file), + } + + +def main(): + parser = argparse.ArgumentParser( + description="Sweep incomplete tasks from markdown files into Godspeed Inbox", + epilog=""" +Examples: + python sweep_tasks.py ~/Documents/Notes ~/Documents/Godspeed + python sweep_tasks.py . ./godspeed --dry-run + python sweep_tasks.py ~/Notes ~/Notes/godspeed --dry-run + """, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + + parser.add_argument( + "notes_dir", + type=Path, + help="Root directory containing markdown files with tasks (e.g., ~/Documents/Notes)", + ) + + parser.add_argument( + "godspeed_dir", + type=Path, + help="Godspeed sync directory where Inbox.md will be created (e.g., ~/Documents/Godspeed)", + ) + + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be done without making changes", + ) + + args = parser.parse_args() + + # Validate directories + if not args.notes_dir.exists(): + print(f"❌ Notes directory does not exist: {args.notes_dir}") + return 1 + + if not args.notes_dir.is_dir(): + print(f"❌ Notes path is not a directory: {args.notes_dir}") + return 1 + + # Godspeed directory will be created if it doesn't exist + + try: + sweeper = TaskSweeper(args.notes_dir, args.godspeed_dir, args.dry_run) + result = sweeper.sweep_tasks() + + if result["swept_tasks"] > 0: + print(f"\n🎉 Successfully swept {result['swept_tasks']} tasks!") + if not args.dry_run: + print(f"💡 Next steps:") + print(f" 1. Review tasks in: {result['inbox_file']}") + print(f" 2. Run 'godspeed upload' to sync to API") + print(f" 3. Organize tasks into appropriate lists in Godspeed app") + else: + print(f"\n✨ No incomplete tasks found to sweep.") + + return 0 + + except Exception as e: + print(f"❌ Error during sweep: {e}") + import traceback + + traceback.print_exc() + return 1 + + +if __name__ == "__main__": + exit(main()) diff --git a/test_cancelled_tasks.py b/test_cancelled_tasks.py new file mode 100644 index 0000000..2e6a166 --- /dev/null +++ b/test_cancelled_tasks.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +""" +Test script for Godspeed cancelled task functionality. +""" + +import tempfile +from pathlib import Path + + +def test_cancelled_task_parsing(): + print("=== Testing Cancelled Task Support ===\n") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + print("1. Testing task status parsing:") + print("-" * 40) + + test_lines = [ + "- [ ] Incomplete task ", + "- [x] Completed task ", + "- [X] Also completed ", + "- [-] Cancelled task ", + ] + + for line in test_lines: + parsed = sync_engine._parse_task_line(line) + if parsed: + local_id, status, title, notes = parsed + icon = {"incomplete": "⏳", "complete": "✅", "cleared": "❌"}[status] + print(f" {icon} {status.upper()}: {title} (ID: {local_id})") + else: + print(f" ❌ Failed to parse: {line}") + + print("\n2. Testing task formatting:") + print("-" * 30) + + tasks = [ + ("task1", "incomplete", "Buy groceries", ""), + ("task2", "complete", "Call dentist", ""), + ("task3", "cleared", "Old project", "No longer needed"), + ] + + for local_id, status, title, notes in tasks: + formatted = sync_engine._format_task_line(local_id, status, title, notes) + print(f" {formatted}") + + print("\n3. Testing roundtrip with all statuses:") + print("-" * 42) + + # Write to file + test_file = sync_dir / "test_statuses.md" + sync_engine._write_list_file(test_file, tasks) + + # Read back + read_tasks = sync_engine._read_list_file(test_file) + + print(f" Original: {len(tasks)} tasks") + print(f" Read back: {len(read_tasks)} tasks") + + for original, read_back in zip(tasks, read_tasks): + orig_id, orig_status, orig_title, orig_notes = original + read_id, read_status, read_title, read_notes = read_back + + if orig_status == read_status and orig_title == read_title: + icon = {"incomplete": "⏳", "complete": "✅", "cleared": "❌"}[ + orig_status + ] + print(f" {icon} {orig_status.upper()}: {orig_title} - ✓ Match") + else: + print(f" ❌ Mismatch:") + print(f" Original: {orig_status}, '{orig_title}'") + print(f" Read: {read_status}, '{read_title}'") + + print("\n4. File content generated:") + print("-" * 25) + with open(test_file, "r") as f: + content = f.read() + print(content) + + print("5. API update simulation:") + print("-" * 27) + print("For cancelled task ([-]), would send:") + print(" PATCH /tasks/xyz {'is_complete': True, 'is_cleared': True}") + print("\nFor completed task ([x]), would send:") + print(" PATCH /tasks/abc {'is_complete': True, 'is_cleared': False}") + print("\nFor incomplete task ([ ]), would send:") + print(" PATCH /tasks/def {'is_complete': False, 'is_cleared': False}") + + +if __name__ == "__main__": + test_cancelled_task_parsing() diff --git a/test_completion_status.py b/test_completion_status.py new file mode 100644 index 0000000..1160fb5 --- /dev/null +++ b/test_completion_status.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +""" +Quick test to verify completion status handling in Godspeed sync. +""" + +import tempfile +from pathlib import Path + + +# Test the markdown parsing for completion status +def test_completion_parsing(): + print("Testing completion status parsing...") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + # Test different completion states + test_lines = [ + "- [ ] Incomplete task ", + "- [x] Completed task ", + "- [X] Also completed ", # Capital X + "- [ ] Another incomplete ", + ] + + for line in test_lines: + parsed = sync_engine._parse_task_line(line) + if parsed: + local_id, status, title, notes = parsed + display = "✓ Complete" if status == "complete" else "○ Incomplete" + print(f" {display}: {title} (ID: {local_id})") + else: + print(f" Failed to parse: {line}") + + +def test_format_task(): + print("\nTesting task formatting...") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + # Test formatting both completion states + incomplete_line = sync_engine._format_task_line( + "abc123", "incomplete", "Buy milk", "" + ) + completed_line = sync_engine._format_task_line( + "def456", "complete", "Call mom", "" + ) + with_notes_line = sync_engine._format_task_line( + "ghi789", "incomplete", "Project work", "Due Friday" + ) + + print(f" Incomplete: {incomplete_line}") + print(f" Completed: {completed_line}") + print(f" With notes: {with_notes_line}") + + +def test_roundtrip(): + print("\nTesting roundtrip parsing...") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + # Original tasks with different completion states + original_tasks = [ + ("task1", "incomplete", "Buy groceries", "From whole foods"), + ("task2", "complete", "Call dentist", ""), + ("task3", "incomplete", "Finish report", "Due Monday"), + ("task4", "complete", "Exercise", "Went for a run"), + ] + + # Write to file + test_file = sync_dir / "test_roundtrip.md" + sync_engine._write_list_file(test_file, original_tasks) + + # Read back + read_tasks = sync_engine._read_list_file(test_file) + + print(f" Original: {len(original_tasks)} tasks") + print(f" Read back: {len(read_tasks)} tasks") + + for i, (original, read_back) in enumerate(zip(original_tasks, read_tasks)): + orig_id, orig_status, orig_title, orig_notes = original + read_id, read_status, read_title, read_notes = read_back + + if orig_status == read_status and orig_title == read_title: + display = "✓ Complete" if orig_status == "complete" else "○ Incomplete" + print(f" {display}: {orig_title} - ✓ Match") + else: + print(f" ✗ Mismatch on task {i + 1}:") + print(f" Original: status={orig_status}, title='{orig_title}'") + print(f" Read: status={read_status}, title='{read_title}'") + + +if __name__ == "__main__": + print("=== Godspeed Completion Status Test ===\n") + + try: + test_completion_parsing() + test_format_task() + test_roundtrip() + + print("\n=== Test Summary ===") + print("✓ Completion status handling is working correctly!") + print("\nExpected behavior:") + print("- [ ] tasks sync as incomplete (is_complete=False)") + print("- [x] tasks sync as completed (is_complete=True)") + print("- Status changes in markdown files will sync to Godspeed") + print("- Status changes in Godspeed will sync to markdown files") + + except Exception as e: + print(f"\n✗ Test failed: {e}") + import traceback + + traceback.print_exc() diff --git a/test_godspeed_sync.py b/test_godspeed_sync.py new file mode 100644 index 0000000..0768fd4 --- /dev/null +++ b/test_godspeed_sync.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +""" +Test script for Godspeed sync functionality. + +This script demonstrates the Godspeed sync tool by creating sample data +and testing various sync scenarios. +""" + +import os +import tempfile +from pathlib import Path +import json + +# Mock data for testing without real API calls +MOCK_LISTS = [ + {"id": "list-1", "name": "Personal"}, + {"id": "list-2", "name": "Work Projects"}, + {"id": "list-3", "name": "Shopping"}, +] + +MOCK_TASKS = [ + { + "id": "task-1", + "title": "Buy groceries", + "list_id": "list-3", + "is_complete": False, + "notes": "Don't forget milk and eggs", + }, + { + "id": "task-2", + "title": "Finish quarterly report", + "list_id": "list-2", + "is_complete": False, + "notes": "Due Friday", + }, + { + "id": "task-3", + "title": "Call dentist", + "list_id": "list-1", + "is_complete": True, + "notes": "", + }, + { + "id": "task-4", + "title": "Review pull requests", + "list_id": "list-2", + "is_complete": False, + "notes": "Check PR #123 and #124", + }, +] + + +class MockGodspeedClient: + """Mock client for testing without hitting real API.""" + + def __init__(self, **kwargs): + pass + + def get_lists(self): + return MOCK_LISTS + + def get_tasks(self, **kwargs): + filtered_tasks = MOCK_TASKS + if kwargs.get("list_id"): + filtered_tasks = [ + t for t in MOCK_TASKS if t["list_id"] == kwargs["list_id"] + ] + if kwargs.get("status"): + if kwargs["status"] == "complete": + filtered_tasks = [t for t in filtered_tasks if t["is_complete"]] + elif kwargs["status"] == "incomplete": + filtered_tasks = [t for t in filtered_tasks if not t["is_complete"]] + + # Mock the API response format + lists_dict = {lst["id"]: lst for lst in MOCK_LISTS} + return {"tasks": filtered_tasks, "lists": lists_dict} + + def create_task(self, **kwargs): + new_task = { + "id": f"task-{len(MOCK_TASKS) + 1}", + "title": kwargs["title"], + "list_id": kwargs.get("list_id"), + "is_complete": False, + "notes": kwargs.get("notes", ""), + } + MOCK_TASKS.append(new_task) + return new_task + + def update_task(self, task_id, **kwargs): + for task in MOCK_TASKS: + if task["id"] == task_id: + task.update(kwargs) + return task + raise Exception(f"Task {task_id} not found") + + def complete_task(self, task_id): + return self.update_task(task_id, is_complete=True) + + +def test_markdown_parsing(): + """Test markdown task parsing functionality.""" + print("Testing markdown parsing...") + + from src.services.godspeed.sync import GodspeedSync + + # Create temporary directory + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + # Test task line parsing + test_lines = [ + "- [ ] Simple task ", + "- [x] Completed task ", + "- [ ] Task with notes \n Some additional notes here", + "- [ ] New task without ID", + ] + + for line in test_lines: + parsed = sync_engine._parse_task_line(line) + if parsed: + local_id, is_complete, title, notes = parsed + print(f" Parsed: {title} (ID: {local_id}, Complete: {is_complete})") + if notes: + print(f" Notes: {notes}") + else: + print(f" Failed to parse: {line}") + + +def test_file_operations(): + """Test file reading and writing operations.""" + print("\nTesting file operations...") + + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + sync_engine = GodspeedSync(None, sync_dir) + + # Create test tasks + test_tasks = [ + ("abc123", False, "Buy milk", "From the grocery store"), + ("def456", True, "Call mom", ""), + ("ghi789", False, "Finish project", "Due next week"), + ] + + # Write tasks to file + test_file = sync_dir / "test_list.md" + sync_engine._write_list_file(test_file, test_tasks) + print(f" Created test file: {test_file}") + + # Read tasks back + read_tasks = sync_engine._read_list_file(test_file) + print(f" Read {len(read_tasks)} tasks back from file") + + for i, (original, read_back) in enumerate(zip(test_tasks, read_tasks)): + if original == read_back: + print(f" Task {i + 1}: ✓ Match") + else: + print(f" Task {i + 1}: ✗ Mismatch") + print(f" Original: {original}") + print(f" Read back: {read_back}") + + +def test_mock_sync(): + """Test sync operations with mock data.""" + print("\nTesting sync with mock data...") + + # Temporarily replace the real client + import src.services.godspeed.sync as sync_module + + original_client_class = sync_module.GodspeedClient + sync_module.GodspeedClient = MockGodspeedClient + + try: + from src.services.godspeed.sync import GodspeedSync + + with tempfile.TemporaryDirectory() as temp_dir: + sync_dir = Path(temp_dir) + + # Create mock client and sync engine + mock_client = MockGodspeedClient() + sync_engine = GodspeedSync(mock_client, sync_dir) + + # Test download + print(" Testing download...") + sync_engine.download_from_api() + + # Check created files + md_files = list(sync_dir.glob("*.md")) + print(f" Created {len(md_files)} markdown files") + + for md_file in md_files: + tasks = sync_engine._read_list_file(md_file) + print(f" {md_file.name}: {len(tasks)} tasks") + + # Test status + status = sync_engine.get_sync_status() + print( + f" Status: {status['local_files']} files, {status['total_local_tasks']} tasks" + ) + + # Test upload (modify a file first) + if md_files: + first_file = md_files[0] + with open(first_file, "a") as f: + f.write("- [ ] New local task \n") + + print(" Testing upload...") + sync_engine.upload_to_api() + print( + f" Upload completed, now {len(MOCK_TASKS)} total tasks in mock data" + ) + + finally: + # Restore original client + sync_module.GodspeedClient = original_client_class + + +def test_cli_integration(): + """Test CLI commands (without real API calls).""" + print("\nTesting CLI integration...") + + # Test that imports work + try: + from src.cli.godspeed import godspeed, get_sync_directory + + print(" ✓ CLI imports successful") + + # Test sync directory detection + sync_dir = get_sync_directory() + print(f" ✓ Sync directory: {sync_dir}") + + except ImportError as e: + print(f" ✗ CLI import failed: {e}") + + +def main(): + """Run all tests.""" + print("=== Godspeed Sync Test Suite ===\n") + + try: + test_markdown_parsing() + test_file_operations() + test_mock_sync() + test_cli_integration() + + print("\n=== Test Summary ===") + print("✓ All tests completed successfully!") + print("\nTo use the real Godspeed sync:") + print("1. Set environment variables:") + print(" export GODSPEED_EMAIL='your@email.com'") + print(" export GODSPEED_PASSWORD='your-password'") + print(" # OR") + print(" export GODSPEED_TOKEN='your-api-token'") + print("") + print("2. Run sync commands:") + print(" python -m src.cli.godspeed download") + print(" python -m src.cli.godspeed status") + print(" python -m src.cli.godspeed sync") + + except Exception as e: + print(f"\n✗ Test failed: {e}") + import traceback + + traceback.print_exc() + + +if __name__ == "__main__": + main() diff --git a/test_task_sweeper.py b/test_task_sweeper.py new file mode 100644 index 0000000..bfb2ff1 --- /dev/null +++ b/test_task_sweeper.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +""" +Test and demo script for the task sweeper functionality. +""" + +import tempfile +from pathlib import Path +import os +import sys + + +def create_test_structure(): + """Create a test directory structure with scattered tasks.""" + # Add the project root to path so we can import + project_root = Path(__file__).parent + sys.path.insert(0, str(project_root)) + + with tempfile.TemporaryDirectory() as temp_dir: + base_dir = Path(temp_dir) + + print("🏗️ Creating test directory structure...") + + # Create year directories with markdown files + (base_dir / "2024" / "projects").mkdir(parents=True) + (base_dir / "2024" / "notes").mkdir(parents=True) + (base_dir / "2025" / "planning").mkdir(parents=True) + (base_dir / "archive").mkdir(parents=True) + (base_dir / "godspeed").mkdir(parents=True) + + # Create test files with various tasks + test_files = { + "2024/projects/website.md": """# Website Redesign Project + +## Overview +This project aims to redesign our company website. + +## Tasks +- [x] Create wireframes +- [ ] Design mockups + Need to use new brand colors +- [ ] Get client approval +- [-] Old approach that was cancelled + +## Notes +The wireframes are complete and approved. +""", + "2024/notes/meeting-notes.md": """# Weekly Team Meeting - Dec 15 + +## Attendees +- Alice, Bob, Charlie + +## Action Items +- [ ] Alice: Update documentation +- [x] Bob: Fix bug #456 +- [ ] Charlie: Review PR #789 + Needs to be done by Friday + +## Discussion +We discussed the quarterly goals. +""", + "2025/planning/goals.md": """# 2025 Goals + +## Q1 Objectives +- [ ] Launch new feature +- [ ] Improve performance by 20% + Focus on database queries + +## Q2 Ideas +- [ ] Consider mobile app + +Some general notes about the year ahead. +""", + "archive/old-project.md": """# Old Project (Archived) + +This project is mostly done but has some lingering tasks. + +- [x] Phase 1 complete +- [-] Phase 2 cancelled +- [ ] Cleanup remaining files + Need to remove temp directories +""", + "random-notes.md": """# Random Notes + +Just some thoughts and incomplete todos: + +- [ ] Call the dentist +- [ ] Buy groceries + - Milk + - Bread + - Eggs + +No other tasks here, just notes. +""", + "godspeed/Personal.md": """# This file should be ignored by the sweeper +- [ ] Existing Godspeed task +""", + } + + # Write test files + for rel_path, content in test_files.items(): + file_path = base_dir / rel_path + with open(file_path, "w") as f: + f.write(content) + + print(f"📁 Created test structure in: {base_dir}") + print(f"📄 Files created:") + for rel_path in test_files.keys(): + print(f" • {rel_path}") + + return base_dir + + +def test_sweeper(): + """Test the task sweeper functionality.""" + print("=" * 60) + print("🧪 TESTING TASK SWEEPER") + print("=" * 60) + + # Create test directory + with tempfile.TemporaryDirectory() as temp_dir: + base_dir = Path(temp_dir) + + # Create the test structure directly here since we can't return from context manager + (base_dir / "2024" / "projects").mkdir(parents=True) + (base_dir / "2024" / "notes").mkdir(parents=True) + (base_dir / "2025" / "planning").mkdir(parents=True) + (base_dir / "archive").mkdir(parents=True) + (base_dir / "godspeed").mkdir(parents=True) + + test_files = { + "2024/projects/website.md": """# Website Redesign Project + +- [x] Create wireframes +- [ ] Design mockups + Need to use new brand colors +- [ ] Get client approval +- [-] Old approach that was cancelled + +Project notes here. +""", + "2024/notes/meeting-notes.md": """# Weekly Team Meeting + +- [ ] Alice: Update documentation +- [x] Bob: Fix bug #456 +- [ ] Charlie: Review PR #789 + Needs to be done by Friday +""", + "2025/planning/goals.md": """# 2025 Goals + +- [ ] Launch new feature +- [ ] Improve performance by 20% + Focus on database queries +""", + "random-notes.md": """# Random Notes + +- [ ] Call the dentist +- [ ] Buy groceries + +Just some notes. +""", + } + + for rel_path, content in test_files.items(): + file_path = base_dir / rel_path + with open(file_path, "w") as f: + f.write(content) + + godspeed_dir = base_dir / "godspeed" + + print(f"\n📁 Test directory: {base_dir}") + print(f"📥 Godspeed directory: {godspeed_dir}") + + # Import and run the sweeper + from sweep_tasks import TaskSweeper + + print(f"\n🧹 Running task sweeper (DRY RUN)...") + sweeper = TaskSweeper(base_dir, godspeed_dir, dry_run=True) + result = sweeper.sweep_tasks() + + print(f"\n🔍 DRY RUN RESULTS:") + print(f" • Would sweep: {result['swept_tasks']} tasks") + print(f" • From files: {len(result['processed_files'])}") + + if result["processed_files"]: + print(f" • Files that would be modified:") + for file_path in result["processed_files"]: + print(f" - {file_path}") + + # Now run for real + print(f"\n🚀 Running task sweeper (REAL)...") + sweeper_real = TaskSweeper(base_dir, godspeed_dir, dry_run=False) + result_real = sweeper_real.sweep_tasks() + + # Check the inbox + inbox_file = godspeed_dir / "Inbox.md" + if inbox_file.exists(): + print(f"\n📥 Inbox.md contents:") + print("-" * 40) + with open(inbox_file, "r") as f: + print(f.read()) + print("-" * 40) + + # Check a cleaned file + website_file = base_dir / "2024" / "projects" / "website.md" + if website_file.exists(): + print(f"\n📄 Cleaned file (website.md) contents:") + print("-" * 30) + with open(website_file, "r") as f: + print(f.read()) + print("-" * 30) + + print(f"\n✅ TEST COMPLETE!") + print(f" • Swept {result_real['swept_tasks']} incomplete tasks") + print(f" • Into: {result_real['inbox_file']}") + + +if __name__ == "__main__": + test_sweeper()