1214 lines
39 KiB
Python
1214 lines
39 KiB
Python
"""Emit Threadwell archive notes from a thread_data.json dump. No fetch, no lift."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
try:
|
|
from twitter.frozen import frozen_match, load_frozen_ids
|
|
from twitter.media_manifest import (
|
|
_from_wire_dict,
|
|
_item_to_wire,
|
|
atomic_write_json,
|
|
find_location,
|
|
item_key,
|
|
merge_manifest_items,
|
|
new_original_item,
|
|
selected_url,
|
|
)
|
|
from twitter.media_refs import remote_markup
|
|
from twitter.models import MediaItem, ThreadData, load_thread
|
|
from twitter.paths import FROZEN, SCRATCH
|
|
from twitter.render import render_branch, render_spine, title_text
|
|
from twitter.slug import branch_file_name, date_prefix, thread_dir_name
|
|
from twitter.tree import (
|
|
by_id,
|
|
branch_roots,
|
|
children_map,
|
|
descendants,
|
|
spine_from_tip,
|
|
spine_ids,
|
|
)
|
|
except ImportError: # pragma: no cover - script-mode import
|
|
from frozen import frozen_match, load_frozen_ids
|
|
from media_manifest import (
|
|
_from_wire_dict,
|
|
_item_to_wire,
|
|
atomic_write_json,
|
|
find_location,
|
|
item_key,
|
|
merge_manifest_items,
|
|
new_original_item,
|
|
selected_url,
|
|
)
|
|
from media_refs import remote_markup
|
|
from models import MediaItem, ThreadData, load_thread
|
|
from paths import FROZEN, SCRATCH
|
|
from render import render_branch, render_spine, title_text
|
|
from slug import branch_file_name, date_prefix, thread_dir_name
|
|
from tree import (
|
|
by_id,
|
|
branch_roots,
|
|
children_map,
|
|
descendants,
|
|
spine_from_tip,
|
|
spine_ids,
|
|
)
|
|
|
|
|
|
def _frontmatter_lines(text: str) -> tuple[list[str], str]:
|
|
"""Return (frontmatter_lines, body_after_closer) for an archive note."""
|
|
if not text.startswith("---\n"):
|
|
return [], text
|
|
end = text.find("\n---", 4)
|
|
if end < 0:
|
|
return [], text
|
|
return text[4:end].splitlines(), text[end + 4 :]
|
|
|
|
|
|
def _tag_values(lines: list[str]) -> list[str]:
|
|
"""Extract tag strings from the ``tags:`` block of a YAML frontmatter lines list."""
|
|
tags: list[str] = []
|
|
in_tags = False
|
|
for line in lines:
|
|
if line == "tags:":
|
|
in_tags = True
|
|
continue
|
|
if in_tags and line.startswith(" - "):
|
|
tags.append(line[4:])
|
|
continue
|
|
if in_tags and line and not line.startswith(" "):
|
|
break
|
|
return tags
|
|
|
|
|
|
def preserve_review_state(
|
|
old: str,
|
|
fresh: str,
|
|
*,
|
|
mechanical_tags: set[str],
|
|
) -> str:
|
|
old_lines, _old_body = _frontmatter_lines(old)
|
|
fresh_lines, fresh_body = _frontmatter_lines(fresh)
|
|
if not old_lines or not fresh_lines:
|
|
return fresh
|
|
old_draft = next(
|
|
(line for line in old_lines if line.startswith("draft: ")),
|
|
"draft: true",
|
|
)
|
|
reviewed = [tag for tag in _tag_values(old_lines) if tag not in mechanical_tags]
|
|
output: list[str] = []
|
|
in_tags = False
|
|
inserted_reviewed = False
|
|
for line in fresh_lines:
|
|
if line.startswith("draft: "):
|
|
output.append(old_draft)
|
|
continue
|
|
if line == "tags:":
|
|
in_tags = True
|
|
output.append(line)
|
|
continue
|
|
if in_tags and line.startswith(" - "):
|
|
output.append(line)
|
|
continue
|
|
if in_tags and not inserted_reviewed:
|
|
output.extend(f" - {tag}" for tag in reviewed if f" - {tag}" not in output)
|
|
inserted_reviewed = True
|
|
in_tags = False
|
|
output.append(line)
|
|
if in_tags and not inserted_reviewed:
|
|
output.extend(f" - {tag}" for tag in reviewed if f" - {tag}" not in output)
|
|
return "---\n" + "\n".join(output) + "\n---" + fresh_body
|
|
|
|
|
|
def missing_local_media(items: list[MediaItem]) -> list[str]:
|
|
missing: list[str] = []
|
|
for item in items:
|
|
local = find_location(item, "local")
|
|
if local is not None and local.integrity == "missing":
|
|
post_id, media_id, role = item_key(item)
|
|
missing.append(f"{post_id}/{media_id}/{role}")
|
|
return missing
|
|
|
|
_POST_ID_LINE: re.Pattern[str] = re.compile(r'^post_id:\s*"?([^"\s]+)"?\s*$')
|
|
_WIKILINK_ITEM: re.Pattern[str] = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EmitResult:
|
|
"""Outcome of one emit pass: which post+handle were written and how many branches."""
|
|
post_id: str
|
|
handle: str
|
|
dir_name: str
|
|
branch_count: int
|
|
missing_media_count: int
|
|
|
|
|
|
def _url_media_id(url: str, index: int) -> str:
|
|
"""Derive a stable media id from the URL's last path segment, falling back to ``m<index>``."""
|
|
segment = Path(unquote(urlparse(url).path)).name
|
|
return segment if segment else f"m{index}"
|
|
|
|
|
|
def _url_ext(url: str) -> str | None:
|
|
"""Extract a lowercase file extension from a URL's ``?format=`` query or path suffix."""
|
|
parsed = urlparse(url)
|
|
fmt = parse_qs(parsed.query).get("format", [None])[0]
|
|
if fmt:
|
|
return fmt.lstrip(".").lower()
|
|
suffix = Path(unquote(parsed.path)).suffix
|
|
if suffix:
|
|
return suffix.lstrip(".").lower()
|
|
return None
|
|
|
|
|
|
def _media_files(media_dir: Path) -> list[Path]:
|
|
if not media_dir.is_dir():
|
|
return []
|
|
return sorted(p for p in media_dir.iterdir() if p.is_file())
|
|
|
|
|
|
def _take_preferred(unused: list[Path], post_id: str, index: int) -> Path | None:
|
|
"""Pop the unused file whose stem equals ``<post_id>_<index>`` if any."""
|
|
want = f"{post_id}_{index}"
|
|
for path in unused:
|
|
if path.stem == want:
|
|
return path
|
|
return None
|
|
|
|
|
|
def _take_fallback(unused: list[Path], post_id: str) -> Path | None:
|
|
"""Pop the first unused file whose name starts with ``<post_id>_`` if any."""
|
|
prefix = f"{post_id}_"
|
|
for path in unused:
|
|
if path.name.startswith(prefix):
|
|
return path
|
|
return None
|
|
|
|
|
|
def collect_media(
|
|
thread: ThreadData,
|
|
input_dir: Path,
|
|
dest_dir: Path,
|
|
*,
|
|
now: str,
|
|
) -> list[MediaItem]:
|
|
"""Copy each post's media files from ``input_dir/media`` into ``dest_dir`` and build canonical MediaItems."""
|
|
unused = _media_files(input_dir / "media")
|
|
items: list[MediaItem] = []
|
|
for post in thread.posts:
|
|
for index, url in enumerate(post.media_urls, start=1):
|
|
media_id = _url_media_id(url, index)
|
|
src = _take_preferred(unused, post.post_id, index)
|
|
if src is None:
|
|
src = _take_fallback(unused, post.post_id)
|
|
if src is not None:
|
|
unused.remove(src)
|
|
ext = src.suffix.lstrip(".") or _url_ext(url) or "bin"
|
|
else:
|
|
ext = _url_ext(url) or "bin"
|
|
filename = f"{post.post_id}_{media_id}_orig.{ext}"
|
|
local_path = dest_dir / filename
|
|
if src is not None:
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, local_path)
|
|
items.append(
|
|
new_original_item(
|
|
post_id=post.post_id,
|
|
media_id=media_id,
|
|
handle=post.handle,
|
|
origin_url=url,
|
|
filename=filename,
|
|
local_path=local_path,
|
|
now=now,
|
|
)
|
|
)
|
|
return items
|
|
|
|
|
|
def selected_media_by_post(
|
|
items: list[MediaItem],
|
|
) -> dict[str, tuple[str, ...]]:
|
|
"""Group embedded items by post_id, returning the HTTPS URL of each visible location."""
|
|
grouped: dict[str, list[str]] = {}
|
|
for item in items:
|
|
if not item.embed:
|
|
continue
|
|
url = selected_url(item)
|
|
if url is not None:
|
|
grouped.setdefault(item.post_id, []).append(url)
|
|
return {post_id: tuple(urls) for post_id, urls in grouped.items()}
|
|
|
|
|
|
def merge_existing_media(
|
|
dest_dir: Path,
|
|
items: list[MediaItem],
|
|
) -> list[MediaItem]:
|
|
"""Merge existing derived/fallback rows with fresh canonical items."""
|
|
old_path = dest_dir / "media.json"
|
|
if not old_path.is_file():
|
|
return items
|
|
try:
|
|
old = json.loads(old_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
return items
|
|
if old.get("schema_version") != 2:
|
|
raise SystemExit(
|
|
"legacy media.json requires: tw.py migrate-media --id <id> --apply"
|
|
)
|
|
existing = _from_wire_dict(old).items
|
|
return list(merge_manifest_items(existing, tuple(items)))
|
|
|
|
|
|
def _legacy_emit_message(*args: object, **kwargs: object) -> None:
|
|
raise SystemExit(
|
|
"legacy media.json requires: tw.py migrate-media --id <id> --apply"
|
|
)
|
|
|
|
|
|
def missing_parent_ids(thread: ThreadData) -> list[str]:
|
|
ids = by_id(thread)
|
|
found: list[str] = []
|
|
seen: set[str] = set()
|
|
for post in thread.posts:
|
|
parent = post.reply_to_id
|
|
if parent is None or parent in ids or parent in seen:
|
|
continue
|
|
seen.add(parent)
|
|
found.append(parent)
|
|
return found
|
|
|
|
|
|
def empty_text_ids(thread: ThreadData) -> list[str]:
|
|
return [p.post_id for p in thread.posts if not (p.text or "").strip()]
|
|
|
|
|
|
def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root") -> str:
|
|
"""Render a human-readable gaps report (quote_of, missing_reply_to, empty_text) for the thread."""
|
|
ids = by_id(thread)
|
|
suggested = spine[-1] if spine else ""
|
|
quote_ids = [p.post_id for p in thread.posts if p.quote_of_id]
|
|
missing = [
|
|
p.post_id
|
|
for p in thread.posts
|
|
if p.reply_to_id is not None and p.reply_to_id not in ids
|
|
]
|
|
empty = empty_text_ids(thread)
|
|
|
|
lines = [
|
|
"# gaps",
|
|
f"input: {input_kind}",
|
|
f"suggested_tip: {suggested}",
|
|
]
|
|
if quote_ids:
|
|
lines.append("quote_of:")
|
|
lines.extend(quote_ids)
|
|
else:
|
|
lines.append("quote_of: unset")
|
|
lines.append("missing_reply_to:")
|
|
lines.extend(missing)
|
|
lines.append("empty_text:")
|
|
lines.extend(empty)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _frontmatter_block(text: str) -> str:
|
|
"""Return the raw YAML block between the ``---`` fences, or ``""`` if not parseable."""
|
|
if not text.startswith("---"):
|
|
return ""
|
|
rest = text[3:]
|
|
if rest.startswith("\n"):
|
|
rest = rest[1:]
|
|
end = rest.find("\n---")
|
|
if end < 0:
|
|
return ""
|
|
return rest[:end]
|
|
|
|
|
|
def _frontmatter_post_id(text: str) -> str | None:
|
|
"""Pull ``post_id`` out of an archive note's YAML frontmatter (returns ``None`` if absent)."""
|
|
for line in _frontmatter_block(text).splitlines():
|
|
match = _POST_ID_LINE.match(line.strip())
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
def collect_existing_ids(vault: Path) -> tuple[set[str], dict[str, Path]]:
|
|
"""Return every archive post_id and spine dirs keyed by spine post_id."""
|
|
threads = vault / "archive" / "threads"
|
|
ids: set[str] = set()
|
|
spines: dict[str, Path] = {}
|
|
if not threads.is_dir():
|
|
return ids, spines
|
|
for md in threads.rglob("*.md"):
|
|
pid = _frontmatter_post_id(md.read_text(encoding="utf-8"))
|
|
if not pid:
|
|
continue
|
|
ids.add(pid)
|
|
if md.name == "index.md" and md.parent.parent.parent == threads:
|
|
spines[pid] = md.parent
|
|
return ids, spines
|
|
|
|
|
|
def _split_frontmatter(text: str) -> tuple[str, str]:
|
|
"""Split an archive note into (frontmatter_with_closing_fence, body). Returns ``("", text)`` if no fence."""
|
|
if not text.startswith("---"):
|
|
return "", text
|
|
end = text.find("\n---", 3)
|
|
if end < 0:
|
|
return "", text
|
|
closer = end + 4
|
|
fm = text[:closer]
|
|
body = text[closer:]
|
|
if body.startswith("\n"):
|
|
fm += "\n"
|
|
body = body[1:]
|
|
elif not fm.endswith("\n"):
|
|
fm += "\n"
|
|
return fm, body
|
|
|
|
|
|
def _upsert_wikilink(path: Path, target: str) -> None:
|
|
"""Insert or move a wikilink target on the handle index.
|
|
|
|
If the handle folder contains a directory whose name matches `target`,
|
|
any other wikilink whose basename does not match a real directory is
|
|
dropped. This keeps the handle index in sync when a folder is reused
|
|
with a new slug (e.g., a tip-climb refresh renames the folder).
|
|
"""
|
|
text = path.read_text(encoding="utf-8")
|
|
fm, body = _split_frontmatter(text)
|
|
if not fm.strip():
|
|
# Frontmatter got nuked by a hand edit. Don't silently emit a
|
|
# bare wikilink list — fall back to a fresh standard frontmatter
|
|
# so the file remains a complete page.
|
|
handle = path.parent.name
|
|
fm = (
|
|
f"---\n"
|
|
f"title: {handle}\n"
|
|
f"type: note\n"
|
|
f"draft: false\n"
|
|
f"description: Archived threads by {handle}.\n"
|
|
f"tags:\n"
|
|
f" - archive\n"
|
|
f" - twitter\n"
|
|
f" - {handle}\n"
|
|
f"---\n"
|
|
)
|
|
kept: list[str] = []
|
|
names: list[str] = []
|
|
handle_dir = path.parent
|
|
real_dirs = {d.name for d in handle_dir.iterdir() if d.is_dir()} if handle_dir.is_dir() else set()
|
|
for line in body.splitlines():
|
|
match = _WIKILINK_ITEM.match(line)
|
|
if match:
|
|
name = match.group(1)
|
|
# wikilinks are vault-root paths; compare the basename to real_dirs
|
|
if name.rsplit("/", 1)[-1] in real_dirs or not real_dirs:
|
|
if name not in names:
|
|
names.append(name)
|
|
else:
|
|
kept.append(line)
|
|
if target not in names:
|
|
names.append(target)
|
|
names.sort()
|
|
while kept and kept[0] == "":
|
|
kept.pop(0)
|
|
while kept and kept[-1] == "":
|
|
kept.pop()
|
|
parts: list[str] = []
|
|
if kept:
|
|
parts.extend(kept)
|
|
parts.append("")
|
|
parts.extend(f"- [[{name}]]" for name in names)
|
|
parts.append("")
|
|
body_text = "\n".join(parts)
|
|
path.write_text(fm.rstrip() + "\n\n" + body_text, encoding="utf-8", newline="\n")
|
|
|
|
|
|
def wiki_thread(handle: str, dir_name: str) -> str:
|
|
"""Build the vault-root wikilink target for a thread spine folder."""
|
|
return f"archive/threads/{handle}/{dir_name}"
|
|
|
|
|
|
def wiki_branch(handle: str, dir_name: str, name: str) -> str:
|
|
"""Build the vault-root wikilink target for one branch file."""
|
|
return f"archive/threads/{handle}/{dir_name}/{name}"
|
|
|
|
|
|
def ensure_handle_index(vault: Path, handle: str, dir_name: str) -> None:
|
|
"""Create the per-handle index.md if missing, otherwise upsert the new thread wikilink."""
|
|
path = vault / "archive" / "threads" / handle / "index.md"
|
|
target = wiki_thread(handle, dir_name)
|
|
if not path.is_file():
|
|
front = (
|
|
"---\n"
|
|
f"title: {handle}\n"
|
|
"type: note\n"
|
|
"draft: false\n"
|
|
f"description: Archived threads by {handle}.\n"
|
|
"tags:\n"
|
|
" - archive\n"
|
|
" - twitter\n"
|
|
f" - {handle}\n"
|
|
"---\n"
|
|
)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
front + f"\n- [[{target}]]\n",
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
return
|
|
_upsert_wikilink(path, target)
|
|
|
|
|
|
def ensure_threads_index(vault: Path, handle: str) -> None:
|
|
"""Create the top-level archive/threads/index.md if missing, otherwise upsert the handle wikilink."""
|
|
path = vault / "archive" / "threads" / "index.md"
|
|
if not path.is_file():
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
(
|
|
"---\n"
|
|
"title: Threads\n"
|
|
"type: note\n"
|
|
"draft: false\n"
|
|
"description: Archived Twitter/X threads and similar sequential posts.\n"
|
|
"tags:\n"
|
|
" - archive\n"
|
|
"---\n"
|
|
"\n"
|
|
"Threads are filed by author handle, then by date and title. "
|
|
"The main note in each folder is that author's own chain. "
|
|
"Replies live beside it.\n"
|
|
"\n"
|
|
"How to read a folder: [[How threads are organized]].\n"
|
|
"\n"
|
|
"Example of the note shape (single spine, no branches): "
|
|
"[[How to archive a thread]]\n"
|
|
"\n"
|
|
f"- [[archive/threads/{handle}]]\n"
|
|
),
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
return
|
|
_upsert_wikilink(path, f"archive/threads/{handle}")
|
|
|
|
|
|
def discover_dumps(root: Path) -> list[Path]:
|
|
"""Return the sorted list of dump subdirectories that contain ``thread_data.json``."""
|
|
dumps = [
|
|
path for path in root.iterdir()
|
|
if path.is_dir() and (path / "thread_data.json").is_file()
|
|
]
|
|
dumps.sort(key=lambda path: path.name)
|
|
return dumps
|
|
|
|
|
|
def _gap_line(
|
|
post_id: str,
|
|
handle: str,
|
|
missing: list[str],
|
|
empty: list[str],
|
|
) -> str:
|
|
miss = ",".join(missing) if missing else "-"
|
|
empt = ",".join(empty) if empty else "-"
|
|
return f"{post_id} {handle} root {miss} {empt}"
|
|
|
|
|
|
def emit(
|
|
input_dir: Path,
|
|
vault: Path,
|
|
slug: str | None,
|
|
archived: str,
|
|
*,
|
|
force: bool = False,
|
|
reuse_dir: Path | None = None,
|
|
tip: str | None = None,
|
|
reconcile_scratch: Path = SCRATCH,
|
|
frozen_ids: set[str] | None = None,
|
|
) -> EmitResult:
|
|
src = input_dir / "thread_data.json"
|
|
if not src.is_file():
|
|
raise SystemExit(f"missing {src}")
|
|
|
|
thread = load_thread(src)
|
|
if not thread.posts:
|
|
raise SystemExit(f"no posts in {src}")
|
|
|
|
if tip:
|
|
spine = spine_from_tip(thread, tip)
|
|
input_kind = "tip"
|
|
else:
|
|
spine = spine_ids(thread)
|
|
input_kind = "root"
|
|
roots = branch_roots(thread, spine)
|
|
ids = by_id(thread)
|
|
kids = children_map(thread)
|
|
first = ids[spine[0]]
|
|
handle = first.handle
|
|
rendered_title = title_text(first.text)
|
|
desired_dir_name = thread_dir_name(date_prefix(first.timestamp), rendered_title, slug)
|
|
|
|
if reuse_dir is None and force:
|
|
_existing_ids, spines = collect_existing_ids(vault)
|
|
# Only accept a candidate whose parent handle matches the
|
|
# OP handle. A thread previously archived at a different
|
|
# author's dir (e.g. an old tip-as-root archive of a cross-
|
|
# author conversation) is treated as orphan, not a reuse.
|
|
candidate = spines.get(thread.root_post_id)
|
|
if candidate is not None and candidate.parent.name == handle:
|
|
reuse_dir = candidate
|
|
if reuse_dir is None:
|
|
for post in thread.posts:
|
|
candidate = spines.get(post.post_id)
|
|
if candidate is not None and candidate.parent.name == handle:
|
|
reuse_dir = candidate
|
|
break
|
|
if reuse_dir is None and slug:
|
|
candidate = vault / "archive" / "threads" / handle / slug
|
|
if candidate.is_dir():
|
|
reuse_dir = candidate
|
|
|
|
if reuse_dir is not None:
|
|
handle = reuse_dir.parent.name
|
|
dir_name = reuse_dir.name
|
|
note_dir = reuse_dir
|
|
asset_dir = vault / "assets" / "threads" / handle / dir_name
|
|
if dir_name != desired_dir_name:
|
|
active_frozen_ids = (
|
|
frozen_ids if frozen_ids is not None else load_frozen_ids(FROZEN)
|
|
)
|
|
match = frozen_match(asset_dir, active_frozen_ids)
|
|
if match is not None:
|
|
raise SystemExit(f"frozen: skipped ({match})")
|
|
new_archive = vault / "archive" / "threads" / handle / desired_dir_name
|
|
new_asset = vault / "assets" / "threads" / handle / desired_dir_name
|
|
move = ReslugMove(
|
|
handle=handle,
|
|
old_dir_name=dir_name,
|
|
new_dir_name=desired_dir_name,
|
|
archive_old=reuse_dir,
|
|
archive_new=new_archive,
|
|
asset_old=asset_dir,
|
|
asset_new=new_asset,
|
|
)
|
|
rename_thread_pair(move)
|
|
frozen_dirs = plan_reslug(vault, active_frozen_ids).frozen_dirs
|
|
old_prefix = f"archive/threads/{handle}/{dir_name}"
|
|
new_prefix = f"archive/threads/{handle}/{desired_dir_name}"
|
|
rewrite_archive_paths(
|
|
vault,
|
|
reconcile_scratch,
|
|
{old_prefix: new_prefix},
|
|
frozen_dirs,
|
|
)
|
|
dir_name = desired_dir_name
|
|
note_dir = new_archive
|
|
asset_dir = new_asset
|
|
else:
|
|
dir_name = desired_dir_name
|
|
note_dir = vault / "archive" / "threads" / handle / dir_name
|
|
asset_dir = vault / "assets" / "threads" / handle / dir_name
|
|
if note_dir.exists() or asset_dir.exists():
|
|
raise SystemExit(f"destination occupied: {note_dir}")
|
|
|
|
branch_names: dict[str, str] = {}
|
|
for rid in roots:
|
|
branch = ids[rid]
|
|
fname = branch_file_name(
|
|
date_prefix(branch.timestamp),
|
|
branch.handle,
|
|
branch.text,
|
|
)
|
|
name = fname[:-3] if fname.endswith(".md") else fname
|
|
branch_names[rid] = name
|
|
|
|
note_dir.mkdir(parents=True, exist_ok=True)
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
now = archived + "T12:00:00Z"
|
|
fresh_items = collect_media(thread, input_dir, asset_dir, now=now)
|
|
items = merge_existing_media(asset_dir, fresh_items)
|
|
media_by_post = selected_media_by_post(items)
|
|
missing_keys = missing_local_media(items)
|
|
|
|
branch_links = {
|
|
rid: wiki_branch(handle, dir_name, name)
|
|
for rid, name in branch_names.items()
|
|
}
|
|
|
|
spine_text = render_spine(
|
|
thread,
|
|
spine,
|
|
roots,
|
|
branch_links,
|
|
archived,
|
|
media_by_post,
|
|
)
|
|
spine_path = note_dir / "index.md"
|
|
if spine_path.is_file():
|
|
spine_text = preserve_review_state(
|
|
spine_path.read_text(encoding="utf-8"),
|
|
spine_text,
|
|
mechanical_tags={"archive", "twitter", handle},
|
|
)
|
|
spine_path.write_text(spine_text, encoding="utf-8", newline="\n")
|
|
|
|
for rid in roots:
|
|
branch_path = note_dir / f"{branch_names[rid]}.md"
|
|
branch_text = render_branch(
|
|
thread,
|
|
rid,
|
|
descendants(rid, kids),
|
|
first.handle,
|
|
archived,
|
|
media_by_post,
|
|
spine_wiki=wiki_thread(handle, dir_name),
|
|
)
|
|
if branch_path.is_file():
|
|
branch_text = preserve_review_state(
|
|
branch_path.read_text(encoding="utf-8"),
|
|
branch_text,
|
|
mechanical_tags={"archive", "twitter", handle},
|
|
)
|
|
branch_path.write_text(branch_text, encoding="utf-8", newline="\n")
|
|
|
|
shutil.copy2(src, asset_dir / "thread_data.json")
|
|
media_doc = {
|
|
"schema_version": 2,
|
|
"root_post_id": thread.root_post_id,
|
|
"items": [_item_to_wire(item) for item in items],
|
|
"mirrors": [],
|
|
}
|
|
atomic_write_json(asset_dir / "media.json", media_doc)
|
|
gaps_text = render_gaps(thread, spine, input_kind=input_kind)
|
|
if missing_keys:
|
|
gaps_text += "\nmissing_media_local:\n" + "\n".join(missing_keys) + "\n"
|
|
(asset_dir / "gaps.md").write_text(
|
|
gaps_text,
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
|
|
ensure_handle_index(vault, handle, dir_name)
|
|
ensure_threads_index(vault, handle)
|
|
|
|
return EmitResult(
|
|
post_id=thread.root_post_id,
|
|
handle=handle,
|
|
dir_name=dir_name,
|
|
branch_count=len(roots),
|
|
missing_media_count=len(missing_keys),
|
|
)
|
|
|
|
|
|
def emit_all(
|
|
dumps_root: Path,
|
|
vault: Path,
|
|
archived: str,
|
|
force: bool,
|
|
) -> tuple[int, int, int, int]:
|
|
dumps = discover_dumps(dumps_root)
|
|
existing_ids, spines = collect_existing_ids(vault)
|
|
lines: list[str] = []
|
|
skipped = 0
|
|
written = 0
|
|
errors = 0
|
|
|
|
for dump in dumps:
|
|
src = dump / "thread_data.json"
|
|
try:
|
|
thread = load_thread(src)
|
|
if not thread.posts:
|
|
raise ValueError(f"no posts in {src}")
|
|
spine = spine_ids(thread)
|
|
handle = by_id(thread)[spine[0]].handle
|
|
missing = missing_parent_ids(thread)
|
|
empty = empty_text_ids(thread)
|
|
post_id = thread.root_post_id
|
|
except Exception as exc:
|
|
logger.warning("emit_all: skipping %s: %s", dump, exc)
|
|
lines.append(f"{dump.name} ? error - -")
|
|
errors += 1
|
|
continue
|
|
|
|
lines.append(_gap_line(post_id, handle, missing, empty))
|
|
|
|
if post_id in existing_ids and not force:
|
|
skipped += 1
|
|
continue
|
|
|
|
reuse = spines.get(post_id) if force else None
|
|
try:
|
|
emit(
|
|
dump,
|
|
vault,
|
|
None,
|
|
archived,
|
|
force=force,
|
|
reuse_dir=reuse,
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("emit_all: skipping %s: %s", dump, exc)
|
|
errors += 1
|
|
continue
|
|
|
|
written += 1
|
|
existing_ids.add(post_id)
|
|
|
|
out = vault / "assets" / "threads" / "_batch_gaps.md"
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")
|
|
return len(dumps), skipped, written, errors
|
|
|
|
|
|
# --- reslug (one-time bulk title/path reconciliation) ---
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReslugMove:
|
|
"""One paired archive/asset directory rename."""
|
|
handle: str
|
|
old_dir_name: str
|
|
new_dir_name: str
|
|
archive_old: Path
|
|
archive_new: Path
|
|
asset_old: Path
|
|
asset_new: Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReslugConflict:
|
|
"""One bulk preflight failure surfaced before any rename."""
|
|
path: Path
|
|
reason: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReslugPlan:
|
|
"""Small immutable one-time corpus repair plan."""
|
|
vault: Path
|
|
moves: tuple[ReslugMove, ...]
|
|
noops: tuple[Path, ...]
|
|
frozen: tuple[Path, ...]
|
|
frozen_dirs: tuple[Path, ...]
|
|
conflicts: tuple[ReslugConflict, ...]
|
|
|
|
@property
|
|
def can_apply(self) -> bool:
|
|
return not self.conflicts
|
|
|
|
|
|
_TITLE_LINE: re.Pattern[str] = re.compile(r"^title:\s*(.+?)\s*$")
|
|
_DATE_LINE: re.Pattern[str] = re.compile(r"^date:\s*(\d{4}-\d{2}-\d{2})\s*$")
|
|
|
|
|
|
def _yaml_scalar_value(raw: str) -> str:
|
|
"""Decode a single-line YAML scalar that is either JSON-quoted or unquoted."""
|
|
s = raw.strip()
|
|
if len(s) >= 2 and s.startswith('"') and s.endswith('"'):
|
|
body = s[1:-1]
|
|
body = body.replace("\\\\", "\x00").replace('\\"', '"').replace("\x00", "\\")
|
|
return body
|
|
return s
|
|
|
|
|
|
def _frontmatter_title_date(text: str) -> tuple[str, str] | None:
|
|
"""Return (title, date) from an archive note's frontmatter or ``None`` when unparseable."""
|
|
block = _frontmatter_block(text)
|
|
if not block:
|
|
return None
|
|
title: str | None = None
|
|
date_value: str | None = None
|
|
for line in block.splitlines():
|
|
stripped = line.strip()
|
|
if title is None:
|
|
match = _TITLE_LINE.match(line)
|
|
if match:
|
|
title = _yaml_scalar_value(match.group(1))
|
|
continue
|
|
if date_value is None:
|
|
match = _DATE_LINE.match(line)
|
|
if match:
|
|
date_value = match.group(1)
|
|
continue
|
|
if title is None or date_value is None:
|
|
return None
|
|
return title, date_value
|
|
|
|
|
|
_RESLUG_EXCLUDED_TOP_LEVEL: tuple[str, ...] = (
|
|
".git",
|
|
"site",
|
|
"assets",
|
|
"secrets",
|
|
"node_modules",
|
|
)
|
|
|
|
|
|
def plan_reslug(vault: Path, frozen_ids: set[str]) -> ReslugPlan:
|
|
"""Build an immutable repair plan for ``archive/threads/<handle>/<thread>`` directories."""
|
|
moves: list[ReslugMove] = []
|
|
noops: list[Path] = []
|
|
frozen_entries: list[Path] = []
|
|
frozen_dirs: list[Path] = []
|
|
conflicts: list[ReslugConflict] = []
|
|
|
|
threads_root = vault / "archive" / "threads"
|
|
assets_root = vault / "assets" / "threads"
|
|
|
|
if not threads_root.is_dir():
|
|
return ReslugPlan(vault, (), (), (), (), ())
|
|
|
|
for handle_dir in sorted(threads_root.iterdir(), key=lambda p: p.name):
|
|
if not handle_dir.is_dir():
|
|
continue
|
|
handle = handle_dir.name
|
|
for thread_dir in sorted(handle_dir.iterdir(), key=lambda p: p.name):
|
|
if not thread_dir.is_dir():
|
|
continue
|
|
old_dir_name = thread_dir.name
|
|
index_path = thread_dir / "index.md"
|
|
asset_dir = assets_root / handle / old_dir_name
|
|
try:
|
|
if not index_path.is_file():
|
|
conflicts.append(
|
|
ReslugConflict(thread_dir, "missing index.md")
|
|
)
|
|
continue
|
|
try:
|
|
text = index_path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
conflicts.append(
|
|
ReslugConflict(index_path, f"unreadable frontmatter: {exc}")
|
|
)
|
|
continue
|
|
parsed = _frontmatter_title_date(text)
|
|
if parsed is None:
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
index_path,
|
|
"missing or malformed frontmatter title/date",
|
|
)
|
|
)
|
|
continue
|
|
title, date_value = parsed
|
|
if not asset_dir.is_dir():
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
index_path,
|
|
f"missing asset pair at {asset_dir}",
|
|
)
|
|
)
|
|
continue
|
|
thread_data = asset_dir / "thread_data.json"
|
|
media_json = asset_dir / "media.json"
|
|
if not thread_data.is_file():
|
|
conflicts.append(
|
|
ReslugConflict(index_path, "missing thread_data.json")
|
|
)
|
|
continue
|
|
if not media_json.is_file():
|
|
conflicts.append(
|
|
ReslugConflict(index_path, "missing media.json")
|
|
)
|
|
continue
|
|
try:
|
|
json.loads(thread_data.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
index_path,
|
|
f"invalid thread_data.json: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
try:
|
|
json.loads(media_json.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
index_path,
|
|
f"invalid media.json: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
expected = thread_dir_name(date_value, title, None)
|
|
match = frozen_match(asset_dir, frozen_ids)
|
|
if old_dir_name == expected:
|
|
noops.append(thread_dir)
|
|
if match is not None:
|
|
frozen_dirs.append(thread_dir)
|
|
continue
|
|
if match is not None:
|
|
frozen_entries.append(thread_dir)
|
|
frozen_dirs.append(thread_dir)
|
|
continue
|
|
new_archive = threads_root / handle / expected
|
|
new_asset = assets_root / handle / expected
|
|
if new_archive.exists():
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
thread_dir,
|
|
f"destination occupied: {new_archive}",
|
|
)
|
|
)
|
|
continue
|
|
if new_asset.exists():
|
|
conflicts.append(
|
|
ReslugConflict(
|
|
thread_dir,
|
|
f"destination occupied: {new_asset}",
|
|
)
|
|
)
|
|
continue
|
|
moves.append(
|
|
ReslugMove(
|
|
handle=handle,
|
|
old_dir_name=old_dir_name,
|
|
new_dir_name=expected,
|
|
archive_old=thread_dir,
|
|
archive_new=new_archive,
|
|
asset_old=asset_dir,
|
|
asset_new=new_asset,
|
|
)
|
|
)
|
|
except OSError as exc:
|
|
conflicts.append(
|
|
ReslugConflict(thread_dir, f"filesystem error: {exc}")
|
|
)
|
|
continue
|
|
|
|
moves.sort(key=lambda m: str(m.archive_old))
|
|
noops.sort(key=str)
|
|
frozen_entries.sort(key=str)
|
|
frozen_dirs.sort(key=str)
|
|
conflicts.sort(key=lambda c: str(c.path))
|
|
return ReslugPlan(
|
|
vault,
|
|
tuple(moves),
|
|
tuple(noops),
|
|
tuple(frozen_entries),
|
|
tuple(frozen_dirs),
|
|
tuple(conflicts),
|
|
)
|
|
|
|
|
|
def rename_thread_pair(move: ReslugMove) -> None:
|
|
"""Validate, rename assets first, then archive; roll the asset back on archive failure."""
|
|
if not move.archive_old.is_dir() or not move.asset_old.is_dir():
|
|
raise RuntimeError(f"source pair missing: {move.archive_old}")
|
|
if move.archive_new.exists() or move.asset_new.exists():
|
|
raise RuntimeError(f"destination occupied: {move.archive_new}")
|
|
move.asset_old.rename(move.asset_new)
|
|
try:
|
|
move.archive_old.rename(move.archive_new)
|
|
except OSError:
|
|
move.asset_new.rename(move.asset_old)
|
|
raise
|
|
|
|
|
|
def rewrite_archive_paths(
|
|
vault: Path,
|
|
scratch: Path,
|
|
replacements: dict[str, str],
|
|
frozen_dirs: tuple[Path, ...],
|
|
) -> tuple[Path, ...]:
|
|
"""Recursively rewrite exact vault-root prefixes in mutable Markdown and Canvas files."""
|
|
if not replacements:
|
|
return ()
|
|
|
|
keys = sorted(replacements.keys(), key=len, reverse=True)
|
|
pattern = re.compile("|".join(re.escape(key) for key in keys))
|
|
|
|
changed: list[Path] = []
|
|
stage_dir = scratch / "reslug_text"
|
|
sequence = 0
|
|
for path in sorted(vault.rglob("*")):
|
|
if not path.is_file():
|
|
continue
|
|
if path.suffix.lower() not in (".md", ".canvas"):
|
|
continue
|
|
try:
|
|
rel_parts = path.relative_to(vault).parts
|
|
except ValueError:
|
|
continue
|
|
if rel_parts and rel_parts[0] in _RESLUG_EXCLUDED_TOP_LEVEL:
|
|
continue
|
|
skip = False
|
|
for frozen in frozen_dirs:
|
|
try:
|
|
path.relative_to(frozen)
|
|
skip = True
|
|
break
|
|
except ValueError:
|
|
continue
|
|
if skip:
|
|
continue
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
continue
|
|
new_text = pattern.sub(lambda m: replacements[m.group(0)], text)
|
|
if new_text == text:
|
|
continue
|
|
stage_dir.mkdir(parents=True, exist_ok=True)
|
|
stage_path = stage_dir / f"{sequence}.tmp"
|
|
sequence += 1
|
|
stage_path.write_text(new_text, encoding="utf-8", newline="\n")
|
|
os.replace(stage_path, path)
|
|
changed.append(path)
|
|
|
|
try:
|
|
stage_dir.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
changed.sort(key=str)
|
|
return tuple(changed)
|
|
|
|
|
|
def apply_reslug_plan(plan: ReslugPlan, scratch: Path) -> tuple[Path, ...]:
|
|
"""Run every rename, then recursively rewrite mutable paths; refuse on conflict."""
|
|
if plan.conflicts:
|
|
reasons = ", ".join(c.reason for c in plan.conflicts)
|
|
raise RuntimeError(
|
|
f"reslug blocked by {len(plan.conflicts)} conflict(s): {reasons}"
|
|
)
|
|
replacements: dict[str, str] = {}
|
|
for move in plan.moves:
|
|
rename_thread_pair(move)
|
|
old_prefix = f"archive/threads/{move.handle}/{move.old_dir_name}"
|
|
new_prefix = f"archive/threads/{move.handle}/{move.new_dir_name}"
|
|
replacements[old_prefix] = new_prefix
|
|
print(f"renamed -> {move.archive_new}")
|
|
rewritten = (
|
|
rewrite_archive_paths(
|
|
plan.vault,
|
|
scratch,
|
|
replacements,
|
|
plan.frozen_dirs,
|
|
)
|
|
if plan.moves
|
|
else ()
|
|
)
|
|
for path in rewritten:
|
|
print(f"rewrote -> {path}")
|
|
return rewritten
|
|
|
|
|
|
def format_reslug_plan(plan: ReslugPlan) -> str:
|
|
"""Return a deterministic human-readable summary of a repair plan."""
|
|
lines: list[str] = []
|
|
for path in plan.noops:
|
|
lines.append(f"[noop] {path.as_posix()}")
|
|
for move in plan.moves:
|
|
lines.append(
|
|
f"[rename] {move.archive_old.as_posix()} -> {move.archive_new.as_posix()}"
|
|
)
|
|
for path in plan.frozen:
|
|
lines.append(f"[frozen] {path.as_posix()}")
|
|
for conflict in plan.conflicts:
|
|
lines.append(f"[conflict] {conflict.path.as_posix()}: {conflict.reason}")
|
|
lines.append(
|
|
f"summary: moves={len(plan.moves)} noops={len(plan.noops)} "
|
|
f"frozen={len(plan.frozen)} conflicts={len(plan.conflicts)}"
|
|
)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Emit Threadwell archive notes from a thread_data.json dump. "
|
|
"No fetch, no lift."
|
|
),
|
|
)
|
|
source = parser.add_mutually_exclusive_group(required=True)
|
|
source.add_argument(
|
|
"--input",
|
|
type=Path,
|
|
help="Dump directory containing thread_data.json",
|
|
)
|
|
source.add_argument(
|
|
"--all",
|
|
dest="all_root",
|
|
type=Path,
|
|
metavar="DUMPS_ROOT",
|
|
help="Convert every dump directory under this root",
|
|
)
|
|
parser.add_argument(
|
|
"--vault",
|
|
required=True,
|
|
type=Path,
|
|
help="Threadwell vault root",
|
|
)
|
|
parser.add_argument(
|
|
"--slug",
|
|
default=None,
|
|
help="Optional slug override for a single --input (ignored with --all)",
|
|
)
|
|
parser.add_argument(
|
|
"--archived",
|
|
default=date.today().isoformat(),
|
|
help="Archive date YYYY-MM-DD (default: today local)",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Emit even if post_id already exists in the vault",
|
|
)
|
|
parser.add_argument(
|
|
"--tip",
|
|
default=None,
|
|
help="Climb reply_to from this post_id for the spine (same handle)",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
if args.all_root is not None:
|
|
seen, skipped, written, errors = emit_all(
|
|
args.all_root,
|
|
args.vault,
|
|
args.archived,
|
|
args.force,
|
|
)
|
|
print(
|
|
f"dumps seen={seen} skipped={skipped} "
|
|
f"written={written} errors={errors}"
|
|
)
|
|
return 0 if errors == 0 else 1
|
|
emit(
|
|
args.input,
|
|
args.vault,
|
|
args.slug,
|
|
args.archived,
|
|
force=args.force,
|
|
tip=args.tip,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|