660 lines
18 KiB
Python
660 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
CodiMD CLI - Main entry point.
|
|
|
|
A stateful CLI for CodiMD collaborative markdown notes.
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import click
|
|
import traceback
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from cli_anything.codimd.core import (
|
|
Config, Session, CodiMDClient,
|
|
NoteManager, UserManager, ExportManager, RevisionManager
|
|
)
|
|
from cli_anything.codimd.utils import output_json, format_error, format_success
|
|
|
|
|
|
# Global context
|
|
ctx = {"json_mode": False}
|
|
|
|
|
|
def handle_error(func):
|
|
"""Error handler wrapper for CLI commands."""
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except PermissionError as e:
|
|
if ctx["json_mode"]:
|
|
output_json(format_error(str(e)))
|
|
else:
|
|
click.echo(f"Error: {e}", err=True)
|
|
sys.exit(1)
|
|
except FileNotFoundError as e:
|
|
if ctx["json_mode"]:
|
|
output_json(format_error(str(e)))
|
|
else:
|
|
click.echo(f"Not found: {e}", err=True)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
if ctx["json_mode"]:
|
|
output_json(format_error(f"{type(e).__name__}: {e}"))
|
|
else:
|
|
click.echo(f"Error: {e}", err=True)
|
|
if "--debug" in sys.argv:
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
return wrapper
|
|
|
|
|
|
def get_client() -> CodiMDClient:
|
|
"""Get authenticated client."""
|
|
config = Config()
|
|
session = Session(config.config_dir)
|
|
|
|
client = CodiMDClient(
|
|
server=config.server,
|
|
session_cookies=session.get_cookies(),
|
|
timeout=config.timeout,
|
|
verify_ssl=config.verify_ssl
|
|
)
|
|
return client, session
|
|
|
|
|
|
@click.group(invoke_without_command=True)
|
|
@click.option("--json", is_flag=True, help="Output in JSON format")
|
|
@click.option("--server", help="CodiMD server URL")
|
|
@click.option("--version", is_flag=True, help="Show version")
|
|
@click.pass_context
|
|
def cli(ctx, json, server, version):
|
|
"""CodiMD CLI - Collaborative markdown notes."""
|
|
if version:
|
|
click.echo("CodiMD CLI v0.1.0")
|
|
return
|
|
|
|
# Set JSON mode
|
|
globals()["ctx"]["json_mode"] = json
|
|
|
|
# Set server if provided
|
|
if server:
|
|
config = Config()
|
|
config.server = server
|
|
|
|
# Show help if no command
|
|
if ctx.invoked_subcommand is None:
|
|
click.echo(ctx.get_help())
|
|
|
|
|
|
@cli.group()
|
|
def note():
|
|
"""Note operations."""
|
|
pass
|
|
|
|
|
|
@cli.group()
|
|
def user():
|
|
"""User operations."""
|
|
pass
|
|
|
|
|
|
@cli.group()
|
|
def export():
|
|
"""Export operations."""
|
|
pass
|
|
|
|
|
|
@cli.group()
|
|
def revision():
|
|
"""Revision operations."""
|
|
pass
|
|
|
|
|
|
@cli.group()
|
|
def config():
|
|
"""Configuration commands."""
|
|
pass
|
|
|
|
|
|
# Note commands
|
|
@note.command("list")
|
|
@click.option("--limit", type=int, help="Limit number of results")
|
|
@handle_error
|
|
def list_notes(limit):
|
|
"""List user's notes."""
|
|
client, session = get_client()
|
|
|
|
if not session.is_authenticated():
|
|
raise PermissionError("Not authenticated. Please login first.")
|
|
|
|
manager = NoteManager(client)
|
|
notes = manager.list_my_notes()
|
|
|
|
if limit:
|
|
notes = notes[:limit]
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"notes": notes, "count": len(notes)}))
|
|
else:
|
|
if not notes:
|
|
click.echo("No notes found.")
|
|
else:
|
|
for n in notes:
|
|
note_id = n.get('id', 'unknown')
|
|
# Decode Unicode escape sequences in text
|
|
text = n.get('text', 'Untitled')
|
|
# Handle both escaped unicode and normal text
|
|
if '\\u' in text:
|
|
try:
|
|
import json
|
|
text = json.loads(f'"{text}"')
|
|
except:
|
|
pass
|
|
click.echo(f"{note_id} - {text}")
|
|
|
|
|
|
@note.command("get")
|
|
@click.argument("note_id")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def get_note(note_id, output):
|
|
"""Get note content."""
|
|
client, session = get_client()
|
|
manager = NoteManager(client)
|
|
|
|
# For content, we need to get it via the download endpoint
|
|
export_manager = ExportManager(client)
|
|
content = export_manager.export_markdown(note_id)
|
|
|
|
if output:
|
|
output.write_text(content)
|
|
click.echo(f"Note saved to {output}")
|
|
else:
|
|
click.echo(content)
|
|
|
|
|
|
@note.command("create")
|
|
@click.option("--content", "-c", help="Note content")
|
|
@click.option("--file", "-f", type=Path, help="Read content from file")
|
|
@click.option("--alias", "-a", help="Note alias")
|
|
@handle_error
|
|
def create_note(content, file, alias):
|
|
"""Create a new note."""
|
|
client, session = get_client()
|
|
|
|
note_content = ""
|
|
if file:
|
|
note_content = file.read_text()
|
|
elif content:
|
|
note_content = content
|
|
|
|
manager = NoteManager(client)
|
|
result = manager.create_note(note_content, alias)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(result))
|
|
else:
|
|
click.echo(f"Note created: {result.get('url')}")
|
|
click.echo(f"ID: {result.get('id')}")
|
|
|
|
|
|
@note.command("update")
|
|
@click.argument("note_id")
|
|
@click.option("--content", "-c", help="New content")
|
|
@click.option("--file", "-f", type=Path, help="Read content from file")
|
|
@handle_error
|
|
def update_note(note_id, content, file):
|
|
"""Update note content."""
|
|
client, session = get_client()
|
|
|
|
note_content = ""
|
|
if file:
|
|
note_content = file.read_text()
|
|
elif content:
|
|
note_content = content
|
|
else:
|
|
raise ValueError("Either --content or --file must be provided")
|
|
|
|
manager = NoteManager(client)
|
|
result = manager.update_note(note_id, note_content)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(result))
|
|
else:
|
|
click.echo(f"Note updated: {note_id}")
|
|
|
|
|
|
@note.command("delete")
|
|
@click.argument("note_id")
|
|
@click.confirmation_option(prompt="Are you sure you want to delete this note?")
|
|
@handle_error
|
|
def delete_note(note_id):
|
|
"""Delete a note."""
|
|
client, session = get_client()
|
|
manager = NoteManager(client)
|
|
manager.delete_note(note_id)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"deleted": note_id}))
|
|
else:
|
|
click.echo(f"Note deleted: {note_id}")
|
|
|
|
|
|
@note.command("info")
|
|
@click.argument("note_id")
|
|
@handle_error
|
|
def note_info(note_id):
|
|
"""Get note metadata."""
|
|
client, session = get_client()
|
|
manager = NoteManager(client)
|
|
info = manager.get_note_info(note_id)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(info))
|
|
else:
|
|
click.echo(f"Note ID: {note_id}")
|
|
for key, value in info.items():
|
|
if key != "id":
|
|
click.echo(f" {key}: {value}")
|
|
|
|
|
|
@note.command("publish")
|
|
@click.argument("note_id")
|
|
@handle_error
|
|
def publish_note(note_id):
|
|
"""Get or create publish link for note."""
|
|
client, session = get_client()
|
|
manager = NoteManager(client)
|
|
result = manager.get_publish_link(note_id)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(result))
|
|
else:
|
|
click.echo(f"Published note: {result.get('url')}")
|
|
|
|
|
|
# User commands
|
|
@user.command("status")
|
|
@handle_error
|
|
def user_status():
|
|
"""Check login status."""
|
|
config = Config()
|
|
session = Session(config.config_dir)
|
|
|
|
if session.is_authenticated():
|
|
user = session.get_user()
|
|
status_data = {
|
|
"authenticated": True,
|
|
"server": config.server,
|
|
"user": user
|
|
}
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(status_data))
|
|
else:
|
|
click.echo(f"✓ Logged in to {config.server}")
|
|
if user:
|
|
# Get name from either direct field or nested profile
|
|
name = user.get("name") or user.get("profile", {}).get("name", "N/A")
|
|
click.echo(f" User: {name}")
|
|
else:
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({
|
|
"authenticated": False,
|
|
"server": config.server
|
|
}))
|
|
else:
|
|
click.echo(f"✗ Not logged in to {config.server}")
|
|
click.echo(" Use 'cli-anything-codimd user login' to login")
|
|
|
|
|
|
@user.command("me")
|
|
@handle_error
|
|
def user_me():
|
|
"""Get current user info."""
|
|
client, session = get_client()
|
|
|
|
if not session.is_authenticated():
|
|
raise PermissionError("Not authenticated. Please login first.")
|
|
|
|
manager = UserManager(client)
|
|
user = manager.get_me()
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(user))
|
|
else:
|
|
from cli_anything.codimd.utils import format_user_info
|
|
click.echo(format_user_info(user))
|
|
|
|
|
|
@user.command("export")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def user_export(output):
|
|
"""Export all user data."""
|
|
client, session = get_client()
|
|
|
|
if not session.is_authenticated():
|
|
raise PermissionError("Not authenticated. Please login first.")
|
|
|
|
manager = UserManager(client)
|
|
data = manager.export_my_data()
|
|
|
|
if output:
|
|
output.write_text(json.dumps(data, indent=2))
|
|
click.echo(f"Data exported to {output}")
|
|
elif ctx["json_mode"]:
|
|
output_json(format_success(data))
|
|
else:
|
|
click.echo(json.dumps(data, indent=2))
|
|
|
|
|
|
@user.command("login")
|
|
@click.option("--email", help="Email for login")
|
|
@click.option("--password", help="Password for login")
|
|
@handle_error
|
|
def user_login(email, password):
|
|
"""Login to CodiMD server."""
|
|
config = Config()
|
|
session = Session(config.config_dir)
|
|
|
|
if not email:
|
|
email = click.prompt("Email")
|
|
if not password:
|
|
password = click.prompt("Password", hide_input=True)
|
|
|
|
# Create client for login
|
|
client = CodiMDClient(
|
|
server=config.server,
|
|
timeout=config.timeout,
|
|
verify_ssl=config.verify_ssl
|
|
)
|
|
|
|
# Perform login
|
|
manager = UserManager(client)
|
|
result = manager.login(email, password)
|
|
|
|
# Store session cookies
|
|
session.set_cookies(result["cookies"])
|
|
|
|
# Store user info if available
|
|
if "user" in result:
|
|
session.set_user(result["user"])
|
|
user_data = result["user"]
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({
|
|
"logged_in": True,
|
|
"user": user_data
|
|
}))
|
|
else:
|
|
from cli_anything.codimd.utils import format_user_info
|
|
click.echo("Login successful!")
|
|
click.echo(format_user_info(user_data))
|
|
else:
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"logged_in": True}))
|
|
else:
|
|
click.echo("Login successful!")
|
|
|
|
|
|
@user.command("logout")
|
|
@handle_error
|
|
def user_logout():
|
|
"""Logout from current session."""
|
|
config = Config()
|
|
session = Session(config.config_dir)
|
|
session.clear()
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"logged_out": True}))
|
|
else:
|
|
click.echo("Logged out successfully")
|
|
|
|
|
|
@user.command("delete")
|
|
@click.option("--token", help="Delete token")
|
|
@click.confirmation_option(prompt="Are you sure you want to delete your account?")
|
|
@handle_error
|
|
def user_delete(token):
|
|
"""Delete current user account."""
|
|
client, session = get_client()
|
|
|
|
if not session.is_authenticated():
|
|
raise PermissionError("Not authenticated.")
|
|
|
|
manager = UserManager(client)
|
|
manager.delete_user(token)
|
|
|
|
# Clear local session
|
|
session.clear()
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"deleted": True}))
|
|
else:
|
|
click.echo("Account deleted successfully")
|
|
|
|
|
|
# Export commands
|
|
@export.command("markdown")
|
|
@click.argument("note_id")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def export_markdown(note_id, output):
|
|
"""Export note as markdown."""
|
|
client, session = get_client()
|
|
manager = ExportManager(client)
|
|
|
|
if output:
|
|
manager.export_markdown(note_id, output)
|
|
click.echo(f"Exported to {output}")
|
|
else:
|
|
content = manager.export_markdown(note_id)
|
|
click.echo(content)
|
|
|
|
|
|
@export.command("pdf")
|
|
@click.argument("note_id")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def export_pdf(note_id, output):
|
|
"""Export note as PDF."""
|
|
client, session = get_client()
|
|
manager = ExportManager(client)
|
|
|
|
if not output:
|
|
output = Path(f"{note_id}.pdf")
|
|
|
|
manager.export_pdf(note_id, output)
|
|
click.echo(f"Exported to {output}")
|
|
|
|
|
|
@export.command("html")
|
|
@click.argument("note_id")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def export_html(note_id, output):
|
|
"""Export note as HTML."""
|
|
client, session = get_client()
|
|
manager = ExportManager(client)
|
|
|
|
if output:
|
|
manager.export_html(note_id, output)
|
|
click.echo(f"Exported to {output}")
|
|
else:
|
|
content = manager.export_html(note_id)
|
|
click.echo(content)
|
|
|
|
|
|
@export.command("slide")
|
|
@click.argument("note_id")
|
|
@click.option("--output", "-o", type=Path, help="Output file")
|
|
@handle_error
|
|
def export_slide(note_id, output):
|
|
"""Export note as reveal.js slides."""
|
|
client, session = get_client()
|
|
manager = ExportManager(client)
|
|
|
|
if output:
|
|
manager.export_slide(note_id, output)
|
|
click.echo(f"Exported to {output}")
|
|
else:
|
|
content = manager.export_slide(note_id)
|
|
click.echo(content)
|
|
|
|
|
|
# Revision commands
|
|
@revision.command("list")
|
|
@click.argument("note_id")
|
|
@handle_error
|
|
def list_revisions(note_id):
|
|
"""List note revisions."""
|
|
client, session = get_client()
|
|
manager = RevisionManager(client)
|
|
revisions = manager.list_revisions(note_id)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({"revisions": revisions}))
|
|
else:
|
|
if not revisions:
|
|
click.echo("No revisions found.")
|
|
else:
|
|
for r in revisions:
|
|
from datetime import datetime
|
|
ts = r.get("time", 0)
|
|
dt = datetime.fromtimestamp(ts / 1000 if ts > 1000000000000 else ts)
|
|
click.echo(f"{dt.isoformat()} - Length: {r.get('length', 0)}")
|
|
|
|
|
|
@revision.command("get")
|
|
@click.argument("note_id")
|
|
@click.argument("timestamp", type=int)
|
|
@handle_error
|
|
def get_revision(note_id, timestamp):
|
|
"""Get note at specific revision time."""
|
|
client, session = get_client()
|
|
manager = RevisionManager(client)
|
|
content = manager.get_revision_at_time(note_id, timestamp)
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success(content))
|
|
else:
|
|
if isinstance(content, dict):
|
|
click.echo(content.get("content", str(content)))
|
|
else:
|
|
click.echo(content)
|
|
|
|
|
|
# Config commands
|
|
@config.command("get")
|
|
@click.argument("key", required=False)
|
|
@handle_error
|
|
def config_get(key):
|
|
"""Get configuration value."""
|
|
config = Config()
|
|
|
|
if key:
|
|
value = getattr(config, key, None)
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({key: str(value)}))
|
|
else:
|
|
click.echo(f"{key}: {value}")
|
|
else:
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({
|
|
"server": config.server,
|
|
"timeout": config.timeout,
|
|
"verify_ssl": config.verify_ssl
|
|
}))
|
|
else:
|
|
click.echo(f"Server: {config.server}")
|
|
click.echo(f"Timeout: {config.timeout}")
|
|
click.echo(f"Verify SSL: {config.verify_ssl}")
|
|
|
|
|
|
@config.command("set")
|
|
@click.argument("key")
|
|
@click.argument("value")
|
|
@handle_error
|
|
def config_set(key, value):
|
|
"""Set configuration value."""
|
|
config = Config()
|
|
|
|
if key == "server":
|
|
config.server = value
|
|
elif key == "timeout":
|
|
config.timeout = int(value)
|
|
elif key == "verify_ssl":
|
|
config.verify_ssl = value.lower() in ("true", "1", "yes")
|
|
else:
|
|
raise ValueError(f"Unknown config key: {key}")
|
|
|
|
if ctx["json_mode"]:
|
|
output_json(format_success({key: value}))
|
|
else:
|
|
click.echo(f"Set {key} = {value}")
|
|
|
|
|
|
# REPL mode
|
|
@cli.command("repl")
|
|
@handle_error
|
|
def repl_mode():
|
|
"""Start interactive REPL mode."""
|
|
import readline
|
|
|
|
config = Config()
|
|
session = Session(config.config_dir)
|
|
client = CodiMDClient(
|
|
server=config.server,
|
|
session_cookies=session.get_cookies()
|
|
)
|
|
|
|
click.echo(f"CodiMD CLI v0.1.0 - REPL Mode")
|
|
click.echo(f"Connected to: {config.server}")
|
|
click.echo("Type 'help' for commands, 'exit' to quit")
|
|
|
|
while True:
|
|
try:
|
|
cmd = click.prompt("codimd", default="", show_default=False).strip()
|
|
|
|
if not cmd:
|
|
continue
|
|
if cmd.lower() in ("exit", "quit"):
|
|
break
|
|
if cmd.lower() == "help":
|
|
click.echo("Available commands:")
|
|
click.echo(" notes - List your notes")
|
|
click.echo(" me - Show user info")
|
|
click.echo(" exit - Exit REPL")
|
|
continue
|
|
|
|
# Handle commands
|
|
if cmd.lower() == "notes":
|
|
manager = NoteManager(client)
|
|
notes = manager.list_my_notes()
|
|
for n in notes[:10]:
|
|
click.echo(f" {n.get('id')} - {n.get('text', 'Untitled')}")
|
|
elif cmd.lower() == "me":
|
|
manager = UserManager(client)
|
|
user = manager.get_me()
|
|
click.echo(f" User: {user.get('profile', {}).get('name', 'N/A')}")
|
|
else:
|
|
click.echo(f"Unknown command: {cmd}")
|
|
|
|
except EOFError:
|
|
break
|
|
except KeyboardInterrupt:
|
|
click.echo("\nUse 'exit' to quit")
|
|
except Exception as e:
|
|
click.echo(f"Error: {e}")
|
|
|
|
click.echo("Goodbye!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|