357 lines
13 KiB
Python
357 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate a complete Threads page from one compact high-level JSON spec."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
|
|
import render_threads_page as renderer
|
|
|
|
REPLY_NAMES = ("林夏", "周野", "陈默", "顾念", "沈言", "宋可心", "许知夏", "苏小鹿")
|
|
REPLY_COLORS = (
|
|
("#5662f6", "#9f45d8"),
|
|
("#15a88b", "#2e88cc"),
|
|
("#f97316", "#db2777"),
|
|
("#0f7d78", "#51b35e"),
|
|
)
|
|
REPLY_TEMPLATES = (
|
|
"关于“{topic}”，这个表达很有意思。",
|
|
"我也在关注“{topic}”，期待看到后续更新。",
|
|
"从另一个角度看，“{topic}”还有很多值得讨论的细节。",
|
|
"这条串文把“{topic}”说得很清楚。",
|
|
"先收藏，也想听听大家对“{topic}”的看法。",
|
|
"内容不长，但“{topic}”这个重点已经交代得很完整。",
|
|
"这正是我最近在思考的“{topic}”问题。",
|
|
"希望评论区可以继续理性讨论“{topic}”。",
|
|
)
|
|
|
|
|
|
def _read_spec(path: Path) -> dict[str, object]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as exc:
|
|
raise renderer.DataValidationError(f"Invalid spec JSON: {exc}") from exc
|
|
if not isinstance(value, dict):
|
|
raise renderer.DataValidationError("Spec JSON must be an object")
|
|
return value
|
|
|
|
|
|
def _string(value: object, path: str, *, required: bool = False) -> str:
|
|
if value is None and not required:
|
|
return ""
|
|
if not isinstance(value, str) or (required and not value.strip()):
|
|
suffix = "a non-empty string" if required else "a string"
|
|
raise renderer.DataValidationError(f"{path} must be {suffix}")
|
|
return value.strip()
|
|
|
|
|
|
def _integer(value: object, path: str, *, default: int = 0) -> int:
|
|
if value is None:
|
|
return default
|
|
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
|
raise renderer.DataValidationError(f"{path} must be a non-negative integer")
|
|
return value
|
|
|
|
|
|
def _boolean(value: object, path: str, *, default: bool = False) -> bool:
|
|
if value is None:
|
|
return default
|
|
if not isinstance(value, bool):
|
|
raise renderer.DataValidationError(f"{path} must be a boolean")
|
|
return value
|
|
|
|
|
|
def _strings(value: object, path: str) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if not isinstance(value, list) or not all(isinstance(item, str) and item.strip() for item in value):
|
|
raise renderer.DataValidationError(f"{path} must be an array of non-empty strings")
|
|
return [item.strip() for item in value]
|
|
|
|
|
|
def _require_chinese(value: str, path: str) -> None:
|
|
if not any("\u3400" <= char <= "\u9fff" for char in value):
|
|
raise renderer.DataValidationError(f"{path} must contain Simplified Chinese reader-visible text")
|
|
|
|
|
|
def _reply_objects(
|
|
count: int,
|
|
topics: list[str],
|
|
custom_replies: list[object],
|
|
) -> list[dict[str, object]]:
|
|
if not 0 <= count <= 8:
|
|
raise renderer.DataValidationError("displayReplyCount must be between 0 and 8")
|
|
if len(custom_replies) > count:
|
|
raise renderer.DataValidationError("displayReplyCount cannot be smaller than replies.length")
|
|
replies: list[dict[str, object]] = []
|
|
for index in range(count):
|
|
name = REPLY_NAMES[index % len(REPLY_NAMES)]
|
|
reply: dict[str, object] = {
|
|
"author": {
|
|
"name": name,
|
|
"initial": name[:1],
|
|
"colors": list(REPLY_COLORS[index % len(REPLY_COLORS)]),
|
|
},
|
|
"time": f"{index + 1} 分钟",
|
|
"text": REPLY_TEMPLATES[index % len(REPLY_TEMPLATES)].format(topic=topics[index % len(topics)]),
|
|
"metrics": {
|
|
"likes": max(0, (count - index) * 9),
|
|
"replies": index % 4,
|
|
"reposts": index % 2,
|
|
},
|
|
}
|
|
if index < len(custom_replies):
|
|
custom = custom_replies[index]
|
|
if isinstance(custom, str):
|
|
reply["text"] = custom.strip()
|
|
elif isinstance(custom, Mapping):
|
|
reply["text"] = _string(
|
|
custom.get("text"),
|
|
f"replies[{index}].text",
|
|
required=True,
|
|
)
|
|
if "author" in custom:
|
|
author_name = _string(
|
|
custom.get("author"),
|
|
f"replies[{index}].author",
|
|
required=True,
|
|
)
|
|
reply["author"]["name"] = author_name
|
|
reply["author"]["initial"] = author_name[:1]
|
|
if "time" in custom:
|
|
reply["time"] = _string(
|
|
custom.get("time"),
|
|
f"replies[{index}].time",
|
|
required=True,
|
|
)
|
|
metrics = reply["metrics"]
|
|
for key in ("likes", "replies", "reposts"):
|
|
if key in custom:
|
|
metrics[key] = _integer(
|
|
custom.get(key),
|
|
f"replies[{index}].{key}",
|
|
)
|
|
else:
|
|
raise renderer.DataValidationError(f"replies[{index}] must be a string or object")
|
|
_require_chinese(str(reply["text"]), f"replies[{index}].text")
|
|
replies.append(reply)
|
|
return replies
|
|
|
|
|
|
def _media(value: object, path: str) -> dict[str, object] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, Mapping):
|
|
raise renderer.DataValidationError(f"{path} must be an object or null")
|
|
media_type = _string(value.get("type"), f"{path}.type") or "image"
|
|
if media_type not in {"image", "video", "webpage"}:
|
|
raise renderer.DataValidationError(f"{path}.type must be image, video, or webpage")
|
|
media: dict[str, object] = {"type": media_type}
|
|
if media_type == "webpage":
|
|
media["pageUrl"] = _string(
|
|
value.get("pageUrl"),
|
|
f"{path}.pageUrl",
|
|
required=True,
|
|
)
|
|
media["title"] = _string(
|
|
value.get("title"),
|
|
f"{path}.title",
|
|
required=True,
|
|
)
|
|
media["description"] = _string(
|
|
value.get("description"),
|
|
f"{path}.description",
|
|
required=True,
|
|
)
|
|
else:
|
|
media["mediaUrl"] = _string(
|
|
value.get("url"),
|
|
f"{path}.url",
|
|
required=True,
|
|
)
|
|
media["alt"] = _string(value.get("alt"), f"{path}.alt") or "用户提供的媒体"
|
|
if value.get("pageUrl"):
|
|
media["pageUrl"] = _string(value.get("pageUrl"), f"{path}.pageUrl")
|
|
return media
|
|
|
|
|
|
def build_document(spec: dict[str, object]) -> dict[str, object]:
|
|
allowed = {
|
|
"accountName",
|
|
"handle",
|
|
"verified",
|
|
"description",
|
|
"pageTitle",
|
|
"topic",
|
|
"replyTopics",
|
|
"displayReplyCount",
|
|
"totalReplyCount",
|
|
"replies",
|
|
"postText",
|
|
"posts",
|
|
"media",
|
|
"likes",
|
|
"reposts",
|
|
"following",
|
|
"followers",
|
|
"threadCount",
|
|
}
|
|
unknown = sorted(set(spec) - allowed)
|
|
if unknown:
|
|
raise renderer.DataValidationError("Unknown spec keys: " + ", ".join(unknown))
|
|
account_name = _string(spec.get("accountName"), "accountName", required=True)
|
|
handle = _string(spec.get("handle"), "handle") or "@threads_user"
|
|
topic = _string(spec.get("topic"), "topic") or account_name
|
|
reply_topics = _strings(spec.get("replyTopics"), "replyTopics") or [topic]
|
|
raw_posts = spec.get("posts")
|
|
if raw_posts is None:
|
|
raw_posts = [
|
|
{
|
|
"text": _string(spec.get("postText"), "postText", required=True),
|
|
"media": spec.get("media") if "media" in spec else "__example__",
|
|
"likes": spec.get("likes", 0),
|
|
"reposts": spec.get("reposts", 0),
|
|
}
|
|
]
|
|
if not isinstance(raw_posts, list) or not 1 <= len(raw_posts) <= 10:
|
|
raise renderer.DataValidationError("posts must contain between 1 and 10 objects")
|
|
if not all(isinstance(item, Mapping) for item in raw_posts):
|
|
raise renderer.DataValidationError("posts must contain only objects")
|
|
|
|
data = renderer._read_json(renderer.DEFAULT_DATA_PATH)
|
|
defaults = data["threads"]["items"]
|
|
custom_replies = spec.get("replies", [])
|
|
if not isinstance(custom_replies, list):
|
|
raise renderer.DataValidationError("replies must be an array")
|
|
display_reply_count = _integer(
|
|
spec.get("displayReplyCount"),
|
|
"displayReplyCount",
|
|
default=max(3, len(custom_replies)),
|
|
)
|
|
total_reply_count = _integer(
|
|
spec.get("totalReplyCount"),
|
|
"totalReplyCount",
|
|
default=display_reply_count,
|
|
)
|
|
if total_reply_count < display_reply_count:
|
|
raise renderer.DataValidationError("totalReplyCount cannot be smaller than displayReplyCount")
|
|
|
|
items: list[dict[str, object]] = []
|
|
for index, raw_post in enumerate(raw_posts):
|
|
post = copy.deepcopy(defaults[index % len(defaults)])
|
|
text = _string(
|
|
raw_post.get("text"),
|
|
f"posts[{index}].text",
|
|
required=True,
|
|
)
|
|
_require_chinese(text, f"posts[{index}].text")
|
|
author_name = _string(raw_post.get("authorName"), f"posts[{index}].authorName") or account_name
|
|
post["id"] = f"thread-{index + 1:03d}"
|
|
post["author"] = {
|
|
"name": author_name,
|
|
"handle": _string(raw_post.get("handle"), f"posts[{index}].handle") or handle,
|
|
"initial": author_name[:1],
|
|
"colors": post["author"]["colors"],
|
|
"verified": _boolean(
|
|
raw_post.get("verified"),
|
|
f"posts[{index}].verified",
|
|
default=_boolean(spec.get("verified"), "verified"),
|
|
),
|
|
}
|
|
post["time"] = _string(raw_post.get("time"), f"posts[{index}].time") or "刚刚"
|
|
post["text"] = text
|
|
post["followable"] = _boolean(
|
|
raw_post.get("followable"),
|
|
f"posts[{index}].followable",
|
|
)
|
|
post["quote"] = None
|
|
if "media" in raw_post and raw_post.get("media") != "__example__":
|
|
post["media"] = _media(raw_post.get("media"), f"posts[{index}].media")
|
|
if index == 0:
|
|
replies = _reply_objects(display_reply_count, reply_topics, custom_replies)
|
|
replies_metric = total_reply_count
|
|
else:
|
|
replies = []
|
|
replies_metric = _integer(
|
|
raw_post.get("totalReplyCount"),
|
|
f"posts[{index}].totalReplyCount",
|
|
)
|
|
post["replies"] = replies
|
|
post["metrics"] = {
|
|
"likes": _integer(raw_post.get("likes"), f"posts[{index}].likes"),
|
|
"replies": replies_metric,
|
|
"reposts": _integer(raw_post.get("reposts"), f"posts[{index}].reposts"),
|
|
}
|
|
post["preview"] = f"查看全部 {replies_metric} 条回复"
|
|
items.append(post)
|
|
|
|
data["site"]["pageTitle"] = _string(spec.get("pageTitle"), "pageTitle") or f"Threads · {account_name}"
|
|
data["site"]["description"] = _string(spec.get("description"), "description") or f"{account_name} 的 Threads 中文页面。"
|
|
data["site"]["currentUser"].update(
|
|
{
|
|
"name": account_name,
|
|
"handle": handle,
|
|
"initial": account_name[:1],
|
|
"stats": {
|
|
"following": str(spec.get("following", "0")),
|
|
"followers": str(spec.get("followers", "0")),
|
|
"threads": str(spec.get("threadCount", len(items))),
|
|
},
|
|
}
|
|
)
|
|
data["threads"]["items"] = items
|
|
return data
|
|
|
|
|
|
def _next_paths(output_dir: Path) -> tuple[Path, Path]:
|
|
for index in range(1, 1001):
|
|
stem = "threads-page" if index == 1 else f"threads-page-{index}"
|
|
data_path = output_dir / f"{stem}-data.json"
|
|
html_path = output_dir / f"{stem}.html"
|
|
if not data_path.exists() and not html_path.exists():
|
|
return data_path, html_path
|
|
raise renderer.DataValidationError("Unable to allocate output filenames")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--spec", type=Path, required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
data = build_document(_read_spec(args.spec))
|
|
summary = renderer.validate_document(data)
|
|
rendered = renderer.render_document(
|
|
data,
|
|
renderer.TEMPLATE_PATH.read_text(encoding="utf-8"),
|
|
renderer.RUNTIME_PATH.read_text(encoding="utf-8"),
|
|
)
|
|
data_path, html_path = _next_paths(args.output_dir)
|
|
renderer._write_new(
|
|
data_path,
|
|
json.dumps(data, ensure_ascii=False, indent=2) + "\n",
|
|
)
|
|
renderer._write_new(html_path, rendered)
|
|
except (renderer.DataValidationError, OSError, UnicodeError) as exc:
|
|
renderer._emit({"status": "error", "action": "fast-threads-page", "error": str(exc)})
|
|
return 2
|
|
renderer._emit(
|
|
{
|
|
"status": "ok",
|
|
"action": "fast-threads-page",
|
|
"data": str(data_path),
|
|
"html": str(html_path),
|
|
**summary,
|
|
}
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|