manual_slop/scripts/twitter_threads/render_markdown.py

"""Render ThreadData into a Markdown file with YAML front-matter."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
try:
 from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, ThreadData, PostData, thread_from_dict
except ImportError:
 from error_types import ErrorInfo, make_error, Result, ThreadData, PostData, thread_from_dict

_EM_DASH = "\u2014"

def _build_id_to_index(thread: ThreadData) -> dict[str, int]:
 return {post.post_id: i + 1 for i, post in enumerate(thread.posts)}

def _format_view_count(value: int | None) -> str:
 if value is None:
  return "null"
 return str(value)

def _root_post(thread: ThreadData) -> PostData:
 for p in thread.posts:
  if p.post_id == thread.root_post_id:
   return p
 return thread.posts[0]

def _render_yaml(thread: ThreadData, title: str) -> list[str]:
 root = _root_post(thread)
 return [
  "---",
  f'title: "{title}"',
  f'author: "{root.author}"',
  f'handle: "@{root.handle}"',
  f'post_url: "{thread.source_url}"',
  f'post_id: "{thread.root_post_id}"',
  f'timestamp: "{root.timestamp}"',
  f'post_count: {len(thread.posts)}',
  f'reply_count: {root.metrics.reply_count}',
  f'repost_count: {root.metrics.repost_count}',
  f'like_count: {root.metrics.like_count}',
  f'view_count: {_format_view_count(root.metrics.view_count)}',
  "---",
 ]

def _media_line(index: int, name: str) -> str:
 ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
 if ext in ("mp4", "mov", "webm", "m4v"):
  return f'<video controls src="./media/{name}"></video>'
 if ext in ("jpg", "jpeg", "png", "gif", "webp", "bmp"):
  return f"![Media {index}](./media/{name})"
 return f"[Media {index}](./media/{name})"

def _render_post(post: PostData, n: int, id_to_index: dict[str, int], media_names: dict[str, list[str]]) -> list[str]:
 lines: list[str] = []
 header = f"## Post {n} ({post.timestamp})"
 if post.reply_to_id is not None and post.reply_to_id in id_to_index:
  header += f" {_EM_DASH} reply to Post {id_to_index[post.reply_to_id]}"
 lines.append(header)
 lines.append("")
 lines.append(post.text)
 lines.append("")
 if post.quote_of_id is not None:
  lines.append(f"> [Quoting post {post.quote_of_id}](https://x.com/i/web/status/{post.quote_of_id})")
  lines.append("")
 names = media_names.get(post.post_id, [])
 if names:
  for k, name in enumerate(names, start=1):
   lines.append(_media_line(k, name))
  lines.append("")
 return lines

def render_markdown(thread: ThreadData, media_names: dict[str, list[str]], output: Path) -> Result[Path]:
 lines: list[str] = []
 id_to_index = _build_id_to_index(thread)
 root = _root_post(thread)
 first_line = root.text.splitlines()[0] if root.text.splitlines() else ""
 title = first_line[:80]
 lines.extend(_render_yaml(thread, title))
 lines.append("")
 lines.append(f"# @{root.handle} {_EM_DASH} {title}")
 lines.append("")
 for i, post in enumerate(thread.posts):
  n = i + 1
  lines.extend(_render_post(post, n, id_to_index, media_names))
 content = "\n".join(lines)
 try:
  output.write_text(content, encoding="utf-8")
 except OSError as e:
  return Result.err(make_error("WriteError", "render_markdown", str(e)))
 return Result.ok(output)

def main(argv: list[str] | None = None) -> int:
 parser = argparse.ArgumentParser(description="Render a Twitter/X thread_data.json to Markdown.")
 parser.add_argument("--input", type=Path, required=True, help="thread_data.json path")
 parser.add_argument("--media-dir", type=Path, required=True, help="media directory to link")
 parser.add_argument("--output", type=Path, required=True, help="output .md path")
 args = parser.parse_args(argv)
 try:
  raw = args.input.read_text(encoding="utf-8")
 except OSError as e:
  sys.stderr.write(f"error: ReadError: {e}\n")
  return 1
 try:
  data = json.loads(raw)
 except json.JSONDecodeError as e:
  sys.stderr.write(f"error: JsonParseError: {e}\n")
  return 1
 thread = thread_from_dict(data)
 media_names: dict[str, list[str]] = {}
 for post in thread.posts:
  if args.media_dir.exists():
   files = sorted(f.name for f in args.media_dir.glob(f"{post.post_id}_*"))
   if files:
    media_names[post.post_id] = files
 result = render_markdown(thread, media_names, args.output)
 if not result.is_ok:
  sys.stderr.write(f"error: {result.error.kind}: {result.error.detail}\n")
  return 1
 print(str(args.output))
 return 0


if __name__ == "__main__":
 sys.exit(main())