Julian Carlson
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def parse_text(path):
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
return path.stem, text, {}
|
|
|
|
|
|
def parse_markdown(path):
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
title = path.stem
|
|
for line in text.split("\n"):
|
|
if line.startswith("# "):
|
|
title = line.lstrip("#").strip()
|
|
break
|
|
return title, text, {}
|
|
|
|
|
|
def parse_pdf(path):
|
|
from pypdf import PdfReader
|
|
reader = PdfReader(str(path))
|
|
pages = [p.extract_text() or "" for p in reader.pages]
|
|
text = "\n\n".join(pages)
|
|
title = path.stem
|
|
if reader.metadata and reader.metadata.title:
|
|
title = reader.metadata.title
|
|
return title, text, {"page_count": len(pages)}
|
|
|
|
|
|
def parse_claude_export(path):
|
|
data = json.loads(path.read_text(encoding="utf-8", errors="replace"))
|
|
|
|
if isinstance(data, dict) and ("name" in data or "title" in data):
|
|
return _format_conversation(data, path.stem)
|
|
elif isinstance(data, list):
|
|
all_text = []
|
|
for conv in data:
|
|
_, text, _ = _format_conversation(conv, "")
|
|
all_text.append(text)
|
|
return path.stem, "\n\n---\n\n".join(all_text), {"conversation_count": len(data)}
|
|
else:
|
|
return path.stem, json.dumps(data, indent=2), {}
|
|
|
|
|
|
def _format_conversation(conv, default_title):
|
|
title = conv.get("name") or conv.get("title") or default_title
|
|
messages = conv.get("messages", []) or conv.get("chat_messages", [])
|
|
|
|
parts = [f"# {title}\n"]
|
|
for msg in messages:
|
|
role_raw = msg.get("role") or msg.get("sender") or "unknown"
|
|
# Normalize to USER/ASSISTANT for consistency with OpenClaw transcripts
|
|
role = "USER" if role_raw == "human" else ("ASSISTANT" if role_raw == "assistant" else role_raw.upper())
|
|
content = msg.get("content") or msg.get("text") or ""
|
|
if isinstance(content, list):
|
|
content = "\n".join(
|
|
block.get("text", "") for block in content
|
|
if isinstance(block, dict) and block.get("type") == "text"
|
|
)
|
|
if not content.strip():
|
|
continue # skip empty blocks (tool_use, tool_result, etc.)
|
|
parts.append(f"\n**{role}**:\n\n{content}\n")
|
|
|
|
return title, "\n".join(parts), {
|
|
"message_count": len(messages),
|
|
"format": "claude_export",
|
|
}
|
|
|
|
|
|
PARSERS = {
|
|
".txt": parse_text,
|
|
".md": parse_markdown,
|
|
".markdown": parse_markdown,
|
|
".pdf": parse_pdf,
|
|
".json": parse_claude_export,
|
|
}
|
|
|
|
|
|
def parse_file(path):
|
|
suffix = path.suffix.lower()
|
|
parser = PARSERS.get(suffix)
|
|
if not parser:
|
|
raise ValueError(f"No parser for {suffix}")
|
|
return parser(path)
|