init

monkeysucker.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""Convert WriteMonkey 3 sheets to markdown files with snippets."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import re
+from collections import defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, Iterable, List, Tuple
+
+
+def _read_records(path: Path) -> Iterable[Tuple[dict, dict]]:
+    with path.open("r", encoding="utf-8", errors="replace") as f:
+        while True:
+            header = f.readline()
+            if not header:
+                break
+            meta = f.readline()
+            data = f.readline()
+            if not meta or not data:
+                break
+            yield json.loads(meta), json.loads(data)
+
+
+def _slugify(text: str, max_len: int = 80) -> str:
+    text = text.strip().lower()
+    text = re.sub(r"[^a-z0-9\- _]+", "", text)
+    text = re.sub(r"[\s_]+", "_", text)
+    text = re.sub(r"-+", "-", text)
+    text = text.strip("_-")
+    return text[:max_len] or "untitled"
+
+
+def _date_str_utc(ms: int | float | None) -> str:
+    if not ms:
+        return "1970-01-01"
+    dt = datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
+    return dt.strftime("%Y-%m-%d")
+
+
+def _filename_date_utc(ms: int | float | None) -> str:
+    if not ms:
+        return "19700101"
+    dt = datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
+    return dt.strftime("%Y%m%d")
+
+
+def _yaml_escape(text: str) -> str:
+    text = text.replace("\\", "\\\\").replace('"', '\\"')
+    return f'"{text}"'
+
+
+def _write_text(path: Path, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8", errors="replace") as f:
+        f.write(content)
+
+
+def _set_mtime(path: Path, ms: int | float | None) -> None:
+    if not ms:
+        return
+    ts = ms / 1000
+    os.utime(path, (ts, ts))
+
+
+def _frontmatter(doc_id: int, title: str, created: str, modified: str, tags: List[str]) -> str:
+    lines = [
+        "---",
+        f"id: {doc_id}",
+        f"title: {_yaml_escape(title)}",
+        f"created: {created}",
+        f"modified: {modified}",
+    ]
+    if tags:
+        lines.append("tags:")
+        for tag in tags:
+            lines.append(f"  - {_yaml_escape(str(tag))}")
+    lines.append("---")
+    return "\n".join(lines) + "\n\n"
+
+
+def _next_available_name(folder: Path, base: str, ext: str, used: Dict[Path, set]) -> str:
+    used_names = used.setdefault(folder, set())
+    candidate = f"{base}{ext}"
+    if candidate not in used_names and not (folder / candidate).exists():
+        used_names.add(candidate)
+        return candidate
+    index = 1
+    while True:
+        candidate = f"{base}_{index}{ext}"
+        if candidate not in used_names and not (folder / candidate).exists():
+            used_names.add(candidate)
+            return candidate
+        index += 1
+
+
+def _pid_to_int(pid) -> int | None:
+    if pid is None:
+        return None
+    if isinstance(pid, int):
+        return pid
+    if isinstance(pid, str) and pid.isdigit():
+        return int(pid)
+    return None
+
+
+def _load_projects(input_dir: Path) -> Dict[int, str]:
+    projects_path = input_dir / "writemonkey3_sheets_projects"
+    if not projects_path.exists():
+        return {}
+    projects: Dict[int, str] = {}
+    for meta, data in _read_records(projects_path):
+        proj_id = meta.get("_id")
+        name = data.get("nam", "")
+        if isinstance(proj_id, int):
+            projects[proj_id] = name
+    return projects
+
+
+def _load_snippets(input_dir: Path) -> Dict[int, List[Tuple[int, str]]]:
+    repo_path = input_dir / "writemonkey3_sheets_repository"
+    if not repo_path.exists():
+        return {}
+    snippets: Dict[int, List[Tuple[int, str]]] = defaultdict(list)
+    for _, data in _read_records(repo_path):
+        if data.get("typ") != "snippet":
+            continue
+        did = data.get("did")
+        if not isinstance(did, int):
+            continue
+        order = data.get("ord")
+        if not isinstance(order, int):
+            order = 0
+        txt = data.get("txt", "")
+        snippets[did].append((order, txt))
+    for did in list(snippets.keys()):
+        snippets[did].sort(key=lambda item: item[0])
+    return snippets
+
+
+def _latest_docs(path: Path) -> List[Tuple[dict, dict]]:
+    latest: Dict[int, Tuple[int, dict, dict]] = {}
+    for meta, data in _read_records(path):
+        doc_id = meta.get("_id")
+        if not isinstance(doc_id, int):
+            continue
+        dtm = data.get("dtm")
+        dtc = data.get("dtc")
+        stamp = dtm if isinstance(dtm, (int, float)) else dtc if isinstance(dtc, (int, float)) else 0
+        current = latest.get(doc_id)
+        if current is None or stamp > current[0]:
+            latest[doc_id] = (int(stamp), meta, data)
+    return [(meta, data) for _, meta, data in latest.values()]
+
+
+def export_documents(input_dir: Path, output_dir: Path) -> None:
+    sheets_path = input_dir / "writemonkey3_sheets"
+    if not sheets_path.exists():
+        raise SystemExit(f"Missing file: {sheets_path}")
+
+    projects = _load_projects(input_dir)
+    snippets = _load_snippets(input_dir)
+    used_names: Dict[Path, set] = {}
+
+    for meta, data in _latest_docs(sheets_path):
+        doc_id = meta.get("_id")
+        if not isinstance(doc_id, int):
+            continue
+
+        title = data.get("nam", "")
+        dtc = data.get("dtc")
+        dtm = data.get("dtm")
+        created_date = _date_str_utc(dtc)
+        modified_date = _date_str_utc(dtm or dtc)
+        filename_date = _filename_date_utc(dtc)
+        slug = _slugify(title)
+        base = f"{filename_date}_{slug}"
+
+        pid = _pid_to_int(data.get("pid"))
+        if pid is not None and pid in projects:
+            project_folder = _slugify(projects[pid])
+            folder = output_dir / project_folder
+        else:
+            folder = output_dir
+
+        filename = _next_available_name(folder, base, ".md", used_names)
+        path = folder / filename
+
+        tags = []
+        met = data.get("met")
+        if isinstance(met, dict):
+            raw_tags = met.get("tags")
+            if isinstance(raw_tags, list) and raw_tags:
+                tags = [str(tag) for tag in raw_tags]
+
+        frontmatter = _frontmatter(doc_id, title, created_date, modified_date, tags)
+        body = data.get("txt", "")
+        _write_text(path, frontmatter + body)
+        _set_mtime(path, dtm or dtc)
+
+        doc_snippets = snippets.get(doc_id)
+        if doc_snippets:
+            snippet_name = path.with_name(path.stem + "_snippets" + path.suffix)
+            snippets_text = "\n---\n".join(text for _, text in doc_snippets)
+            _write_text(snippet_name, frontmatter + snippets_text)
+            _set_mtime(snippet_name, dtm or dtc)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Convert WriteMonkey 3 sheets to markdown files.")
+    parser.add_argument("input_dir", help="Directory containing WriteMonkey sheets")
+    parser.add_argument("output_dir", help="Output directory")
+    args = parser.parse_args()
+
+    input_dir = Path(args.input_dir)
+    output_dir = Path(args.output_dir)
+    if output_dir.exists():
+        shutil.rmtree(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    export_documents(input_dir, output_dir)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())