Files
Threadwell/scripts/twitter/emit_archive.py
T

833 lines
26 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 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.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.render import render_branch, render_spine
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 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 render import render_branch, render_spine
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 unique_dir_name(parent: Path, base: str) -> str:
"""Return ``base`` if unused, else ``base-2``, ``base-3``, ... until a free name is found."""
if not (parent / base).exists():
return base
n = 2
while (parent / f"{base}-{n}").exists():
n += 1
return f"{base}-{n}"
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,
) -> 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
first_line = (first.text or "").split("\n", 1)[0]
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:
dir_name = reuse_dir.name
handle = reuse_dir.parent.name
note_dir = reuse_dir
asset_dir = vault / "assets" / "threads" / handle / dir_name
else:
dir_name = thread_dir_name(date_prefix(first.timestamp), first_line, slug)
if slug is None:
dir_name = unique_dir_name(
vault / "archive" / "threads" / handle,
dir_name,
)
note_dir = vault / "archive" / "threads" / handle / dir_name
asset_dir = vault / "assets" / "threads" / handle / dir_name
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
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())