monkeysucker/monkeysucker.py

#!/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())