Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
222 changes: 222 additions & 0 deletions scripts/generate_caption.py
Original file line number Diff line number Diff line change
@@ -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())
161 changes: 161 additions & 0 deletions scripts/multi_export.py
Original file line number Diff line number Diff line change
@@ -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 <input.mp4> --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())
Loading
Loading