Jesse.MarkowitzandClaude Opus 4.6
4ccd918eb1
Claude's list endpoint returns conversations with a `name` field rather than `title`, so every Claude row was falling through to "Untitled". Also set no_wrap + ellipsis overflow and tune column widths so the table renders one row per conversation in Windows Command Prompt (80 cols). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1127 lines
45 KiB
Python
1127 lines
45 KiB
Python
"""CLI entry point for ai-chat-exporter."""
|
|
|
|
import importlib.metadata
|
|
import logging
|
|
import platform
|
|
import shutil
|
|
import sys
|
|
import traceback
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from src.cache import Cache, CacheError
|
|
from src.config import ConfigError
|
|
from src.logging_config import setup_logging
|
|
from src.providers.base import ProviderError
|
|
|
|
console = Console()
|
|
err_console = Console(stderr=True)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TOS_NOTICE = """\
|
|
⚠️ IMPORTANT — TERMS OF SERVICE NOTICE
|
|
|
|
This tool accesses ChatGPT and Claude using internal web API endpoints
|
|
that are not officially supported or documented by OpenAI or Anthropic.
|
|
|
|
This approach may conflict with their Terms of Service:
|
|
• OpenAI ToS: https://openai.com/policies/terms-of-use
|
|
• Anthropic ToS: https://www.anthropic.com/legal/consumer-terms
|
|
|
|
By using this tool you accept that:
|
|
- You are using it at your own risk
|
|
- Your account could potentially be suspended for automated access
|
|
- These APIs may break at any time without notice
|
|
- This tool is for personal archival use only, not commercial use
|
|
|
|
Type 'yes' to acknowledge and continue, or Ctrl+C to exit: \
|
|
"""
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# CLI group
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@click.group()
|
|
@click.version_option(
|
|
version=importlib.metadata.version("ai-chat-exporter"),
|
|
prog_name="ai-chat-exporter",
|
|
)
|
|
@click.option("--verbose", "-v", is_flag=True, help="Enable DEBUG output to console.")
|
|
@click.option("--quiet", "-q", is_flag=True, help="Show WARNING and above only.")
|
|
@click.option("--debug", is_flag=True, help="DEBUG + full tracebacks + redacted API bodies.")
|
|
@click.option("--no-log-file", is_flag=True, help="Disable file logging.")
|
|
@click.pass_context
|
|
def cli(ctx: click.Context, verbose: bool, quiet: bool, debug: bool, no_log_file: bool) -> None:
|
|
"""Export ChatGPT and Claude conversations to Markdown for personal archival."""
|
|
ctx.ensure_object(dict)
|
|
|
|
# Determine console log level
|
|
if debug or verbose:
|
|
level = logging.DEBUG
|
|
elif quiet:
|
|
level = logging.WARNING
|
|
else:
|
|
level = logging.INFO
|
|
|
|
# Determine log file path from env (setup_logging handles "none")
|
|
import os
|
|
log_file = os.getenv("LOG_FILE", "./cache/logs/exporter.log")
|
|
|
|
setup_logging(level=level, log_file=log_file, no_log_file=no_log_file)
|
|
|
|
ctx.obj["debug"] = debug
|
|
ctx.obj["verbose"] = verbose
|
|
|
|
# Initialise cache (needed for ToS gate on every command)
|
|
import os
|
|
cache_dir = Path(os.getenv("CACHE_DIR", "./cache")).expanduser()
|
|
try:
|
|
cache = Cache(cache_dir)
|
|
except CacheError as e:
|
|
err_console.print(f"[red]Cache error: {e}[/red]")
|
|
sys.exit(1)
|
|
|
|
ctx.obj["cache"] = cache
|
|
|
|
# ToS gate: must happen before any command executes
|
|
if not cache.is_tos_acknowledged():
|
|
try:
|
|
answer = click.prompt(TOS_NOTICE, default="", show_default=False).strip().lower()
|
|
except (click.Abort, KeyboardInterrupt):
|
|
console.print("\n[yellow]Exiting.[/yellow]")
|
|
sys.exit(0)
|
|
if answer != "yes":
|
|
console.print("[yellow]You must type 'yes' to proceed. Exiting.[/yellow]")
|
|
sys.exit(0)
|
|
cache.acknowledge_tos()
|
|
console.print("[green]Acknowledged. Proceeding.[/green]\n")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# auth command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command()
|
|
@click.pass_context
|
|
def auth(ctx: click.Context) -> None:
|
|
"""Interactive setup wizard for session tokens.
|
|
|
|
Guides you through finding and saving your ChatGPT and Claude session
|
|
tokens. Tokens are never echoed to the terminal.
|
|
|
|
Token lifetimes:
|
|
ChatGPT (__Secure-next-auth.session-token): ~7 days (JWT)
|
|
Claude (sessionKey): ~30 days (opaque string)
|
|
"""
|
|
os_name = platform.system()
|
|
|
|
console.print("\n[bold cyan]AI Chat Exporter — Token Setup Wizard[/bold cyan]\n")
|
|
console.print("Session tokens let this tool access your chat history without your password.")
|
|
console.print("They are stored in browser cookies and have limited lifetimes.\n")
|
|
|
|
# Offer choice of provider to configure
|
|
providers = click.prompt(
|
|
"Which provider(s) do you want to configure? [chatgpt/claude/both]",
|
|
default="both",
|
|
).strip().lower()
|
|
|
|
configure_chatgpt = providers in ("chatgpt", "both")
|
|
configure_claude = providers in ("claude", "both")
|
|
|
|
if configure_chatgpt:
|
|
_auth_chatgpt(os_name)
|
|
|
|
if configure_claude:
|
|
_auth_claude(os_name)
|
|
|
|
console.print("\n[green]Done! Run 'ai-chat-exporter doctor' to verify your setup.[/green]")
|
|
|
|
|
|
def _auth_chatgpt(os_name: str) -> None:
|
|
import jwt as pyjwt
|
|
|
|
console.print("\n[bold]─── ChatGPT ───[/bold]")
|
|
console.print("1. Open [link=https://chatgpt.com]https://chatgpt.com[/link] and log in.")
|
|
if os_name == "Darwin":
|
|
console.print("2. Press [bold]Cmd+Option+I[/bold] to open DevTools → Application tab.")
|
|
else:
|
|
console.print("2. Press [bold]F12[/bold] to open DevTools → Application tab.")
|
|
console.print("3. Expand [bold]Cookies[/bold] → [bold]https://chatgpt.com[/bold]")
|
|
console.print("4. Find [bold]__Secure-next-auth.session-token[/bold] → copy the Value.")
|
|
console.print(" (Token starts with 'eyJ...' — it is a long JWT string)")
|
|
console.print("5. Paste it below (input is hidden).\n")
|
|
|
|
token = click.prompt("ChatGPT session token", hide_input=True, default="", show_default=False).strip()
|
|
if not token:
|
|
console.print("[yellow]Skipped ChatGPT token.[/yellow]")
|
|
return
|
|
|
|
# Validate
|
|
if not token.startswith("eyJ"):
|
|
console.print("[yellow]Warning: token doesn't look like a JWT (expected 'eyJ...').[/yellow]")
|
|
|
|
expiry_str = ""
|
|
try:
|
|
payload = pyjwt.decode(token, options={"verify_signature": False})
|
|
exp = payload.get("exp")
|
|
if exp:
|
|
from datetime import timezone
|
|
expiry = datetime.fromtimestamp(exp, tz=timezone.utc)
|
|
expiry_str = expiry.strftime("%Y-%m-%d %H:%M UTC")
|
|
console.print(f"[green]Token decoded — expires: {expiry_str}[/green]")
|
|
except Exception:
|
|
console.print("[yellow]Could not decode token expiry.[/yellow]")
|
|
|
|
# Live validation — exchange session token for an access token
|
|
_valid = False
|
|
_error: str | None = None
|
|
with console.status("[dim]Validating token with ChatGPT API…[/dim]"):
|
|
try:
|
|
from src.providers.chatgpt import ChatGPTProvider
|
|
_prov = ChatGPTProvider(session_token=token)
|
|
_prov._fetch_access_token()
|
|
_valid = True
|
|
except ProviderError as e:
|
|
_error = str(e.original)
|
|
except Exception as e:
|
|
_error = str(e)
|
|
|
|
if _valid:
|
|
console.print("[green]✓ Token verified — connected to ChatGPT API.[/green]")
|
|
else:
|
|
console.print(f"[red]✗ Token validation failed: {_error}[/red]")
|
|
|
|
_write_token_to_env("CHATGPT_SESSION_TOKEN", token)
|
|
|
|
# --- ChatGPT Projects ---
|
|
console.print("\n[bold]ChatGPT Projects (optional)[/bold]")
|
|
console.print(
|
|
"Project conversations are stored separately and are not included in the\n"
|
|
"default conversation listing. To export them, you need each project's ID.\n"
|
|
)
|
|
console.print("How to find a project ID:")
|
|
console.print(" 1. Open ChatGPT and click into a Project in the left sidebar.")
|
|
console.print(" 2. Look at the browser URL — it will look like:")
|
|
console.print(" [dim]https://chatgpt.com/g/[bold]g-p-68c2b2b3037c8191890036fb4ae3ed9f[/bold]-my-project/project[/dim]")
|
|
console.print(" 3. Copy the part starting with [bold]g-p-[/bold] up to (but not including) the slug.")
|
|
console.print(" Enter multiple IDs separated by commas. Leave blank to skip.\n")
|
|
|
|
project_ids_raw = click.prompt(
|
|
"ChatGPT project IDs (comma-separated, e.g. g-p-xxx,g-p-yyy)",
|
|
default="",
|
|
show_default=False,
|
|
).strip()
|
|
|
|
if project_ids_raw:
|
|
ids = [pid.strip() for pid in project_ids_raw.split(",") if pid.strip()]
|
|
valid = [pid for pid in ids if pid.startswith("g-p-")]
|
|
invalid = [pid for pid in ids if not pid.startswith("g-p-")]
|
|
if invalid:
|
|
console.print(f"[yellow]Warning: skipping IDs that don't start with 'g-p-': {invalid}[/yellow]")
|
|
if valid:
|
|
_write_token_to_env("CHATGPT_PROJECT_IDS", ",".join(valid))
|
|
console.print(f"[green]Saved {len(valid)} project ID(s).[/green]")
|
|
else:
|
|
console.print("[yellow]No valid project IDs — skipping.[/yellow]")
|
|
else:
|
|
console.print("[dim]Skipped project IDs.[/dim]")
|
|
|
|
|
|
def _auth_claude(os_name: str) -> None:
|
|
console.print("\n[bold]─── Claude ───[/bold]")
|
|
console.print("1. Open [link=https://claude.ai]https://claude.ai[/link] and log in.")
|
|
if os_name == "Darwin":
|
|
console.print("2. Press [bold]Cmd+Option+I[/bold] to open DevTools → Application tab.")
|
|
else:
|
|
console.print("2. Press [bold]F12[/bold] to open DevTools → Application tab.")
|
|
console.print("3. Expand [bold]Cookies[/bold] → [bold]https://claude.ai[/bold]")
|
|
console.print("4. Find [bold]sessionKey[/bold] → copy the Value.")
|
|
console.print(" (Note: Claude tokens expire after ~30 days; a 401 error is the only signal.)")
|
|
console.print("5. Paste it below (input is hidden).\n")
|
|
|
|
key = click.prompt("Claude session key", hide_input=True, default="", show_default=False).strip()
|
|
if not key:
|
|
console.print("[yellow]Skipped Claude token.[/yellow]")
|
|
return
|
|
|
|
# Live validation — fetch org ID (the first call any Claude operation makes)
|
|
_valid = False
|
|
_error: str | None = None
|
|
with console.status("[dim]Validating token with Claude API…[/dim]"):
|
|
try:
|
|
from src.providers.claude import ClaudeProvider
|
|
_prov = ClaudeProvider(session_key=key)
|
|
_prov._get_org_id()
|
|
_valid = True
|
|
except ProviderError as e:
|
|
_error = str(e.original)
|
|
except Exception as e:
|
|
_error = str(e)
|
|
|
|
if _valid:
|
|
console.print("[green]✓ Token verified — connected to Claude API.[/green]")
|
|
else:
|
|
console.print(f"[red]✗ Token validation failed: {_error}[/red]")
|
|
|
|
_write_token_to_env("CLAUDE_SESSION_KEY", key)
|
|
|
|
|
|
def _write_token_to_env(key: str, value: str) -> None:
|
|
"""Write or update a key in .env, offering to create the file if it doesn't exist."""
|
|
env_path = Path(".env")
|
|
|
|
if click.confirm(f"Write {key} to .env?", default=True):
|
|
if not env_path.exists():
|
|
# Create from example if available
|
|
example = Path(".env.example")
|
|
if example.exists():
|
|
import shutil as _shutil
|
|
_shutil.copy2(example, env_path)
|
|
console.print("[dim]Created .env from .env.example[/dim]")
|
|
else:
|
|
env_path.touch()
|
|
|
|
lines = env_path.read_text(encoding="utf-8").splitlines(keepends=True)
|
|
updated = False
|
|
new_lines = []
|
|
for line in lines:
|
|
if line.startswith(f"{key}=") or line.startswith(f"{key} ="):
|
|
new_lines.append(f"{key}={value}\n")
|
|
updated = True
|
|
else:
|
|
new_lines.append(line)
|
|
|
|
if not updated:
|
|
new_lines.append(f"\n{key}={value}\n")
|
|
|
|
env_path.write_text("".join(new_lines), encoding="utf-8")
|
|
import os
|
|
os.chmod(env_path, 0o600)
|
|
console.print(f"[green]{key} written to .env (permissions: 600)[/green]")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# doctor command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command()
|
|
@click.pass_context
|
|
def doctor(ctx: click.Context) -> None:
|
|
"""Validate configuration and API connectivity.
|
|
|
|
Checks token presence, format, expiry, directory permissions, disk space,
|
|
and live API reachability. Exits with code 1 if any checks fail.
|
|
"""
|
|
checks = _run_doctor_checks()
|
|
_print_doctor_table(checks)
|
|
|
|
if any(not c["pass"] for c in checks):
|
|
sys.exit(1)
|
|
|
|
|
|
def _run_doctor_checks() -> list[dict]:
|
|
"""Run all doctor checks and return results."""
|
|
import os
|
|
import jwt as pyjwt
|
|
from datetime import timezone
|
|
from dotenv import load_dotenv
|
|
|
|
# Load .env so doctor works without the user having to export vars manually
|
|
load_dotenv(override=False)
|
|
|
|
checks = []
|
|
|
|
def add(name: str, passed: bool, detail: str = "") -> None:
|
|
checks.append({"name": name, "pass": passed, "detail": detail})
|
|
|
|
# Token presence
|
|
chatgpt_token = os.getenv("CHATGPT_SESSION_TOKEN", "").strip() or None
|
|
claude_key = os.getenv("CLAUDE_SESSION_KEY", "").strip() or None
|
|
|
|
add("CHATGPT_SESSION_TOKEN set", bool(chatgpt_token), "" if chatgpt_token else "Set in .env or run 'auth'")
|
|
add("CLAUDE_SESSION_KEY set", bool(claude_key), "" if claude_key else "Set in .env or run 'auth'")
|
|
|
|
# ChatGPT token format + expiry
|
|
if chatgpt_token:
|
|
is_jwt = chatgpt_token.startswith("eyJ")
|
|
add("ChatGPT token is valid JWT", is_jwt, "" if is_jwt else "Expected token starting with 'eyJ'")
|
|
if is_jwt:
|
|
try:
|
|
payload = pyjwt.decode(chatgpt_token, options={"verify_signature": False})
|
|
exp = payload.get("exp")
|
|
if exp:
|
|
expiry = datetime.fromtimestamp(exp, tz=timezone.utc)
|
|
now = datetime.now(tz=timezone.utc)
|
|
delta = expiry - now
|
|
detail = f"Expires {expiry.strftime('%Y-%m-%d %H:%M UTC')} ({delta.days}d)"
|
|
ok = delta.total_seconds() > 0
|
|
add("ChatGPT token not expired", ok, detail)
|
|
if ok and delta.total_seconds() < 86400:
|
|
add("ChatGPT token expiry warning", False, "Expires in < 24h — refresh soon")
|
|
else:
|
|
add("ChatGPT token expiry", False, "JWT has no 'exp' claim")
|
|
except pyjwt.exceptions.DecodeError:
|
|
# JWE (encrypted JWT) — cannot decode without the server key.
|
|
# This is normal for ChatGPT's current token format. Token is present and valid.
|
|
add("ChatGPT token expiry", True, "Encrypted token (JWE) — expiry not decodable client-side")
|
|
|
|
# Claude key
|
|
if claude_key:
|
|
add("Claude session key non-empty", True, "Expiry cannot be decoded (opaque token)")
|
|
|
|
# Directories
|
|
export_dir = Path(os.getenv("EXPORT_DIR", "./exports")).expanduser()
|
|
cache_dir = Path(os.getenv("CACHE_DIR", "./cache")).expanduser()
|
|
|
|
for label, dirpath in [("Export dir writable", export_dir), ("Cache dir writable", cache_dir)]:
|
|
try:
|
|
dirpath.mkdir(parents=True, exist_ok=True)
|
|
test = dirpath / ".doctor_write_test"
|
|
test.touch()
|
|
test.unlink()
|
|
add(label, True, str(dirpath))
|
|
except OSError as e:
|
|
add(label, False, str(e))
|
|
|
|
# Disk space
|
|
try:
|
|
usage = shutil.disk_usage(export_dir if export_dir.exists() else Path("."))
|
|
free_mb = usage.free // (1024 * 1024)
|
|
add("Disk space (≥100MB free)", free_mb >= 100, f"{free_mb}MB free")
|
|
except OSError as e:
|
|
add("Disk space check", False, str(e))
|
|
|
|
# API reachability
|
|
if chatgpt_token:
|
|
try:
|
|
from src.providers.chatgpt import ChatGPTProvider
|
|
p = ChatGPTProvider(chatgpt_token)
|
|
results = p.list_conversations(offset=0, limit=1)
|
|
add("ChatGPT API reachable", True, f"Got {len(results)} result(s)")
|
|
except ProviderError as e:
|
|
add("ChatGPT API reachable", False, str(e.original)[:80])
|
|
except Exception as e:
|
|
add("ChatGPT API reachable", False, str(e)[:80])
|
|
|
|
if claude_key:
|
|
try:
|
|
from src.providers.claude import ClaudeProvider
|
|
p = ClaudeProvider(claude_key)
|
|
results = p.list_conversations(offset=0, limit=1)
|
|
add("Claude API reachable", True, f"Got {len(results)} result(s)")
|
|
except ProviderError as e:
|
|
add("Claude API reachable", False, str(e.original)[:80])
|
|
except Exception as e:
|
|
add("Claude API reachable", False, str(e)[:80])
|
|
|
|
return checks
|
|
|
|
|
|
def _print_doctor_table(checks: list[dict]) -> None:
|
|
table = Table(title="Doctor Check Results", show_header=True)
|
|
table.add_column("Check", style="bold")
|
|
table.add_column("Status", justify="center")
|
|
table.add_column("Detail")
|
|
|
|
for c in checks:
|
|
status = "[green]✓ PASS[/green]" if c["pass"] else "[red]✗ FAIL[/red]"
|
|
table.add_row(c["name"], status, c.get("detail", ""))
|
|
|
|
console.print(table)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# export command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"--provider",
|
|
type=click.Choice(["chatgpt", "claude", "all"], case_sensitive=False),
|
|
default="all",
|
|
show_default=True,
|
|
help="Which provider to export.",
|
|
)
|
|
@click.option(
|
|
"--format",
|
|
"fmt",
|
|
type=click.Choice(["markdown", "json", "both"], case_sensitive=False),
|
|
default="markdown",
|
|
show_default=True,
|
|
help="Output format.",
|
|
)
|
|
@click.option(
|
|
"--output",
|
|
"output_dir",
|
|
default=None,
|
|
type=click.Path(),
|
|
help="Override EXPORT_DIR.",
|
|
)
|
|
@click.option(
|
|
"--since",
|
|
default=None,
|
|
help="Only export conversations updated after this date (YYYY-MM-DD).",
|
|
)
|
|
@click.option(
|
|
"--project",
|
|
"project_filter",
|
|
default=None,
|
|
help=(
|
|
"Only export conversations in a matching project (case-insensitive substring). "
|
|
"Use 'none' for conversations outside any project."
|
|
),
|
|
)
|
|
@click.option("--dry-run", is_flag=True, help="Show what would be exported without writing anything.")
|
|
@click.pass_context
|
|
def export(
|
|
ctx: click.Context,
|
|
provider: str,
|
|
fmt: str,
|
|
output_dir: str | None,
|
|
since: str | None,
|
|
project_filter: str | None,
|
|
dry_run: bool,
|
|
) -> None:
|
|
"""Export new and updated conversations to Markdown or JSON.
|
|
|
|
Every run is automatically resumable — only new or updated conversations
|
|
are exported. Re-running the same command after an interruption will pick
|
|
up exactly where it left off.
|
|
"""
|
|
debug = ctx.obj.get("debug", False)
|
|
cache: Cache = ctx.obj["cache"]
|
|
|
|
# Load config (may raise ConfigError)
|
|
try:
|
|
from src.config import load_config
|
|
cfg = _load_config_or_exit(debug)
|
|
except SystemExit:
|
|
return
|
|
|
|
# First-run: auto-doctor
|
|
if not cache.last_run():
|
|
console.print("[dim]First run — checking configuration…[/dim]")
|
|
checks = _run_doctor_checks()
|
|
_print_doctor_table(checks)
|
|
if any(not c["pass"] for c in checks):
|
|
err_console.print(
|
|
"[red]Doctor checks failed. Fix the issues above before exporting.[/red]"
|
|
)
|
|
sys.exit(1)
|
|
|
|
export_base = Path(output_dir).expanduser() if output_dir else cfg.export_dir
|
|
structure = cfg.output_structure
|
|
|
|
# Resolve since datetime
|
|
since_dt: datetime | None = None
|
|
if since:
|
|
try:
|
|
since_dt = datetime.fromisoformat(since)
|
|
except ValueError:
|
|
err_console.print(f"[red]Invalid --since date: '{since}'. Use YYYY-MM-DD.[/red]")
|
|
sys.exit(1)
|
|
|
|
# Determine which providers to run
|
|
providers_to_run = _resolve_providers(provider, cfg)
|
|
if not providers_to_run:
|
|
err_console.print(
|
|
"[red]No providers configured. Run 'ai-chat-exporter auth' to set up tokens.[/red]"
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Build exporters
|
|
from src.exporters.markdown import MarkdownExporter
|
|
from src.exporters.json_export import JSONExporter
|
|
|
|
md_exporter = MarkdownExporter(export_base, structure) if fmt in ("markdown", "both") else None
|
|
json_exporter = JSONExporter(export_base, structure) if fmt in ("json", "both") else None
|
|
|
|
# Summary counters
|
|
summary: dict[str, dict[str, int]] = {}
|
|
|
|
for prov_name, prov_instance in providers_to_run:
|
|
summary[prov_name] = {"exported": 0, "skipped": 0, "failed": 0}
|
|
|
|
console.print(f"\n[bold cyan][{prov_name.upper()}][/bold cyan] Fetching conversation list…")
|
|
|
|
try:
|
|
all_convs = prov_instance.fetch_all_conversations(since=since_dt)
|
|
except ProviderError as e:
|
|
_handle_provider_error(e, debug)
|
|
summary[prov_name]["failed"] += len(all_convs) if "all_convs" in dir() else 0
|
|
continue
|
|
|
|
if project_filter is not None:
|
|
all_convs = _filter_by_project(all_convs, project_filter)
|
|
console.print(
|
|
f" [dim]--project filter '{project_filter}': {len(all_convs)} matching conversations.[/dim]"
|
|
)
|
|
|
|
to_export = cache.get_new_or_updated(prov_name, all_convs)
|
|
skipped = len(all_convs) - len(to_export)
|
|
summary[prov_name]["skipped"] = skipped
|
|
|
|
if dry_run:
|
|
_print_dry_run_table(prov_name, to_export, prov_instance, export_base, structure, skipped)
|
|
continue
|
|
|
|
if not to_export:
|
|
console.print(f" [dim]{skipped} conversations already up to date.[/dim]")
|
|
continue
|
|
|
|
console.print(f" [dim]{len(to_export)} to export, {skipped} already up to date.[/dim]")
|
|
|
|
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
|
|
|
|
with Progress(
|
|
SpinnerColumn(),
|
|
TextColumn("[progress.description]{task.description}"),
|
|
BarColumn(),
|
|
TaskProgressColumn(),
|
|
console=console,
|
|
) as progress:
|
|
task = progress.add_task(f"Exporting {prov_name}…", total=len(to_export))
|
|
|
|
for raw_conv in to_export:
|
|
conv_id = raw_conv.get("id") or raw_conv.get("uuid", "unknown")
|
|
try:
|
|
full_raw = prov_instance.get_conversation(conv_id)
|
|
# Propagate provider annotations from the listing summary
|
|
# (e.g. _project_name set by ChatGPT project fetching) into
|
|
# the full detail so normalize_conversation can use them.
|
|
for key, val in raw_conv.items():
|
|
if key.startswith("_") and key not in full_raw:
|
|
full_raw[key] = val
|
|
normalized = prov_instance.normalize_conversation(full_raw)
|
|
|
|
exported_path: Path | None = None
|
|
if md_exporter:
|
|
exported_path = md_exporter.export(normalized)
|
|
if json_exporter:
|
|
exported_path = json_exporter.export(normalized)
|
|
|
|
# Write to cache immediately after successful export
|
|
cache.mark_exported(prov_name, conv_id, {
|
|
"title": normalized.get("title", ""),
|
|
"project": normalized.get("project"),
|
|
"updated_at": normalized.get("updated_at", ""),
|
|
"file_path": str(exported_path) if exported_path else "",
|
|
})
|
|
summary[prov_name]["exported"] += 1
|
|
progress.advance(task)
|
|
|
|
except ProviderError as e:
|
|
logger.error("Failed to export conversation %s: %s", conv_id[:8], e)
|
|
summary[prov_name]["failed"] += 1
|
|
progress.advance(task)
|
|
continue
|
|
except OSError as e:
|
|
logger.error("File write failed for conversation %s: %s", conv_id[:8], e)
|
|
summary[prov_name]["failed"] += 1
|
|
progress.advance(task)
|
|
continue
|
|
|
|
if not dry_run:
|
|
_print_export_summary(summary)
|
|
|
|
|
|
def _resolve_providers(provider: str, cfg) -> list[tuple[str, object]]:
|
|
"""Return (name, instance) pairs for configured providers."""
|
|
result = []
|
|
|
|
def try_add(prov_name: str, token: str | None, provider_cls):
|
|
if not token:
|
|
if provider == prov_name or provider == "all":
|
|
logging.getLogger(__name__).warning(
|
|
"[%s] Skipping — token not configured.", prov_name
|
|
)
|
|
return
|
|
try:
|
|
result.append((prov_name, provider_cls(token)))
|
|
except ProviderError as e:
|
|
logging.getLogger(__name__).warning(
|
|
"[%s] Could not initialise provider: %s", prov_name, e
|
|
)
|
|
|
|
from src.providers.chatgpt import ChatGPTProvider
|
|
from src.providers.claude import ClaudeProvider
|
|
|
|
if provider in ("chatgpt", "all"):
|
|
if cfg.chatgpt_session_token:
|
|
try:
|
|
result.append((
|
|
"chatgpt",
|
|
ChatGPTProvider(
|
|
session_token=cfg.chatgpt_session_token,
|
|
project_ids=cfg.chatgpt_project_ids,
|
|
),
|
|
))
|
|
except ProviderError as e:
|
|
logging.getLogger(__name__).warning(
|
|
"[chatgpt] Could not initialise provider: %s", e
|
|
)
|
|
elif provider == "chatgpt" or provider == "all":
|
|
logging.getLogger(__name__).warning("[chatgpt] Skipping — token not configured.")
|
|
if provider in ("claude", "all"):
|
|
try_add("claude", cfg.claude_session_key, ClaudeProvider)
|
|
|
|
return result
|
|
|
|
|
|
def _print_dry_run_table(prov_name, to_export, prov_instance, export_base, structure, skipped):
|
|
from src.utils import generate_filename, build_export_path
|
|
|
|
table = Table(title=f"[DRY RUN] {prov_name.upper()} — Would export {len(to_export)} conversations")
|
|
table.add_column("Title")
|
|
table.add_column("Project")
|
|
table.add_column("Destination")
|
|
table.add_column("Updated")
|
|
|
|
for conv in to_export[:50]: # cap display at 50
|
|
title = conv.get("title") or "Untitled"
|
|
project = conv.get("project_title") or conv.get("project", {})
|
|
if isinstance(project, dict):
|
|
project = project.get("name", "")
|
|
project = project or None
|
|
updated = (conv.get("updated_at") or conv.get("update_time") or "")[:10]
|
|
created = conv.get("created_at") or conv.get("create_time") or ""
|
|
conv_id = conv.get("id") or conv.get("uuid", "")
|
|
filename = generate_filename(title, conv_id, created or "2000-01-01")
|
|
dest = build_export_path(export_base, prov_name, project, created or "2000-01-01", filename, structure)
|
|
table.add_row(title[:50], str(project or "no-project"), str(dest), updated)
|
|
|
|
if len(to_export) > 50:
|
|
table.add_row(f"… and {len(to_export) - 50} more", "", "", "")
|
|
|
|
console.print(table)
|
|
console.print(f" [dim]{skipped} conversations already cached (would be skipped).[/dim]")
|
|
|
|
|
|
def _raw_project_name(conv: dict) -> str | None:
|
|
"""Extract the project name from a raw conversation summary dict.
|
|
|
|
Handles both ChatGPT (annotated _project_name) and Claude (project dict).
|
|
"""
|
|
# ChatGPT: annotated during fetch_all_conversations
|
|
if "_project_name" in conv:
|
|
return conv["_project_name"] or None
|
|
# Claude: project is a dict with a 'name' key, or a plain string
|
|
project = conv.get("project")
|
|
if isinstance(project, dict):
|
|
return project.get("name") or None
|
|
if isinstance(project, str):
|
|
return project or None
|
|
return None
|
|
|
|
|
|
def _filter_by_project(convs: list[dict], project_filter: str) -> list[dict]:
|
|
"""Filter conversations by project name.
|
|
|
|
project_filter='none' → keep only conversations with no project.
|
|
Otherwise → case-insensitive substring match on the project name.
|
|
"""
|
|
want_none = project_filter.lower() == "none"
|
|
needle = project_filter.lower()
|
|
|
|
result = []
|
|
for conv in convs:
|
|
name = _raw_project_name(conv)
|
|
if want_none:
|
|
if name is None:
|
|
result.append(conv)
|
|
else:
|
|
if name and needle in name.lower():
|
|
result.append(conv)
|
|
return result
|
|
|
|
|
|
def _print_export_summary(summary: dict[str, dict[str, int]]) -> None:
|
|
table = Table(title="Export Summary")
|
|
table.add_column("Provider", style="bold")
|
|
table.add_column("Exported", justify="right")
|
|
table.add_column("Skipped", justify="right")
|
|
table.add_column("Failed", justify="right")
|
|
|
|
for prov, counts in summary.items():
|
|
table.add_row(
|
|
prov.capitalize(),
|
|
str(counts["exported"]),
|
|
str(counts["skipped"]),
|
|
f"[red]{counts['failed']}[/red]" if counts["failed"] else "0",
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# list command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command(name="list")
|
|
@click.option(
|
|
"--provider",
|
|
type=click.Choice(["chatgpt", "claude", "all"], case_sensitive=False),
|
|
default="all",
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
"--project",
|
|
"project_filter",
|
|
default=None,
|
|
help=(
|
|
"Only list conversations in a matching project (case-insensitive substring). "
|
|
"Use 'none' for conversations outside any project."
|
|
),
|
|
)
|
|
@click.pass_context
|
|
def list_conversations(ctx: click.Context, provider: str, project_filter: str | None) -> None:
|
|
"""List conversations without exporting them."""
|
|
debug = ctx.obj.get("debug", False)
|
|
cfg = _load_config_or_exit(debug)
|
|
providers_to_run = _resolve_providers(provider, cfg)
|
|
|
|
for prov_name, prov_instance in providers_to_run:
|
|
console.print(f"\n[bold cyan][{prov_name.upper()}][/bold cyan]")
|
|
try:
|
|
all_convs = prov_instance.fetch_all_conversations()
|
|
except ProviderError as e:
|
|
_handle_provider_error(e, debug)
|
|
continue
|
|
|
|
if project_filter is not None:
|
|
all_convs = _filter_by_project(all_convs, project_filter)
|
|
|
|
# no_wrap + overflow="ellipsis" prevents Rich from wrapping cells to
|
|
# multiple lines on narrow terminals (e.g. Windows Command Prompt),
|
|
# which can otherwise make the output look garbled. Widths are tuned
|
|
# to fit within an 80-column terminal.
|
|
# Total width budget for 80-column terminals:
|
|
# borders (5) + padding (4 cols * 2) = 13 chars of overhead
|
|
# remaining 67 chars split: 34 title + 15 project + 10 date + 8 id
|
|
table = Table(show_lines=False, expand=False, padding=(0, 1))
|
|
table.add_column("Title", no_wrap=True, overflow="ellipsis", max_width=34)
|
|
table.add_column("Project", no_wrap=True, overflow="ellipsis", max_width=15)
|
|
table.add_column("Updated", no_wrap=True, min_width=10)
|
|
table.add_column("ID", no_wrap=True, min_width=8)
|
|
|
|
for conv in all_convs:
|
|
# ChatGPT uses "title"; Claude uses "name".
|
|
title = conv.get("title") or conv.get("name") or "Untitled"
|
|
project = _raw_project_name(conv) or ""
|
|
updated = (conv.get("updated_at") or conv.get("update_time") or "")[:10]
|
|
conv_id = (conv.get("id") or conv.get("uuid") or "")[:8]
|
|
table.add_row(title, project, updated, conv_id)
|
|
|
|
console.print(table)
|
|
console.print(f"Total: {len(all_convs)} conversations")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# cache command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--show", is_flag=True, help="Show cache statistics.")
|
|
@click.option("--clear", is_flag=True, help="Clear cached entries.")
|
|
@click.option(
|
|
"--provider",
|
|
type=click.Choice(["chatgpt", "claude", "all"], case_sensitive=False),
|
|
default="all",
|
|
help="Provider to target (used with --clear).",
|
|
)
|
|
@click.pass_context
|
|
def cache(ctx: click.Context, show: bool, clear: bool, provider: str) -> None:
|
|
"""Manage the local export cache (sync manifest)."""
|
|
cache_obj: Cache = ctx.obj["cache"]
|
|
|
|
if show:
|
|
stats = cache_obj.stats()
|
|
table = Table(title="Cache Statistics")
|
|
table.add_column("Provider", style="bold")
|
|
table.add_column("Cached Conversations", justify="right")
|
|
for prov, count in stats.items():
|
|
table.add_row(prov.capitalize(), str(count))
|
|
last = cache_obj.last_run()
|
|
console.print(table)
|
|
console.print(f"Last run: {last or 'never'}")
|
|
|
|
if clear:
|
|
prov_arg = None if provider == "all" else provider
|
|
if click.confirm(f"Clear cache for {'all providers' if prov_arg is None else prov_arg}?"):
|
|
cache_obj.clear(prov_arg)
|
|
console.print("[green]Cache cleared.[/green]")
|
|
|
|
if not show and not clear:
|
|
console.print("Specify --show or --clear. Use --help for options.")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# joplin command
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"--provider",
|
|
type=click.Choice(["chatgpt", "claude", "all"], case_sensitive=False),
|
|
default="all",
|
|
show_default=True,
|
|
help="Which provider's conversations to sync to Joplin.",
|
|
)
|
|
@click.option(
|
|
"--project",
|
|
"project_filter",
|
|
default=None,
|
|
help=(
|
|
"Only sync conversations in a matching project (case-insensitive substring). "
|
|
"Use 'none' for conversations outside any project."
|
|
),
|
|
)
|
|
@click.option("--dry-run", is_flag=True, help="Show what would be synced without sending anything to Joplin.")
|
|
@click.pass_context
|
|
def joplin(ctx: click.Context, provider: str, project_filter: str | None, dry_run: bool) -> None:
|
|
"""Sync exported conversations to Joplin as notes.
|
|
|
|
Reads the local export cache and pushes exported Markdown files to Joplin
|
|
via its local REST API. Requires Joplin desktop to be running with the
|
|
Web Clipper service enabled.
|
|
|
|
Notebooks are created automatically based on provider and project:
|
|
exports/chatgpt/my-project/ → "ChatGPT - My Project" notebook
|
|
exports/claude/no-project/ → "Claude - No Project" notebook
|
|
|
|
Re-running is safe: notes are updated (not duplicated) on subsequent runs.
|
|
|
|
Setup:
|
|
1. Open Joplin desktop.
|
|
2. Go to Tools → Options → Web Clipper.
|
|
3. Enable the Web Clipper service.
|
|
4. Copy the Authorization token.
|
|
5. Set JOPLIN_API_TOKEN=<token> in your .env file.
|
|
"""
|
|
debug = ctx.obj.get("debug", False)
|
|
cache_obj: Cache = ctx.obj["cache"]
|
|
|
|
cfg = _load_config_or_exit(debug)
|
|
|
|
if not cfg.joplin_api_token:
|
|
err_console.print(
|
|
"[red]JOPLIN_API_TOKEN is not set.[/red]\n"
|
|
" 1. Open Joplin → Tools → Options → Web Clipper.\n"
|
|
" 2. Enable the Web Clipper service.\n"
|
|
" 3. Copy the Authorization token.\n"
|
|
" 4. Add [bold]JOPLIN_API_TOKEN=<token>[/bold] to your .env file."
|
|
)
|
|
sys.exit(1)
|
|
|
|
from src.joplin import JoplinClient, JoplinError, notebook_title
|
|
|
|
client = JoplinClient(cfg.joplin_api_url, cfg.joplin_api_token)
|
|
|
|
if not dry_run:
|
|
console.print(f"[dim]Connecting to Joplin at {cfg.joplin_api_url}…[/dim]")
|
|
try:
|
|
if not client.ping():
|
|
err_console.print(
|
|
"[red]Joplin is not responding.[/red] "
|
|
"Make sure Joplin desktop is open and Web Clipper is enabled."
|
|
)
|
|
sys.exit(1)
|
|
# Ping succeeded but doesn't validate the token — check auth separately
|
|
client.validate_token()
|
|
except JoplinError as e:
|
|
err_console.print(f"[red]Joplin connection error:[/red] {e}")
|
|
sys.exit(1)
|
|
console.print("[green]Joplin connected and token validated.[/green]")
|
|
|
|
# Determine which providers to process
|
|
providers_to_sync: list[str] = []
|
|
if provider in ("chatgpt", "all"):
|
|
providers_to_sync.append("chatgpt")
|
|
if provider in ("claude", "all"):
|
|
providers_to_sync.append("claude")
|
|
|
|
summary: dict[str, dict[str, int]] = {}
|
|
|
|
for prov_name in providers_to_sync:
|
|
summary[prov_name] = {"created": 0, "updated": 0, "skipped": 0, "failed": 0}
|
|
|
|
pending = cache_obj.get_joplin_pending(prov_name)
|
|
logger.debug("[joplin] %s: %d pending before filter", prov_name, len(pending))
|
|
|
|
# Apply --project filter against the cached entry's project field
|
|
if project_filter is not None:
|
|
want_none = project_filter.lower() == "none"
|
|
needle = project_filter.lower()
|
|
filtered = []
|
|
for conv_id, entry in pending:
|
|
proj = entry.get("project") or None
|
|
if want_none:
|
|
if proj is None or proj == "no-project":
|
|
filtered.append((conv_id, entry))
|
|
else:
|
|
if proj and needle in proj.lower():
|
|
filtered.append((conv_id, entry))
|
|
logger.debug(
|
|
"[joplin] %s: --project %r filtered %d → %d",
|
|
prov_name, project_filter, len(pending), len(filtered),
|
|
)
|
|
pending = filtered
|
|
|
|
if not pending:
|
|
console.print(f"\n[bold cyan][{prov_name.upper()}][/bold cyan] All up to date — nothing to sync.")
|
|
continue
|
|
|
|
console.print(
|
|
f"\n[bold cyan][{prov_name.upper()}][/bold cyan] "
|
|
f"{len(pending)} conversation(s) to sync to Joplin."
|
|
)
|
|
|
|
if dry_run:
|
|
_print_joplin_dry_run_table(prov_name, pending)
|
|
continue
|
|
|
|
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
|
|
|
|
with Progress(
|
|
SpinnerColumn(),
|
|
TextColumn("[progress.description]{task.description}"),
|
|
BarColumn(),
|
|
TaskProgressColumn(),
|
|
console=console,
|
|
) as progress:
|
|
task = progress.add_task(f"Syncing {prov_name}…", total=len(pending))
|
|
|
|
for conv_id, entry in pending:
|
|
file_path = entry.get("file_path", "")
|
|
title = entry.get("title") or "Untitled"
|
|
project = entry.get("project") or None
|
|
existing_note_id = entry.get("joplin_note_id")
|
|
action = "update" if existing_note_id else "create"
|
|
|
|
logger.debug(
|
|
"[joplin] %s %s/%s: %s (file=%s)",
|
|
action, prov_name, conv_id[:8], title[:60], file_path,
|
|
)
|
|
|
|
try:
|
|
# Read the exported Markdown file
|
|
body = Path(file_path).read_text(encoding="utf-8")
|
|
logger.debug("[joplin] Read %d chars from %s", len(body), file_path)
|
|
|
|
# Get or create the notebook
|
|
nb_title = notebook_title(prov_name, project)
|
|
notebook_id = client.get_or_create_notebook(nb_title)
|
|
|
|
if existing_note_id:
|
|
client.update_note(existing_note_id, title, body)
|
|
cache_obj.mark_joplin_synced(prov_name, conv_id, existing_note_id)
|
|
summary[prov_name]["updated"] += 1
|
|
else:
|
|
note_id = client.create_note(title, body, notebook_id)
|
|
cache_obj.mark_joplin_synced(prov_name, conv_id, note_id)
|
|
summary[prov_name]["created"] += 1
|
|
|
|
except FileNotFoundError:
|
|
logger.warning(
|
|
"[joplin] Skipping %s/%s — exported file not found: %s",
|
|
prov_name, conv_id[:8], file_path,
|
|
)
|
|
summary[prov_name]["skipped"] += 1
|
|
except JoplinError as e:
|
|
logger.error(
|
|
"[joplin] Failed to %s note for %s/%s: %s",
|
|
action, prov_name, conv_id[:8], e,
|
|
)
|
|
summary[prov_name]["failed"] += 1
|
|
except OSError as e:
|
|
logger.error(
|
|
"[joplin] File read error for %s/%s (%s): %s",
|
|
prov_name, conv_id[:8], file_path, e,
|
|
)
|
|
summary[prov_name]["failed"] += 1
|
|
finally:
|
|
progress.advance(task)
|
|
|
|
if not dry_run:
|
|
_print_joplin_summary(summary)
|
|
|
|
|
|
def _print_joplin_dry_run_table(prov_name: str, pending: list[tuple[str, dict]]) -> None:
|
|
from src.joplin import notebook_title
|
|
|
|
table = Table(title=f"[DRY RUN] {prov_name.upper()} — Would sync {len(pending)} conversation(s)")
|
|
table.add_column("Title")
|
|
table.add_column("Project")
|
|
table.add_column("Notebook")
|
|
table.add_column("Action")
|
|
|
|
for conv_id, entry in pending[:50]:
|
|
title = entry.get("title") or "Untitled"
|
|
project = entry.get("project") or "no-project"
|
|
nb = notebook_title(prov_name, entry.get("project"))
|
|
action = "update" if entry.get("joplin_note_id") else "create"
|
|
table.add_row(title[:50], project[:30], nb, action)
|
|
|
|
if len(pending) > 50:
|
|
table.add_row(f"… and {len(pending) - 50} more", "", "", "")
|
|
|
|
console.print(table)
|
|
|
|
|
|
def _print_joplin_summary(summary: dict[str, dict[str, int]]) -> None:
|
|
table = Table(title="Joplin Sync Summary")
|
|
table.add_column("Provider", style="bold")
|
|
table.add_column("Created", justify="right")
|
|
table.add_column("Updated", justify="right")
|
|
table.add_column("Skipped", justify="right")
|
|
table.add_column("Failed", justify="right")
|
|
|
|
for prov, counts in summary.items():
|
|
table.add_row(
|
|
prov.capitalize(),
|
|
str(counts["created"]),
|
|
str(counts["updated"]),
|
|
str(counts["skipped"]),
|
|
f"[red]{counts['failed']}[/red]" if counts["failed"] else "0",
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Helpers
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _load_config_or_exit(debug: bool):
|
|
"""Load config, printing a clean error on failure."""
|
|
from src.config import load_config
|
|
try:
|
|
return load_config()
|
|
except ConfigError as e:
|
|
err_console.print(f"[red]Configuration error:[/red] {e}")
|
|
if debug:
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
def _handle_provider_error(e: ProviderError, debug: bool) -> None:
|
|
err_console.print(f"[red]Provider error ({e.provider_name}/{e.operation}):[/red] {e.original}")
|
|
if debug:
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|