diff --git a/scripts/generate_caption.py b/scripts/generate_caption.py new file mode 100644 index 0000000..df2449b --- /dev/null +++ b/scripts/generate_caption.py @@ -0,0 +1,222 @@ +#!/usr/bin/env python3 +"""Generate a Xiaohongshu发布文案 (caption note) from a cleaned script. + +Output: a JSON with title, caption_body, tags, and publish-time hint. + - title: ≤18 chars, first 18 chars include 2 keywords (TF-IDF) so + the feed-thumbnail preview surfaces them. + - body: 200-500 chars, emoji every 50-80 chars, repeats target + keyword once per ~300 chars (TF-IDF normalisation). + - tags: 3-6 hashtags. Mix垂直+长尾 (no pure hot-tag spam). + - publish_time_hint: per audience profile. + +This module is rule-based — no LLM dependency. For richer rewriting use +rewrite_script.py first, then run generate_caption.py on the result. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from collections import Counter +from typing import Iterable, List, Optional + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from content_guard import enforce as enforce_platform_rules, HardBlock # noqa: E402 + + +EMOJI_PALETTE = ["📌", "✨", "💡", "🔥", "👇", "✅", "🚀", "📈"] + +# Words that should not become tags +STOPWORDS_ZH = set(""" +的 了 在 是 我 你 他 她 它 我们 你们 他们 这 那 就 也 都 还 又 把 被 让 +不 没 有 会 能 要 想 觉得 因为 所以 但是 然后 而且 就是 这种 那种 +""".split()) + + +def extract_keywords(text: str, *, top_n: int = 8) -> List[str]: + """Return the top-n bigram/short-word candidates, ranked by TF. + + Pure rule-based (no external library). Strips stopwords, keeps 2-6 + char tokens, dedupes. + """ + # Strip markdown headings / punctuation + cleaned = re.sub(r"#+\s+|[*_`]", "", text) + # Split by punctuation/whitespace + tokens = re.split(r"[\s,。!?、;:()\(\),.!?;:\-—\n]+", cleaned) + counts: Counter = Counter() + for tok in tokens: + tok = tok.strip() + if not tok: + continue + if tok in STOPWORDS_ZH: + continue + if len(tok) < 2 or len(tok) > 8: + continue + if re.fullmatch(r"\d+", tok): + continue + counts[tok] += 1 + return [w for w, _ in counts.most_common(top_n)] + + +def synthesize_title(hook_text: Optional[str], keywords: List[str], + *, max_chars: int = 18) -> str: + """Pick a title ≤max_chars that surfaces 2 keywords up front when possible.""" + if hook_text: + title = hook_text.strip() + # If the hook is short enough, use it directly + if 4 <= len(title) <= max_chars: + return title + # Else, truncate at a clean break + if len(title) > max_chars: + cut = title[:max_chars] + # find last punctuation/space to avoid cutting mid-word + for sep in "?!。,、 ": + idx = cut.rfind(sep) + if idx >= 4: + return cut[:idx] + return cut + + if len(keywords) >= 2: + kw1, kw2 = keywords[0], keywords[1] + title = f"{kw1}{kw2}:3个值得知道的事" + if len(title) <= max_chars: + return title + return f"{kw1}{kw2}:值得收藏" + if keywords: + return f"{keywords[0]}:值得收藏" + return "今日分享" + + +def synthesize_body(script_text: str, keywords: List[str], *, + emoji_every_chars: int = 60, + min_chars: int = 200, + max_chars: int = 500) -> str: + """Build the 正文 body. Take the cleaned script content and intersperse + emoji every ~60 chars; cap at max_chars.""" + body = re.sub(r"#+\s+", "", script_text) + body = re.sub(r"\n{2,}", "\n\n", body).strip() + + # Truncate + if len(body) > max_chars: + body = body[:max_chars] + # cut at the last natural break + for sep in "\n。!?": + idx = body.rfind(sep) + if idx >= min_chars: + body = body[:idx + 1] + break + + # Sprinkle emoji + out_chars = [] + last_emoji = 0 + for i, ch in enumerate(body): + out_chars.append(ch) + if (i - last_emoji) >= emoji_every_chars and ch in "。!?\n": + emoji = EMOJI_PALETTE[(i // emoji_every_chars) % len(EMOJI_PALETTE)] + out_chars.append(f" {emoji}") + last_emoji = i + + return "".join(out_chars) + + +def synthesize_tags(keywords: List[str], *, max_tags: int = 6, + min_tags: int = 3) -> List[str]: + """Return 3-6 # tags. Mix垂类keyword + 长尾.""" + tags = [] + for kw in keywords[:max_tags]: + tags.append("#" + kw) + while len(tags) < min_tags and len(keywords) > 0: + # pad with the most popular keyword again, as compound + tags.append("#" + keywords[0]) + return tags[:max_tags] + + +def publish_time_hint(profile: Optional[dict]) -> str: + if not profile: + return "weekday 21:00-22:30" + windows = profile.get("publishing", {}).get("preferred_windows") or [] + if isinstance(windows, list) and windows: + return windows[0] if isinstance(windows[0], str) else str(windows[0]) + return "weekday 21:00-22:30" + + +def generate_caption(script_text: str, *, + hook_text: Optional[str] = None, + profile_name: Optional[str] = None, + strict: bool = True) -> dict: + keywords = extract_keywords(script_text) + title = synthesize_title(hook_text, keywords) + body = synthesize_body(script_text, keywords) + tags = synthesize_tags(keywords) + + profile = None + if profile_name: + try: + from profiles import load_profile + profile = load_profile(profile_name) + except (FileNotFoundError, ImportError): + pass + + payload = { + "title": title, + "caption_body": body, + "tags": tags, + "publish_time_hint": publish_time_hint(profile), + } + + if strict: + # Content-guard the title and body. Tags pre-pended with # are fine. + enforce_platform_rules([title], context="title", strict=True) + enforce_platform_rules([body], context="caption", strict=False) + # ^ caption uses non-strict (warnings only) to leave room for emoji density + + return payload + + +def main() -> int: + p = argparse.ArgumentParser(description="Generate Xiaohongshu caption (note body)") + p.add_argument("--script", required=True, help="Path to clean_script.md") + p.add_argument("--hook", default=None, help="Override hook text used as title") + p.add_argument("--profile", default=None, help="Audience profile (tech_pro, lifestyle)") + p.add_argument("--output", default=None, help="JSON output path; stdout if omitted") + p.add_argument("--no-strict", action="store_true", + help="Skip content-guard rejection (warnings still emitted to stderr)") + args = p.parse_args() + + if not os.path.isfile(args.script): + print(f"script not found: {args.script}", file=sys.stderr) + return 1 + with open(args.script, encoding="utf-8") as f: + script_text = f.read() + + hook_text = args.hook + if not hook_text: + # Try to extract from `## Hook` block + m = re.search(r"^##\s+Hook\s*\n+([^\n#]+)", script_text, re.MULTILINE) + if m: + hook_text = m.group(1).strip() + + try: + payload = generate_caption( + script_text, hook_text=hook_text, + profile_name=args.profile, strict=not args.no_strict, + ) + except HardBlock as exc: + print(f"🚫 Caption rejected by content guard: {exc}", file=sys.stderr) + return 2 + + out_text = json.dumps(payload, ensure_ascii=False, indent=2) + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(out_text) + print(f"✅ caption → {args.output}") + else: + print(out_text) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/multi_export.py b/scripts/multi_export.py new file mode 100644 index 0000000..d20eb28 --- /dev/null +++ b/scripts/multi_export.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Export one source video into three platform-specific deliverables. + +Presets: + xhs — 3:4 (1080×1440) for Xiaohongshu feed. Centre-crop from 9:16 master. + douyin — 9:16 (1080×1920) for Douyin/TikTok. Re-encode if needed. + wxch — 9:16 (1080×1920), capped at 60s for 微信视频号 social sharing. + +Usage: + python3 scripts/multi_export.py --output-dir ./output \\ + --platforms xhs douyin wxch +""" +from __future__ import annotations + +import argparse +import dataclasses +import json +import os +import subprocess +import sys +from typing import List, Optional + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from utils import get_video_info, get_ffmpeg_encode_args # noqa: E402 + + +@dataclasses.dataclass(frozen=True) +class PlatformPreset: + name: str + width: int + height: int + max_duration_seconds: Optional[float] + crf: int # for x264 fallback + audio_bitrate: str + notes: str + + +PRESETS = { + "xhs": PlatformPreset( + name="xhs", width=1080, height=1440, # 3:4 + max_duration_seconds=None, crf=22, audio_bitrate="192k", + notes="Xiaohongshu/RED: 3:4 fills the feed thumbnail (+~40% display area).", + ), + "douyin": PlatformPreset( + name="douyin", width=1080, height=1920, # 9:16 + max_duration_seconds=None, crf=22, audio_bitrate="192k", + notes="Douyin/TikTok: full-screen 9:16.", + ), + "wxch": PlatformPreset( + name="wxch", width=1080, height=1920, # 9:16 ≤60s + max_duration_seconds=60.0, crf=23, audio_bitrate="160k", + notes="WeChat视频号: 9:16, ideally ≤60s for social-graph distribution.", + ), +} + + +def _source_aspect_filter(src_w: int, src_h: int, dst_w: int, dst_h: int) -> str: + """Build an ffmpeg -vf filter that turns src into dst dimensions. + + Rules: + - same aspect: simple scale + - source is taller (9:16 → 3:4): centre-crop top+bottom + - source is wider (16:9 → 9:16): centre-crop sides + """ + src_ratio = src_w / src_h + dst_ratio = dst_w / dst_h + if abs(src_ratio - dst_ratio) < 1e-3: + return f"scale={dst_w}:{dst_h}" + + if src_ratio < dst_ratio: + # source is taller than dst — crop top/bottom or letterbox sides + # we centre-crop the source to dst aspect, then scale + target_h = src_w / dst_ratio + return f"crop=in_w:{int(target_h)}:(in_w-out_w)/2:(in_h-out_h)/2,scale={dst_w}:{dst_h}" + else: + # source is wider than dst — crop sides + target_w = src_h * dst_ratio + return f"crop={int(target_w)}:in_h:(in_w-out_w)/2:(in_h-out_h)/2,scale={dst_w}:{dst_h}" + + +def build_ffmpeg_command(input_path: str, output_path: str, preset: PlatformPreset, + src_w: int, src_h: int, src_duration: float) -> List[str]: + """Build the ffmpeg command for one platform preset.""" + vf = _source_aspect_filter(src_w, src_h, preset.width, preset.height) + cmd: List[str] = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-i", input_path] + + if preset.max_duration_seconds and src_duration > preset.max_duration_seconds: + cmd.extend(["-t", f"{preset.max_duration_seconds:.2f}"]) + + encode = get_ffmpeg_encode_args() + cmd.extend(["-vf", vf]) + cmd.extend(encode) + cmd.extend([ + "-crf", str(preset.crf), + "-c:a", "aac", "-b:a", preset.audio_bitrate, + "-movflags", "+faststart", + output_path, + ]) + return cmd + + +def export_one(input_path: str, output_dir: str, preset: PlatformPreset) -> Optional[str]: + src_duration, src_w, src_h, _fps, _rot = get_video_info(input_path) + + os.makedirs(output_dir, exist_ok=True) + base = os.path.splitext(os.path.basename(input_path))[0] + out_path = os.path.join(output_dir, f"{base}_{preset.name}.mp4") + + cmd = build_ffmpeg_command(input_path, out_path, preset, src_w, src_h, src_duration) + print(f"\n[{preset.name}] {preset.width}×{preset.height} → {out_path}") + print(f" {preset.notes}") + try: + subprocess.run(cmd, check=True) + except subprocess.CalledProcessError as exc: + print(f" ❌ ffmpeg failed (returncode {exc.returncode})", file=sys.stderr) + return None + return out_path + + +def main() -> int: + p = argparse.ArgumentParser(description="Multi-platform export") + p.add_argument("input", help="Master video path (typically 9:16 1080×1920)") + p.add_argument("--output-dir", default="./output") + p.add_argument("--platforms", nargs="+", default=list(PRESETS), + choices=list(PRESETS), + help="Platforms to export (default: all)") + p.add_argument("--dry-run", action="store_true", + help="Print the ffmpeg commands without running") + args = p.parse_args() + + if not os.path.isfile(args.input): + print(f"Input not found: {args.input}", file=sys.stderr) + return 1 + + src_duration, src_w, src_h, _fps, _rot = get_video_info(args.input) + print(f"Source: {src_w}×{src_h}, {src_duration:.2f}s") + + summary: List[dict] = [] + for plat in args.platforms: + preset = PRESETS[plat] + out_path = os.path.join(args.output_dir, f"{os.path.splitext(os.path.basename(args.input))[0]}_{plat}.mp4") + cmd = build_ffmpeg_command(args.input, out_path, preset, src_w, src_h, src_duration) + if args.dry_run: + print(f"\n[{plat}] {' '.join(cmd)}") + summary.append({"platform": plat, "output": out_path, "cmd": cmd}) + else: + result = export_one(args.input, args.output_dir, preset) + summary.append({"platform": plat, "output": result}) + + manifest = os.path.join(args.output_dir, "multi_export_manifest.json") + os.makedirs(args.output_dir, exist_ok=True) + with open(manifest, "w", encoding="utf-8") as f: + json.dump(summary, f, ensure_ascii=False, indent=2) + print(f"\n✅ manifest → {manifest}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/profiles/__init__.py b/scripts/profiles/__init__.py index d7b6519..a7109dd 100644 --- a/scripts/profiles/__init__.py +++ b/scripts/profiles/__init__.py @@ -137,8 +137,23 @@ def load_profile(name: str) -> Dict[str, Any]: def list_profiles() -> List[str]: + """List audience profile names. Skips underscore-prefixed YAML files + (those are companion configs like _fonts.yaml, not audience profiles).""" out = [] for entry in os.listdir(_PROFILES_DIR): - if entry.endswith(".yaml"): + if entry.endswith(".yaml") and not entry.startswith("_"): out.append(entry[:-5]) return sorted(out) + + +def load_fonts_preset(preset: str = "tech_ai") -> dict: + """Load `_fonts.yaml` and return the named preset's font mapping.""" + path = os.path.join(_PROFILES_DIR, "_fonts.yaml") + if not os.path.isfile(path): + raise FileNotFoundError(path) + with open(path, encoding="utf-8") as f: + data = _yaml_safe_load(f.read()) + presets = data.get("presets", {}) + if preset not in presets: + raise KeyError(f"font preset {preset!r} not found; available: {list(presets)}") + return presets[preset] diff --git a/scripts/profiles/_fonts.yaml b/scripts/profiles/_fonts.yaml new file mode 100644 index 0000000..6b14fad --- /dev/null +++ b/scripts/profiles/_fonts.yaml @@ -0,0 +1,35 @@ +# Title-font palette per design preset. +# Each preset names a primary font (used at the largest size) and a fallback +# chain. find_chinese_font() handles auto-download from FONT_CATALOG; this +# YAML just maps "design intent" → font name(s). + +presets: + tech_ai: + label: 科技/AI/工具盘点 + primary: Smiley Sans # 得意黑 + fallbacks: [Source Han Sans SC Heavy, PingFang SC Semibold] + weight: Heavy + + data_review: + label: 数据复盘/榜单/教程干货 + primary: Alimama Shu Hei # 阿里妈妈数黑体 + fallbacks: [Source Han Sans SC Heavy, Smiley Sans] + weight: Heavy + + warm_lifestyle: + label: 生活方式/亲子/温柔向 + primary: Alimama Fang Yuan # 阿里妈妈方圆体 + fallbacks: [PingFang SC Semibold, STHeiti Medium] + weight: Medium + + business_serious: + label: 商务/职场/严肃论点 + primary: Source Han Sans SC Heavy # 思源黑体 + fallbacks: [PingFang SC Semibold, STHeiti Medium] + weight: Heavy + + food_youth: + label: 美食/穿搭/学生向 + primary: Maoken Naizhi # 奶酪体 + fallbacks: [ZCOOL XiaoWei, PingFang SC Semibold] + weight: Medium diff --git a/tests/test_generate_caption.py b/tests/test_generate_caption.py new file mode 100644 index 0000000..e06679b --- /dev/null +++ b/tests/test_generate_caption.py @@ -0,0 +1,102 @@ +"""generate_caption — title/body/tag synthesis + content guard integration.""" +import json +import os +import subprocess +import sys + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(REPO, "scripts")) + +from generate_caption import ( # noqa: E402 + extract_keywords, synthesize_title, synthesize_body, + synthesize_tags, generate_caption, +) + + +def test_extract_keywords_finds_top_terms(): + text = """ + AI 失业焦虑 真的让我反思了很多。 + AI 让我接到更多客户,也让我交付更快。 + AI 是工具,不是替代我们的对手。 + """ + kws = extract_keywords(text, top_n=5) + assert len(kws) >= 1 + assert "AI" in kws or "失业焦虑" in kws + + +def test_synthesize_title_uses_hook_when_short_enough(): + title = synthesize_title("AI失业焦虑?我看到更多机会", ["AI", "焦虑"]) + assert title == "AI失业焦虑?我看到更多机会" + + +def test_synthesize_title_truncates_long_hook(): + long_hook = "我用 AI 做了一件让我自己都不敢相信的事情而且坚持了 6 个月" + title = synthesize_title(long_hook, []) + assert len(title) <= 18 + + +def test_synthesize_title_falls_back_to_keywords(): + title = synthesize_title(None, ["AI", "失业焦虑"]) + assert "AI" in title or "失业焦虑" in title + + +def test_synthesize_body_within_limits(): + long_script = "AI 真的能帮我们。" * 200 + body = synthesize_body(long_script, ["AI"]) + assert 100 <= len(body) <= 700 # account for emoji insertion + + +def test_synthesize_body_contains_emoji(): + body = synthesize_body("AI 帮我提效。" * 30, ["AI"]) + # at least one emoji from the palette should appear + palette = "📌✨💡🔥👇✅🚀📈" + assert any(c in body for c in palette) + + +def test_synthesize_tags_within_count(): + tags = synthesize_tags(["AI", "失业", "创业", "工作"], min_tags=3, max_tags=5) + assert 3 <= len(tags) <= 5 + assert all(t.startswith("#") for t in tags) + + +def test_generate_caption_full_round_trip(tmp_path): + script = (tmp_path / "clean.md") + script.write_text( + "# Clean Script\n\n## Hook\n" + "AI失业焦虑?我看到更多机会\n\n" + "## Pain\n之前我也焦虑,怕 AI 替代人。\n\n" + "## Turn\n但客户找我的次数没减少。\n\n" + "## Value\nAI 让我交付更快。\n\n" + "## CTA\n你是焦虑还是抓住机会?\n" + ) + payload = generate_caption(open(script).read()) + assert "title" in payload and 4 <= len(payload["title"]) <= 25 + assert "caption_body" in payload + assert isinstance(payload["tags"], list) + assert 3 <= len(payload["tags"]) <= 6 + assert payload["publish_time_hint"] + + +def test_generate_caption_rejects_hardblock_title(): + # script with a diversion phrase that becomes the hook → title trips guard + script = "## Hook\n加微信 wx123\n## Value\nx" + with pytest.raises(Exception): # HardBlock or similar + generate_caption(script) + + +def test_cli(tmp_path): + script = tmp_path / "clean.md" + script.write_text( + "## Hook\nAI失业焦虑\n## Pain\n我担心被替代\n## CTA\n你怎么看?\n" + ) + out_path = tmp_path / "caption.json" + out = subprocess.run( + [sys.executable, os.path.join(REPO, "scripts/generate_caption.py"), + "--script", str(script), "--output", str(out_path)], + capture_output=True, text=True, + ) + assert out.returncode == 0, f"stderr: {out.stderr}" + payload = json.loads(out_path.read_text()) + assert "title" in payload diff --git a/tests/test_multi_export.py b/tests/test_multi_export.py new file mode 100644 index 0000000..aa78ff1 --- /dev/null +++ b/tests/test_multi_export.py @@ -0,0 +1,69 @@ +"""multi_export — aspect-ratio filter selection + command-build smoke tests. + +We don't actually run ffmpeg in tests (large videos, slow). Instead we +verify the constructed commands and aspect filter strings. +""" +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")) + +from multi_export import PRESETS, _source_aspect_filter, build_ffmpeg_command # noqa: E402 + + +def test_three_presets_present(): + assert set(PRESETS.keys()) >= {"xhs", "douyin", "wxch"} + + +def test_xhs_is_3_4(): + p = PRESETS["xhs"] + assert (p.width, p.height) == (1080, 1440) + + +def test_douyin_is_9_16(): + p = PRESETS["douyin"] + assert (p.width, p.height) == (1080, 1920) + + +def test_wxch_caps_duration_to_60s(): + assert PRESETS["wxch"].max_duration_seconds == 60.0 + + +def test_aspect_filter_same_ratio(): + # 9:16 → 9:16: just scale + f = _source_aspect_filter(1080, 1920, 1080, 1920) + assert f.startswith("scale=") + + +def test_aspect_filter_9_16_to_3_4_crops_top_bottom(): + f = _source_aspect_filter(1080, 1920, 1080, 1440) + assert "crop=" in f and "scale=1080:1440" in f + + +def test_aspect_filter_16_9_to_9_16_crops_sides(): + f = _source_aspect_filter(1920, 1080, 1080, 1920) + assert "crop=" in f and "scale=1080:1920" in f + + +def test_build_command_includes_t_when_capped(): + """wxch caps duration → -t should appear in the command.""" + cmd = build_ffmpeg_command("in.mp4", "out.mp4", PRESETS["wxch"], + src_w=1080, src_h=1920, src_duration=120.0) + assert "-t" in cmd + t_idx = cmd.index("-t") + assert cmd[t_idx + 1] == "60.00" + + +def test_build_command_no_t_when_short(): + """When source is already shorter than the cap, no -t is added.""" + cmd = build_ffmpeg_command("in.mp4", "out.mp4", PRESETS["wxch"], + src_w=1080, src_h=1920, src_duration=30.0) + assert "-t" not in cmd + + +def test_build_command_uses_faststart(): + cmd = build_ffmpeg_command("in.mp4", "out.mp4", PRESETS["xhs"], + src_w=1080, src_h=1920, src_duration=60.0) + assert "-movflags" in cmd + idx = cmd.index("-movflags") + assert cmd[idx + 1] == "+faststart" diff --git a/tests/test_profiles.py b/tests/test_profiles.py index 211c387..bab6d68 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -34,6 +34,14 @@ def test_missing_profile_raises(): load_profile("nonexistent_profile_xyz") +def test_fonts_presets_loadable(): + from profiles import load_fonts_preset + tech = load_fonts_preset("tech_ai") + assert tech["primary"] == "Smiley Sans" + business = load_fonts_preset("business_serious") + assert "Source Han Sans" in business["primary"] + + def test_required_fields_present(): """Every profile must define duration, cut, subtitle, audio, aspect.""" for name in list_profiles():