sirtimbly/luk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

godspeed app sync

Browse Source
This commit is contained in:
Tim Bendt
2025-08-20 08:30:54 -04:00
parent ca6e4cdf5d
commit c46d53b261
17 changed files with 3170 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
GODSPEED_SYNC.md Normal file
View File

@@ -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 <!-- id:abc123 -->` 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 <!-- id:abc123 -->
- [x] Buy groceries <!-- id:def456 -->
Don't forget milk and eggs
- [-] Old project <!-- id:ghi789 -->
- [ ] New task I just added <!-- id:jkl012 -->
# Work_Projects.md
- [ ] Finish quarterly report <!-- id:xyz890 -->
Due Friday
- [-] Cancelled meeting <!-- id:uvw567 -->
```
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 <!-- id:abc123 -->
- [x] Completed task <!-- id:def456 -->
- [X] Also completed (capital X works too) <!-- id:ghi789 -->
- [-] Cancelled/cleared task <!-- id:jkl012 -->
- [ ] Task with notes <!-- id:mno345 -->
Notes go on the next line, indented
```
### Important Notes:
- **Don't remove the `<!-- id:xxx -->` 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.

134
TASK_SWEEPER.md Normal file
View File

@@ -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.

110
demo_cancelled_workflow.py Normal file
View File

@@ -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 <!-- id:task1 -->
- [-] Get approval from stakeholders <!-- id:task2 -->
Stakeholders decided to cancel this feature
- [-] Implement feature <!-- id:task3 -->
No longer needed since feature was cancelled
- [-] Write documentation <!-- id:task4 -->
Documentation not needed for cancelled feature
- [-] Deploy to production <!-- id:task5 -->
Cannot deploy cancelled feature
- [ ] Archive project files <!-- id:task6 -->
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()

103
demo_completion_sync.py Normal file
View File

@@ -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 <!-- id:task005 -->\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()

2
pyproject.toml
View File

@@ -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",

4
src/cli/__init__.py
View File

@@ -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")

616
src/cli/godspeed.py Normal file
View File

@@ -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 <!-- id:abc123 -->
task_pattern = (
r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*<!--\s*id:(\w+)\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 <!--)
title = title_and_notes.split("<!--")[0].strip()
# Generate ID if missing
if not local_id:
if hasattr(self, "sync_engine") and self.sync_engine:
local_id = self.sync_engine._generate_local_id()
else:
import uuid
local_id = str(uuid.uuid4())[:8]
return local_id, status, title, ""
def _parse_markdown_file(self, file_path: Path):
"""Parse a markdown file and extract tasks and non-task content."""
if not file_path.exists():
return [], []
tasks = []
non_task_lines = []
try:
import builtins
with builtins.open(str(file_path), "r", encoding="utf-8") as f:
lines = f.readlines()
except Exception as e:
click.echo(f" ⚠️ Error reading {file_path}: {e}")
return [], []
for i, line in enumerate(lines):
line = line.rstrip()
# Check if this line looks like a task
if line.strip().startswith("- ["):
# Always use fallback parsing
parsed = self._parse_task_line_fallback(line)
if parsed:
tasks.append(parsed)
continue
# Not a task, keep as regular content
non_task_lines.append(line)
return tasks, non_task_lines
def _write_tasks_to_file(self, file_path: Path, tasks):
"""Write tasks to a markdown file."""
if not tasks:
return
file_path.parent.mkdir(parents=True, exist_ok=True)
import builtins
# Read existing content if file exists
existing_content = ""
if file_path.exists():
with builtins.open(str(file_path), "r", encoding="utf-8") as f:
existing_content = f.read()
# Format new tasks
new_task_lines = []
for local_id, status, title, notes in tasks:
if self.sync_engine:
formatted = self.sync_engine._format_task_line(
local_id, status, title, notes
)
else:
# Fallback formatting
checkbox = {"incomplete": "[ ]", "complete": "[x]", "cleared": "[-]"}[
status
]
formatted = f"- {checkbox} {title} <!-- id:{local_id} -->"
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} <!-- id:{local_id} -->"
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 <notes_dir> [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()

300
src/cli/sync.py
View File

@@ -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:
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}"
)
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} | Outbox: {pending_email_count} pending. Starting sync...",
"yellow",
)
)
elif mail_changes and calendar_changes:
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)})",

0
src/services/godspeed/__init__.py Normal file
View File

129
src/services/godspeed/client.py Normal file
View File

@@ -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)

87
src/services/godspeed/config.py Normal file
View File

@@ -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),
}

395
src/services/godspeed/sync.py Normal file
View File

@@ -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 <!-- id:abc123 -->
# - [x] Completed task <!-- id:def456 -->
# - [-] Cleared/cancelled task <!-- id:ghi789 -->
# - [ ] Task with notes <!-- id:jkl012 --> Some notes here
task_pattern = r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*<!--\s*id:(\w+)\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("<!--")[0].strip()
notes = extra_notes.strip() if extra_notes else ""
if not local_id:
local_id = self._generate_local_id()
return local_id, status, title_parts, notes
def _format_task_line(
self, local_id: str, status: str, title: str, notes: str = ""
) -> str:
"""Format a task as a markdown line with ID tracking."""
if status == "complete":
checkbox = "[x]"
elif status == "cleared":
checkbox = "[-]"
else:
checkbox = "[ ]"
line = f"- {checkbox} {title} <!-- id:{local_id} -->"
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"),
}

381
sweep_tasks.py Executable file
View File

@@ -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 <!-- id:abc123 -->
task_pattern = (
r"^\s*-\s*\[([xX\s\-])\]\s*(.+?)(?:\s*<!--\s*id:(\w+)\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 <!--)
title = title_and_notes.split("<!--")[0].strip()
# Generate ID if missing
if not local_id:
import uuid
local_id = str(uuid.uuid4())[:8]
return local_id, status, title, ""
def _parse_markdown_file(self, file_path: Path) -> Tuple[List[Tuple], List[str]]:
"""Parse a markdown file and extract tasks and non-task content."""
if not file_path.exists():
return [], []
tasks = []
non_task_lines = []
try:
with open(file_path, "r", encoding="utf-8") as f:
lines = f.readlines()
except Exception as e:
print(f" ⚠️ Error reading {file_path}: {e}")
return [], []
i = 0
while i < len(lines):
line = lines[i].rstrip()
# Check if this line looks like a task
if line.strip().startswith("- ["):
# Try to parse with sync engine first
if self.sync_engine:
# Collect potential multi-line task
task_block = [line]
j = i + 1
while (
j < len(lines)
and lines[j].strip()
and not lines[j].strip().startswith("- [")
):
task_block.append(lines[j].rstrip())
j += 1
task_text = "\n".join(task_block)
parsed = self.sync_engine._parse_task_line(task_text)
if parsed:
tasks.append(parsed)
i = j # Skip the lines we've processed
continue
# Fallback parsing
parsed = self._parse_task_line_fallback(line)
if parsed:
tasks.append(parsed)
i += 1
continue
# Not a task, keep as regular content
non_task_lines.append(line)
i += 1
return tasks, non_task_lines
def _write_tasks_to_file(self, file_path: Path, tasks: List[Tuple]):
"""Write tasks to a markdown file."""
if not tasks:
return
file_path.parent.mkdir(parents=True, exist_ok=True)
# Read existing content if file exists
existing_content = ""
if file_path.exists():
with open(file_path, "r", encoding="utf-8") as f:
existing_content = f.read()
# Format new tasks
new_task_lines = []
for local_id, status, title, notes in tasks:
if self.sync_engine:
formatted = self.sync_engine._format_task_line(
local_id, status, title, notes
)
else:
# Fallback formatting
checkbox = {"incomplete": "[ ]", "complete": "[x]", "cleared": "[-]"}[
status
]
formatted = f"- {checkbox} {title} <!-- id:{local_id} -->"
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"<!-- Swept from {rel_path} on {datetime.now().strftime('%Y-%m-%d %H:%M')} -->"
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} <!-- id:{local_id} -->"
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())

95
test_cancelled_tasks.py Normal file
View File

@@ -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 <!-- id:abc123 -->",
"- [x] Completed task <!-- id:def456 -->",
"- [X] Also completed <!-- id:ghi789 -->",
"- [-] Cancelled task <!-- id:jkl012 -->",
]
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()

123
test_completion_status.py Normal file
View File

@@ -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 <!-- id:abc123 -->",
"- [x] Completed task <!-- id:def456 -->",
"- [X] Also completed <!-- id:ghi789 -->", # Capital X
"- [ ] Another incomplete <!-- id:jkl012 -->",
]
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()

270
test_godspeed_sync.py Normal file
View File

@@ -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 <!-- id:abc123 -->",
"- [x] Completed task <!-- id:def456 -->",
"- [ ] Task with notes <!-- id:ghi789 -->\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 <!-- id:newlocal -->\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()

218
test_task_sweeper.py Normal file
View File

@@ -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 <!-- id:wire123 -->
- [ ] Design mockups <!-- id:mock456 -->
Need to use new brand colors
- [ ] Get client approval <!-- id:appr789 -->
- [-] Old approach that was cancelled <!-- id:old999 -->
## 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 <!-- id:doc123 -->
- [x] Bob: Fix bug #456 <!-- id:bug456 -->
- [ ] Charlie: Review PR #789 <!-- id:pr789 -->
Needs to be done by Friday
## Discussion
We discussed the quarterly goals.
""",
"2025/planning/goals.md": """# 2025 Goals
## Q1 Objectives
- [ ] Launch new feature <!-- id:feat2025 -->
- [ ] Improve performance by 20% <!-- id:perf2025 -->
Focus on database queries
## Q2 Ideas
- [ ] Consider mobile app <!-- id:mobile2025 -->
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 <!-- id:p1done -->
- [-] Phase 2 cancelled <!-- id:p2cancel -->
- [ ] Cleanup remaining files <!-- id:cleanup123 -->
Need to remove temp directories
""",
"random-notes.md": """# Random Notes
Just some thoughts and incomplete todos:
- [ ] Call the dentist <!-- id:dentist99 -->
- [ ] Buy groceries <!-- id:grocery99 -->
- Milk
- Bread
- Eggs
No other tasks here, just notes.
""",
"godspeed/Personal.md": """# This file should be ignored by the sweeper
- [ ] Existing Godspeed task <!-- id:existing1 -->
""",
}
# 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 <!-- id:wire123 -->
- [ ] Design mockups <!-- id:mock456 -->
Need to use new brand colors
- [ ] Get client approval <!-- id:appr789 -->
- [-] Old approach that was cancelled <!-- id:old999 -->
Project notes here.
""",
"2024/notes/meeting-notes.md": """# Weekly Team Meeting
- [ ] Alice: Update documentation <!-- id:doc123 -->
- [x] Bob: Fix bug #456 <!-- id:bug456 -->
- [ ] Charlie: Review PR #789 <!-- id:pr789 -->
Needs to be done by Friday
""",
"2025/planning/goals.md": """# 2025 Goals
- [ ] Launch new feature <!-- id:feat2025 -->
- [ ] Improve performance by 20% <!-- id:perf2025 -->
Focus on database queries
""",
"random-notes.md": """# Random Notes
- [ ] Call the dentist <!-- id:dentist99 -->
- [ ] Buy groceries <!-- id:grocery99 -->
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()