from __future__ import annotations from dataclasses import asdict from pathlib import Path import json import re from typing import Any from .schema import Conversation, Message from .utils import extract_title_from_markdown, markdown_body, read_text, stable_id CHATGPT_SHARE_RE = re.compile(r"https?://(?:chatgpt\.com|chat\.openai\.com)/share/[A-Za-z0-9_-]+", re.I) def _page_text_to_message(path: Path, source_root: Path, text: str) -> Conversation: rel_path = path.relative_to(source_root).as_posix() title = extract_title_from_markdown(text, path.stem) shares = CHATGPT_SHARE_RE.findall(text) convo_id = stable_id("chatgpt", rel_path) message = Message( message_id=f"{convo_id}:1", role="assistant", content=markdown_body(text), metadata={"path": rel_path, "shares": shares, "kind": "notion_page"}, ) return Conversation( source="chatgpt", conversation_id=convo_id, title=title, raw_path=rel_path, messages=[message], metadata={"kind": "notion_page", "shares": shares}, ) def import_chatgpt(raw_dir: Path) -> list[Conversation]: conversations: list[Conversation] = [] for path in sorted(raw_dir.rglob("*")): if not path.is_file(): continue if path.name.startswith("."): continue suffix = path.suffix.lower() if suffix not in {".md", ".txt", ".csv"}: continue text = read_text(path) conversations.append(_page_text_to_message(path, raw_dir, text)) return conversations def _flatten_claude_message(msg: dict[str, Any]) -> str: blocks = msg.get("content") or [] parts: list[str] = [] if isinstance(blocks, list): for block in blocks: if isinstance(block, dict): if block.get("type") == "text" and block.get("text"): parts.append(str(block.get("text"))) elif block.get("text"): parts.append(str(block.get("text"))) elif block is not None: parts.append(str(block)) if not parts and msg.get("text"): parts.append(str(msg.get("text"))) return "\\n\\n".join(parts).strip() def import_claude(json_path: Path) -> list[Conversation]: root = json.loads(json_path.read_text(encoding="utf-8")) conversations: list[Conversation] = [] for item in root: convo_id = item.get("uuid") or stable_id("claude", item.get("name", ""), item.get("created_at", "")) messages: list[Message] = [] for idx, msg in enumerate(item.get("chat_messages") or [], start=1): content = _flatten_claude_message(msg) sender = (msg.get("sender") or "").lower() role = "user" if sender in {"human", "user"} else "assistant" if sender in {"assistant", "claude"} else "system" messages.append(Message( message_id=msg.get("uuid") or f"{convo_id}:{idx}", role=role, content=content, created_at=msg.get("created_at"), metadata={ "parent_message_uuid": msg.get("parent_message_uuid"), "sender": msg.get("sender"), "attachments": msg.get("attachments") or [], "files": msg.get("files") or [], }, )) conversations.append(Conversation( source="claude", conversation_id=convo_id, title=item.get("name") or item.get("summary", "")[:80] or convo_id, created_at=item.get("created_at"), updated_at=item.get("updated_at"), raw_path=str(json_path.relative_to(json_path.parents[1]).as_posix()), messages=messages, metadata={"kind": "claude_conversation", "account": item.get("account"), "summary": item.get("summary")}, )) return conversations def _claude_rel(path: Path, raw_dir: Path) -> str: return path.relative_to(raw_dir).as_posix() def import_claude_memories(raw_dir: Path) -> list[Conversation]: path = raw_dir / "memories.json" if not path.exists(): return [] root = json.loads(path.read_text(encoding="utf-8")) conversations: list[Conversation] = [] for idx, item in enumerate(root if isinstance(root, list) else [root], start=1): content = item.get("conversations_memory") if isinstance(item, dict) else str(item) if not content: continue convo_id = stable_id("claude_memory", str(idx), content[:200]) conversations.append(Conversation( source="claude_memory", conversation_id=convo_id, title="Claude exported memory", raw_path=_claude_rel(path, raw_dir), messages=[Message( message_id=f"{convo_id}:1", role="system", content=str(content), metadata={"kind": "claude_exported_memory"}, )], tags=["memory", "claude"], metadata={"kind": "claude_exported_memory", "sensitivity": "sensitive_review_required"}, )) return conversations def import_claude_projects(raw_dir: Path) -> list[Conversation]: projects_dir = raw_dir / "projects" if not projects_dir.exists(): return [] conversations: list[Conversation] = [] for path in sorted(projects_dir.glob("*.json")): item = json.loads(path.read_text(encoding="utf-8")) project_name = item.get("name") or path.stem description = item.get("description") or "" docs = item.get("docs") or [] if description: convo_id = stable_id("claude_project", path.name, "description") conversations.append(Conversation( source="claude_project", conversation_id=convo_id, title=f"Claude project: {project_name}", created_at=item.get("created_at"), updated_at=item.get("updated_at"), raw_path=_claude_rel(path, raw_dir), messages=[Message( message_id=f"{convo_id}:1", role="system", content=description, metadata={"kind": "claude_project_description", "project_uuid": item.get("uuid")}, )], tags=["project", project_name], metadata={"kind": "claude_project", "project_name": project_name, "project_uuid": item.get("uuid")}, )) for doc_idx, doc in enumerate(docs, start=1): content = doc.get("content") or "" if not content.strip(): continue filename = doc.get("filename") or f"doc-{doc_idx}.md" convo_id = stable_id("claude_project_doc", path.name, doc.get("uuid", ""), filename) conversations.append(Conversation( source="claude_project", conversation_id=convo_id, title=filename, created_at=item.get("created_at"), updated_at=item.get("updated_at"), raw_path=f"{_claude_rel(path, raw_dir)}#docs/{doc_idx}:{filename}", messages=[Message( message_id=f"{convo_id}:1", role="user", content=content, metadata={"kind": "claude_project_doc", "project_uuid": item.get("uuid"), "doc_uuid": doc.get("uuid"), "filename": filename}, )], tags=["project", project_name], metadata={"kind": "claude_project_doc", "project_name": project_name, "project_uuid": item.get("uuid"), "doc_uuid": doc.get("uuid"), "filename": filename}, )) return conversations def import_claude_export(raw_dir: Path) -> list[Conversation]: conversations = import_claude(raw_dir / "conversations.json") conversations.extend(import_claude_memories(raw_dir)) conversations.extend(import_claude_projects(raw_dir)) return conversations def conversations_to_records(conversations: list[Conversation]) -> list[dict[str, Any]]: return [c.to_dict() for c in conversations]