Bggg Tiktok Cut

.DS_Store
__pycache__/
*.pyc
projects/*
!projects/.gitkeep

{
  "skill_name": "bggg-tiktok-cut",
  "evals": [
    {
      "id": 1,
      "prompt": "这里有一个 Seedance 生成的横屏产品视频和一段 BGM，帮我剪成 30 秒 TikTok 竖屏广告，加前 3 秒 hook、大字字幕和轻 BGM。",
      "expected_output": "Creates a project folder, probes source media, writes an edit_plan.json with 1080x1920 target, uses blur-bg or cover reframe, renders final_tiktok.mp4, and reports paths plus verification.",
      "files": [],
      "expectations": [
        "The plan uses TikTok vertical target dimensions.",
        "The workflow includes media probing and frame inspection before final render.",
        "The final output path and render_report.json path are provided."
      ]
    },
    {
      "id": 2,
      "prompt": "把这个 AI 视频素材文件夹批量做成 5 条 TikTok 短视频，每条不同 hook，保留项目文件和字幕文件。",
      "expected_output": "Creates separate plans or project folders for each output, keeps artifacts organized, renders named MP4s, and preserves captions/metadata.",
      "files": [],
      "expectations": [
        "The workflow does not overwrite outputs from previous variants.",
        "Each variant has its own hook or plan.",
        "Captions and render reports are retained."
      ]
    },
    {
      "id": 3,
      "prompt": "这个 AI 生成的视频没有旁白，只有画面。请做一个适合 TikTok 的视觉向短片，加标题、节奏字幕和音乐。",
      "expected_output": "Uses visual inspection rather than transcript-first logic, creates captions/overlays from the supplied concept, adds BGM, and renders a TikTok-safe vertical video.",
      "files": [],
      "expectations": [
        "The skill handles no-speech source material without requiring Whisper.",
        "The edit plan includes overlays or captions that carry the story.",
        "The output includes audio through BGM or a safe fallback."
      ]
    }
  ]
}

#!/usr/bin/env python3
"""Create a clean TikTok cut project folder."""

from __future__ import annotations

import argparse
import json
import shutil
from datetime import datetime
from pathlib import Path

from media_common import VIDEO_EXTS, AUDIO_EXTS, IMAGE_EXTS, media_info, slugify


DEFAULT_PLAN = {
    "version": 1,
    "project": {
        "title": "TikTok cut",
        "platform": "tiktok",
        "target": {"width": 1080, "height": 1920, "fps": 30},
        "notes": "Replace the clips, captions, overlays, and bgm fields before rendering.",
    },
    "settings": {
        "fit": "blur-bg",
        "grade": "punch",
        "caption_style": "tiktok-bold",
        "caption_safe_zone": "middle",
        "voice_volume": 1.0,
        "output_name": "final_tiktok.mp4",
    },
    "clips": [
        {
            "source": "raw/example.mp4",
            "start": 0.0,
            "end": 5.0,
            "speed": 1.0,
            "fit": "blur-bg",
            "anchor": "center",
            "label": "HOOK",
        }
    ],
    "captions": [
        {"start": 0.0, "end": 2.0, "text": "替换成大字字幕", "style": "tiktok-bold"}
    ],
    "overlays": [
        {"start": 0.0, "end": 2.2, "text": "3秒钩子", "position": "top", "style": "hook"}
    ],
    "bgm": {
        "path": "",
        "volume": 0.12,
        "start": 0.0,
        "fade_in": 0.3,
        "fade_out": 0.8,
    },
    "export": {
        "crf": 20,
        "preset": "fast",
        "audio_bitrate": "192k",
        "faststart": True,
    },
}


def copy_inputs(inputs: list[Path], project_dir: Path) -> list[dict]:
    copied: list[dict] = []
    raw_dir = project_dir / "raw"
    bgm_dir = project_dir / "assets" / "bgm"
    image_dir = project_dir / "assets" / "images"

    for src in inputs:
        src = src.expanduser().resolve()
        if not src.exists():
            raise SystemExit(f"Input not found: {src}")
        suffix = src.suffix.lower()
        if suffix in VIDEO_EXTS:
            dest_dir = raw_dir
        elif suffix in AUDIO_EXTS:
            dest_dir = bgm_dir
        elif suffix in IMAGE_EXTS:
            dest_dir = image_dir
        else:
            dest_dir = project_dir / "imports"
        dest_dir.mkdir(parents=True, exist_ok=True)
        dest = dest_dir / src.name
        if dest.exists():
            stem = dest.stem
            counter = 2
            while dest.exists():
                dest = dest_dir / f"{stem}-{counter}{dest.suffix}"
                counter += 1
        shutil.copy2(src, dest)
        record = {"source": str(src), "copied_to": str(dest.relative_to(project_dir))}
        if suffix in VIDEO_EXTS | AUDIO_EXTS:
            try:
                record["media"] = media_info(dest)
            except Exception as exc:
                record["probe_error"] = str(exc)
        copied.append(record)
    return copied


def main() -> None:
    parser = argparse.ArgumentParser(description="Initialize a bggg-tiktok-cut project.")
    parser.add_argument("project_dir", type=Path, help="Destination project directory.")
    parser.add_argument("--name", help="Human-readable project name.")
    parser.add_argument("--inputs", nargs="*", type=Path, default=[], help="Files to copy into the project.")
    parser.add_argument("--force", action="store_true", help="Allow using an existing directory.")
    args = parser.parse_args()

    project_dir = args.project_dir.expanduser().resolve()
    if project_dir.exists() and any(project_dir.iterdir()) and not args.force:
        raise SystemExit(f"Project directory is not empty: {project_dir}. Use --force to reuse it.")

    for rel in [
        "raw",
        "audio",
        "transcripts",
        "plans",
        "captions",
        "assets/bgm",
        "assets/images",
        "assets/overlays",
        "renders",
        "diagnostics/frames",
        "metadata",
        "exports",
    ]:
        (project_dir / rel).mkdir(parents=True, exist_ok=True)

    title = args.name or project_dir.name
    now = datetime.now().astimezone().isoformat(timespec="seconds")
    copied = copy_inputs(args.inputs, project_dir) if args.inputs else []

    manifest = {
        "name": title,
        "slug": slugify(title),
        "created_at": now,
        "project_dir": str(project_dir),
        "inputs": copied,
        "outputs": {
            "default_plan": "plans/edit_plan.template.json",
            "final": "renders/final_tiktok.mp4",
        },
    }
    (project_dir / "metadata" / "project.json").write_text(
        json.dumps(manifest, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    plan = dict(DEFAULT_PLAN)
    plan["project"] = dict(DEFAULT_PLAN["project"])
    plan["project"]["title"] = title
    if copied:
        video_inputs = [c for c in copied if c.get("copied_to", "").startswith("raw/")]
        if video_inputs:
            first = video_inputs[0]
            dur = first.get("media", {}).get("duration", 5.0)
            plan["clips"] = [
                {
                    "source": first["copied_to"],
                    "start": 0.0,
                    "end": min(float(dur or 5.0), 8.0),
                    "speed": 1.0,
                    "fit": "blur-bg",
                    "anchor": "center",
                    "label": "HOOK",
                }
            ]
    (project_dir / "plans" / "edit_plan.template.json").write_text(
        json.dumps(plan, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    print(json.dumps(manifest, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate a conservative starter edit plan from a project folder."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from media_common import VIDEO_EXTS, media_info


def main() -> None:
    parser = argparse.ArgumentParser(description="Create a starter TikTok edit plan from raw videos.")
    parser.add_argument("project_dir", type=Path, help="Project directory created by init_project.py.")
    parser.add_argument("--title", default="TikTok cut", help="Title used for the hook overlay.")
    parser.add_argument("--target-seconds", type=float, default=30.0, help="Maximum output duration.")
    parser.add_argument("--each-max", type=float, default=8.0, help="Max seconds taken from each source clip.")
    parser.add_argument("--fit", default="blur-bg", choices=["blur-bg", "cover", "contain"])
    parser.add_argument("--grade", default="punch", choices=["none", "neutral", "punch", "warm", "soft"])
    parser.add_argument("--out", type=Path, help="Output plan path.")
    args = parser.parse_args()

    project_dir = args.project_dir.expanduser().resolve()
    raw_dir = project_dir / "raw"
    files = [p for p in sorted(raw_dir.glob("*")) if p.suffix.lower() in VIDEO_EXTS]
    if not files:
        raise SystemExit(f"No video files found in {raw_dir}")

    clips = []
    remaining = args.target_seconds
    for path in files:
        if remaining <= 0:
            break
        info = media_info(path)
        duration = float(info.get("duration") or 0)
        if duration <= 0:
            continue
        take = min(duration, args.each_max, remaining)
        clips.append(
            {
                "source": str(path.relative_to(project_dir)),
                "start": 0.0,
                "end": round(take, 3),
                "speed": 1.0,
                "fit": args.fit,
                "anchor": "center",
                "label": "BEAT",
            }
        )
        remaining -= take

    if not clips:
        raise SystemExit("No usable video clips found.")

    first_caption_end = min(2.8, sum(c["end"] - c["start"] for c in clips))
    plan = {
        "version": 1,
        "project": {
            "title": args.title,
            "platform": "tiktok",
            "target": {"width": 1080, "height": 1920, "fps": 30},
        },
        "settings": {
            "fit": args.fit,
            "grade": args.grade,
            "caption_style": "tiktok-bold",
            "voice_volume": 1.0,
            "output_name": "final_tiktok.mp4",
        },
        "clips": clips,
        "captions": [],
        "overlays": [
            {
                "start": 0.0,
                "end": round(first_caption_end, 3),
                "text": args.title,
                "position": "top",
                "style": "hook",
            }
        ],
        "bgm": {"path": "", "volume": 0.12, "start": 0.0, "fade_in": 0.3, "fade_out": 0.8},
        "export": {"crf": 20, "preset": "fast", "audio_bitrate": "192k", "faststart": True},
    }

    out_path = args.out.expanduser().resolve() if args.out else project_dir / "plans" / "edit_plan.auto.json"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(plan, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    print(json.dumps({"plan": str(out_path), "clips": len(clips)}, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Shared helpers for bggg-tiktok-cut scripts."""

from __future__ import annotations

import json
import math
import re
import shutil
import subprocess
import sys
from pathlib import Path


VIDEO_EXTS = {".mp4", ".mov", ".m4v", ".mkv", ".webm", ".avi"}
AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"}
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"}


def run(cmd: list[str], *, capture: bool = False, quiet: bool = False) -> subprocess.CompletedProcess[str]:
    if not quiet:
        preview = " ".join(str(part) for part in cmd[:10])
        if len(cmd) > 10:
            preview += " ..."
        print(f"$ {preview}", file=sys.stderr)
    return subprocess.run(
        cmd,
        check=True,
        text=True,
        capture_output=capture,
    )


def require_binary(name: str) -> None:
    if shutil.which(name) is None:
        raise SystemExit(f"Missing required binary: {name}. Install it and retry.")


def slugify(value: str, fallback: str = "tiktok-cut") -> str:
    value = value.strip().lower()
    value = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "-", value)
    value = re.sub(r"-+", "-", value).strip("-")
    return value or fallback


def ffprobe_json(path: Path) -> dict:
    require_binary("ffprobe")
    result = run(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format:stream",
            "-of",
            "json",
            str(path),
        ],
        capture=True,
        quiet=True,
    )
    return json.loads(result.stdout)


def parse_rate(rate: str | None) -> float | None:
    if not rate or rate == "0/0":
        return None
    if "/" in rate:
        num, den = rate.split("/", 1)
        try:
            den_f = float(den)
            if den_f == 0:
                return None
            return float(num) / den_f
        except ValueError:
            return None
    try:
        return float(rate)
    except ValueError:
        return None


def media_info(path: Path) -> dict:
    path = path.resolve()
    data = ffprobe_json(path)
    streams = data.get("streams", [])
    v_streams = [s for s in streams if s.get("codec_type") == "video"]
    a_streams = [s for s in streams if s.get("codec_type") == "audio"]
    fmt = data.get("format", {})
    duration = fmt.get("duration")
    if duration is None and v_streams:
        duration = v_streams[0].get("duration")
    try:
        duration_f = float(duration) if duration is not None else 0.0
    except ValueError:
        duration_f = 0.0

    video = v_streams[0] if v_streams else {}
    fps = parse_rate(video.get("avg_frame_rate")) or parse_rate(video.get("r_frame_rate"))
    width = int(video.get("width") or 0)
    height = int(video.get("height") or 0)
    rotation = video.get("rotation")
    if rotation is None:
        for side in video.get("side_data_list", []) or []:
            if "rotation" in side:
                rotation = side.get("rotation")
                break

    return {
        "path": str(path),
        "name": path.name,
        "suffix": path.suffix.lower(),
        "duration": round(duration_f, 3),
        "width": width,
        "height": height,
        "fps": round(fps, 3) if fps else None,
        "has_video": bool(v_streams),
        "has_audio": bool(a_streams),
        "video_codec": video.get("codec_name"),
        "audio_codec": a_streams[0].get("codec_name") if a_streams else None,
        "color_transfer": video.get("color_transfer"),
        "pix_fmt": video.get("pix_fmt"),
        "rotation": rotation,
        "aspect_ratio": round(width / height, 4) if width and height else None,
        "bit_rate": int(fmt.get("bit_rate")) if str(fmt.get("bit_rate", "")).isdigit() else None,
    }


def list_media_files(path: Path) -> list[Path]:
    if path.is_file():
        return [path]
    files: list[Path] = []
    for child in sorted(path.rglob("*")):
        if child.is_file() and child.suffix.lower() in VIDEO_EXTS | AUDIO_EXTS | IMAGE_EXTS:
            files.append(child)
    return files


def seconds_at_percent(duration: float, percent: float) -> float:
    if duration <= 0:
        return 0.0
    return max(0.0, min(duration - 0.1, duration * percent))


def clamp(value: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, value))


def atempo_chain(speed: float) -> str:
    """Build an ffmpeg atempo chain. atempo supports 0.5..100 in FFmpeg 8,
    but the 0.5..2.0 chain remains portable across older installs."""
    if math.isclose(speed, 1.0, rel_tol=0.001):
        return ""
    if speed <= 0:
        raise ValueError("speed must be positive")
    factors: list[float] = []
    remaining = speed
    while remaining > 2.0:
        factors.append(2.0)
        remaining /= 2.0
    while remaining < 0.5:
        factors.append(0.5)
        remaining /= 0.5
    factors.append(remaining)
    return ",".join(f"atempo={f:.6g}" for f in factors)

#!/usr/bin/env python3
"""Probe media and optionally export diagnostic frames for editing decisions."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from media_common import list_media_files, media_info, require_binary, run, seconds_at_percent


def sample_frames(video: Path, info: dict, out_dir: Path, count: int) -> list[str]:
    if not info.get("has_video") or count <= 0:
        return []
    require_binary("ffmpeg")
    out_dir.mkdir(parents=True, exist_ok=True)
    duration = float(info.get("duration") or 0)
    if count == 1:
        percents = [0.5]
    else:
        percents = [0.08 + (0.84 * i / (count - 1)) for i in range(count)]
    outputs: list[str] = []
    for index, pct in enumerate(percents, 1):
        ts = seconds_at_percent(duration, pct)
        out = out_dir / f"{video.stem}_{index:02d}_{ts:.2f}s.jpg"
        run(
            [
                "ffmpeg",
                "-y",
                "-ss",
                f"{ts:.3f}",
                "-i",
                str(video),
                "-frames:v",
                "1",
                "-q:v",
                "2",
                "-update",
                "1",
                str(out),
            ],
            quiet=True,
        )
        outputs.append(str(out))
    return outputs


def main() -> None:
    parser = argparse.ArgumentParser(description="Probe video/audio/image files for a TikTok edit.")
    parser.add_argument("paths", nargs="+", type=Path, help="Media files or directories.")
    parser.add_argument("--out", type=Path, help="Write JSON inventory to this path.")
    parser.add_argument("--frames-dir", type=Path, help="Export diagnostic JPG frames here.")
    parser.add_argument("--sample-count", type=int, default=5, help="Frames per video when --frames-dir is set.")
    args = parser.parse_args()

    files: list[Path] = []
    for path in args.paths:
        files.extend(list_media_files(path.expanduser()))
    if not files:
        raise SystemExit("No media files found.")

    inventory = {"files": [], "summary": {"count": len(files), "total_duration": 0.0}}
    for file_path in files:
        info = media_info(file_path)
        inventory["summary"]["total_duration"] += float(info.get("duration") or 0)
        if args.frames_dir and info.get("has_video"):
            frame_dir = args.frames_dir.expanduser().resolve() / file_path.stem
            info["diagnostic_frames"] = sample_frames(file_path, info, frame_dir, args.sample_count)
        inventory["files"].append(info)
    inventory["summary"]["total_duration"] = round(inventory["summary"]["total_duration"], 3)

    text = json.dumps(inventory, indent=2, ensure_ascii=False) + "\n"
    if args.out:
        args.out.expanduser().resolve().parent.mkdir(parents=True, exist_ok=True)
        args.out.expanduser().resolve().write_text(text, encoding="utf-8")
    print(text, end="")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Render a TikTok-ready vertical cut from a JSON edit plan.

The renderer intentionally keeps the creative decisions in the plan and makes
the mechanical parts deterministic: per-clip normalization, safe-zone captions,
BGM mix, optional watermark, and final 1080x1920 export.
"""

from __future__ import annotations

import argparse
import json
import math
import re
import shutil
from pathlib import Path

from media_common import atempo_chain, media_info, require_binary, run


GRADE_FILTERS = {
    "none": "",
    "neutral": "eq=contrast=1.03:saturation=1.02",
    "punch": "eq=contrast=1.08:saturation=1.12:gamma=1.0,unsharp=5:5:0.35:3:3:0.12",
    "warm": "eq=contrast=1.06:saturation=1.08:gamma_r=1.03:gamma_b=0.97",
    "soft": "eq=contrast=0.98:saturation=0.96,unsharp=3:3:0.18",
}


def load_json(path: Path) -> dict:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Invalid JSON in {path}: {exc}") from exc


def resolve_project_root(plan_path: Path, explicit: Path | None) -> Path:
    if explicit:
        return explicit.expanduser().resolve()
    if plan_path.parent.name == "plans":
        return plan_path.parent.parent.resolve()
    return plan_path.parent.resolve()


def resolve_path(value: str | None, project_root: Path) -> Path | None:
    if not value:
        return None
    path = Path(value).expanduser()
    if path.is_absolute():
        return path.resolve()
    return (project_root / path).resolve()


def ass_time(seconds: float) -> str:
    seconds = max(0.0, seconds)
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = seconds - h * 3600 - m * 60
    return f"{h}:{m:02d}:{s:05.2f}"


def ass_escape(text: str) -> str:
    return (
        str(text)
        .replace("{", "\\{")
        .replace("}", "\\}")
        .replace("\r\n", "\n")
        .replace("\n", "\\N")
    )


def ffmpeg_filter_path(path: Path) -> str:
    text = str(path.resolve())
    return text.replace("\\", "\\\\").replace("'", "\\'")


def is_cjk(text: str) -> bool:
    return bool(re.search(r"[\u3400-\u9fff]", text))


def wrap_caption(text: str, max_chars: int = 18) -> str:
    text = re.sub(r"\s+", " ", str(text)).strip()
    if not text:
        return ""
    if "\\N" in text or "\n" in text:
        return text.replace("\n", "\\N")
    if is_cjk(text):
        chunks = [text[i : i + max_chars] for i in range(0, len(text), max_chars)]
        return "\\N".join(chunks[:2])
    words = text.split()
    lines: list[str] = []
    current: list[str] = []
    for word in words:
        candidate = " ".join(current + [word])
        if len(candidate) > max_chars and current:
            lines.append(" ".join(current))
            current = [word]
        else:
            current.append(word)
    if current:
        lines.append(" ".join(current))
    return "\\N".join(lines[:2])


def parse_srt_timestamp(value: str) -> float:
    hms, ms = value.strip().replace(".", ",").split(",", 1)
    h, m, s = [int(part) for part in hms.split(":")]
    return h * 3600 + m * 60 + s + int(ms[:3].ljust(3, "0")) / 1000


def parse_srt(path: Path) -> list[dict]:
    text = path.read_text(encoding="utf-8-sig")
    blocks = re.split(r"\n\s*\n", text.strip())
    captions: list[dict] = []
    for block in blocks:
        lines = [line.strip("\ufeff") for line in block.splitlines() if line.strip()]
        if not lines:
            continue
        if re.fullmatch(r"\d+", lines[0]):
            lines = lines[1:]
        if not lines or "-->" not in lines[0]:
            continue
        start_s, end_s = [part.strip() for part in lines[0].split("-->", 1)]
        captions.append(
            {
                "start": parse_srt_timestamp(start_s),
                "end": parse_srt_timestamp(end_s),
                "text": " ".join(lines[1:]).strip(),
            }
        )
    return captions


def grade_filter(name_or_filter: str | None) -> str:
    if not name_or_filter:
        return GRADE_FILTERS["punch"]
    return GRADE_FILTERS.get(name_or_filter, name_or_filter)


def crop_anchor(anchor: str) -> tuple[str, str]:
    anchor = (anchor or "center").lower()
    if "left" in anchor:
        x = "0"
    elif "right" in anchor:
        x = "iw-ow"
    else:
        x = "(iw-ow)/2"
    if "top" in anchor:
        y = "0"
    elif "bottom" in anchor:
        y = "ih-oh"
    else:
        y = "(ih-oh)/2"
    return x, y


def video_fit_filter(width: int, height: int, fit: str, anchor: str) -> str:
    fit = (fit or "blur-bg").lower()
    x, y = crop_anchor(anchor)
    cover = (
        f"scale={width}:{height}:force_original_aspect_ratio=increase,"
        f"crop={width}:{height}:{x}:{y},setsar=1"
    )
    contain = (
        f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
        f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1"
    )
    if fit == "cover":
        return cover
    if fit == "contain":
        return contain
    if fit == "blur-bg":
        fg = f"scale={width}:{height}:force_original_aspect_ratio=decrease,setsar=1"
        return (
            f"split=2[vbg][vfg];"
            f"[vbg]{cover},gblur=sigma=32,eq=saturation=0.82:brightness=-0.04[bg];"
            f"[vfg]{fg}[fg];"
            f"[bg][fg]overlay=(W-w)/2:(H-h)/2:format=auto,setsar=1"
        )
    raise SystemExit(f"Unknown fit mode '{fit}'. Use cover, contain, or blur-bg.")


def normalize_clips(plan: dict) -> list[dict]:
    clips = plan.get("clips") or []
    if not clips:
        raise SystemExit("Plan has no clips.")
    normalized: list[dict] = []
    output_cursor = 0.0
    for index, clip in enumerate(clips):
        start = float(clip.get("start", 0.0))
        end = float(clip.get("end", 0.0))
        if end <= start:
            raise SystemExit(f"Clip #{index + 1} has invalid start/end: {start} -> {end}")
        speed = float(clip.get("speed", 1.0))
        if speed <= 0:
            raise SystemExit(f"Clip #{index + 1} speed must be positive.")
        input_duration = end - start
        output_duration = input_duration / speed
        item = dict(clip)
        item.update(
            {
                "index": index,
                "start": start,
                "end": end,
                "speed": speed,
                "input_duration": input_duration,
                "output_start": output_cursor,
                "output_end": output_cursor + output_duration,
                "output_duration": output_duration,
            }
        )
        normalized.append(item)
        output_cursor += output_duration
    return normalized


def render_segment(
    clip: dict,
    source_path: Path,
    out_path: Path,
    target: dict,
    defaults: dict,
    export: dict,
    source_info: dict,
) -> None:
    width = int(target.get("width", 1080))
    height = int(target.get("height", 1920))
    fps = int(target.get("fps", 30))
    fit = clip.get("fit") or defaults.get("fit") or "blur-bg"
    anchor = clip.get("anchor") or defaults.get("anchor") or "center"
    speed = float(clip["speed"])
    output_duration = float(clip["output_duration"])
    fade = min(0.03, max(0.0, output_duration / 4))

    vf = video_fit_filter(width, height, fit, anchor)
    g = grade_filter(clip.get("grade") or defaults.get("grade"))
    if g:
        vf = f"{vf},{g}"
    vf = f"[0:v]setpts=PTS-STARTPTS,setpts=PTS/{speed:.8f},{vf},fps={fps},format=yuv420p[vout]"

    cmd = [
        "ffmpeg",
        "-y",
        "-ss",
        f"{float(clip['start']):.3f}",
        "-i",
        str(source_path),
        "-t",
        f"{float(clip['input_duration']):.3f}",
    ]

    audio_source_label = "0:a"
    if not source_info.get("has_audio"):
        cmd.extend(
            [
                "-f",
                "lavfi",
                "-t",
                f"{output_duration:.3f}",
                "-i",
                "anullsrc=channel_layout=stereo:sample_rate=48000",
            ]
        )
        audio_source_label = "1:a"

    audio_filters = [f"[{audio_source_label}]asetpts=PTS-STARTPTS"]
    tempo = atempo_chain(speed)
    if tempo:
        audio_filters.append(tempo)
    clip_volume = float(clip.get("volume", defaults.get("voice_volume", 1.0)))
    audio_filters.append(f"volume={clip_volume:.4f}")
    if fade > 0:
        fade_out_start = max(0.0, output_duration - fade)
        audio_filters.append(f"afade=t=in:st=0:d={fade:.3f}")
        audio_filters.append(f"afade=t=out:st={fade_out_start:.3f}:d={fade:.3f}")
    audio_filters.append("aresample=48000")
    audio_filters.append("aformat=channel_layouts=stereo")
    af = ",".join(audio_filters) + "[aout]"
    filter_complex = f"{vf};{af}"

    cmd.extend(
        [
            "-filter_complex",
            filter_complex,
            "-map",
            "[vout]",
            "-map",
            "[aout]",
            "-c:v",
            "libx264",
            "-preset",
            str(export.get("preset", "fast")),
            "-crf",
            str(export.get("crf", 20)),
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            str(export.get("audio_bitrate", "192k")),
            "-ar",
            "48000",
            "-shortest",
            str(out_path),
        ]
    )
    run(cmd)


def concat_segments(paths: list[Path], out_path: Path, work_dir: Path) -> None:
    list_path = work_dir / "concat.txt"
    list_path.write_text("".join(f"file '{path.resolve()}'\n" for path in paths), encoding="utf-8")
    run(["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(list_path), "-c", "copy", str(out_path)])


def style_for_event(event: dict, fallback: str) -> str:
    style = (event.get("style") or fallback or "caption").lower()
    if style in {"hook", "top", "headline"}:
        return "Hook"
    if style in {"badge", "center"}:
        return "Badge"
    if style in {"minimal", "clean"}:
        return "Minimal"
    return "Caption"


def collect_caption_events(plan: dict, clips: list[dict], project_root: Path) -> list[dict]:
    events: list[dict] = []
    captions_file = resolve_path(plan.get("captions_file"), project_root)
    if captions_file and captions_file.exists():
        events.extend(parse_srt(captions_file))
    events.extend(plan.get("captions") or [])

    for clip in clips:
        if clip.get("caption"):
            events.append(
                {
                    "start": clip["output_start"],
                    "end": min(clip["output_end"], clip["output_start"] + 4.0),
                    "text": clip["caption"],
                    "style": clip.get("caption_style", "tiktok-bold"),
                }
            )

    overlays = []
    for overlay in plan.get("overlays") or []:
        item = dict(overlay)
        item.setdefault("style", "hook")
        overlays.append(item)
    events.extend(overlays)
    return events


def write_ass(plan: dict, clips: list[dict], project_root: Path, out_path: Path, total_duration: float) -> Path | None:
    settings = plan.get("settings") or {}
    events = collect_caption_events(plan, clips, project_root)
    events = [e for e in events if e.get("text") and float(e.get("end", 0)) > float(e.get("start", 0))]
    if not events:
        return None

    target = (plan.get("project") or {}).get("target") or {}
    width = int(target.get("width", 1080))
    height = int(target.get("height", 1920))
    font = settings.get("font", "Arial")
    caption_size = int(settings.get("caption_font_size", 86))
    hook_size = int(settings.get("hook_font_size", 92))
    margin_lr = 70
    caption_margin = int(height * 0.30)
    hook_margin = int(height * 0.08)

    header = f"""[Script Info]
ScriptType: v4.00+
PlayResX: {width}
PlayResY: {height}
WrapStyle: 0
ScaledBorderAndShadow: yes

[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Caption,{font},{caption_size},&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,1,0,0,0,100,100,0,0,1,7,2,2,{margin_lr},{margin_lr},{caption_margin},1
Style: Hook,{font},{hook_size},&H0000FFFF,&H00FFFFFF,&H00000000,&H8A000000,1,0,0,0,100,100,0,0,1,8,2,8,{margin_lr},{margin_lr},{hook_margin},1
Style: Badge,{font},{int(caption_size * 0.85)},&H00FFFFFF,&H000000FF,&H00000000,&H96000000,1,0,0,0,100,100,0,0,3,8,0,5,{margin_lr},{margin_lr},0,1
Style: Minimal,{font},{int(caption_size * 0.72)},&H00FFFFFF,&H000000FF,&H66000000,&H00000000,0,0,0,0,100,100,0,0,1,3,1,2,{margin_lr},{margin_lr},{caption_margin},1

[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
    lines = [header]
    default_style = settings.get("caption_style", "tiktok-bold")
    for event in events:
        start = max(0.0, float(event.get("start", 0.0)))
        end = min(total_duration, float(event.get("end", 0.0)))
        if end <= start:
            continue
        style = style_for_event(event, default_style)
        text = wrap_caption(str(event["text"]), int(event.get("max_chars", 18)))
        effect = ""
        if style in {"Hook", "Badge"}:
            effect = r"{\fad(120,120)}"
        lines.append(
            f"Dialogue: 0,{ass_time(start)},{ass_time(end)},{style},,0,0,0,,{effect}{ass_escape(text)}"
        )

    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return out_path


def watermark_overlay(position: str, margin: int) -> tuple[str, str]:
    position = (position or "top-right").lower()
    if "left" in position:
        x = str(margin)
    elif "right" in position:
        x = f"W-w-{margin}"
    else:
        x = "(W-w)/2"
    if "bottom" in position:
        y = f"H-h-{margin}"
    elif "top" in position:
        y = str(margin)
    else:
        y = "(H-h)/2"
    return x, y


def final_render(
    base_path: Path,
    ass_path: Path | None,
    plan: dict,
    project_root: Path,
    out_path: Path,
    total_duration: float,
) -> None:
    export = plan.get("export") or {}
    settings = plan.get("settings") or {}
    cmd = ["ffmpeg", "-y", "-i", str(base_path)]
    filter_parts: list[str] = []
    video_label = "[0:v]"
    next_input = 1

    bgm = plan.get("bgm") or {}
    bgm_path = resolve_path(bgm.get("path"), project_root)
    if bgm_path:
        if not bgm_path.exists():
            raise SystemExit(f"BGM file not found: {bgm_path}")
        cmd.extend(["-stream_loop", "-1", "-i", str(bgm_path)])
        next_input += 1

    watermark = plan.get("watermark") or {}
    watermark_path = resolve_path(watermark.get("path"), project_root)
    if watermark_path:
        if not watermark_path.exists():
            raise SystemExit(f"Watermark file not found: {watermark_path}")
        cmd.extend(["-loop", "1", "-i", str(watermark_path)])
        wm_index = next_input
        next_input += 1
        wm_width = int(watermark.get("width", 180))
        opacity = float(watermark.get("opacity", 0.85))
        margin = int(watermark.get("margin", 48))
        x, y = watermark_overlay(watermark.get("position", "top-right"), margin)
        filter_parts.append(
            f"[{wm_index}:v]format=rgba,scale={wm_width}:-1,colorchannelmixer=aa={opacity:.3f}[wm];"
            f"{video_label}[wm]overlay={x}:{y}:format=auto[vwm]"
        )
        video_label = "[vwm]"

    if ass_path:
        filter_parts.append(f"{video_label}subtitles=filename='{ffmpeg_filter_path(ass_path)}'[vout]")
    else:
        filter_parts.append(f"{video_label}null[vout]")

    voice_volume = float(settings.get("voice_volume", 1.0))
    if bgm_path:
        bgm_volume = float(bgm.get("volume", 0.12))
        bgm_start = float(bgm.get("start", 0.0))
        fade_in = max(0.0, float(bgm.get("fade_in", 0.3)))
        fade_out = max(0.0, float(bgm.get("fade_out", 0.8)))
        fade_out_start = max(0.0, total_duration - fade_out)
        filter_parts.append(f"[0:a]volume={voice_volume:.4f},aresample=48000[a0]")
        filter_parts.append(
            f"[1:a]atrim=start={bgm_start:.3f}:duration={total_duration:.3f},"
            f"asetpts=PTS-STARTPTS,volume={bgm_volume:.4f},"
            f"afade=t=in:st=0:d={fade_in:.3f},"
            f"afade=t=out:st={fade_out_start:.3f}:d={fade_out:.3f},"
            f"aresample=48000,aformat=channel_layouts=stereo[a1]"
        )
        filter_parts.append("[a0][a1]amix=inputs=2:duration=first:dropout_transition=2[aout]")
    else:
        filter_parts.append(f"[0:a]volume={voice_volume:.4f},aresample=48000,aformat=channel_layouts=stereo[aout]")

    cmd.extend(
        [
            "-filter_complex",
            ";".join(filter_parts),
            "-map",
            "[vout]",
            "-map",
            "[aout]",
            "-c:v",
            "libx264",
            "-preset",
            str(export.get("preset", "fast")),
            "-crf",
            str(export.get("crf", 20)),
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            str(export.get("audio_bitrate", "192k")),
            "-t",
            f"{total_duration:.3f}",
        ]
    )
    if export.get("faststart", True):
        cmd.extend(["-movflags", "+faststart"])
    cmd.append(str(out_path))
    run(cmd)


def main() -> None:
    parser = argparse.ArgumentParser(description="Render a TikTok-ready cut from an edit plan.")
    parser.add_argument("plan", type=Path, help="Path to edit_plan.json.")
    parser.add_argument("--project-root", type=Path, help="Project root. Defaults to parent of plans/.")
    parser.add_argument("--output", type=Path, help="Output mp4 path.")
    parser.add_argument("--keep-work", action="store_true", help="Keep normalized segment files.")
    args = parser.parse_args()

    require_binary("ffmpeg")
    require_binary("ffprobe")

    plan_path = args.plan.expanduser().resolve()
    plan = load_json(plan_path)
    project_root = resolve_project_root(plan_path, args.project_root)
    renders_dir = project_root / "renders"
    work_dir = renders_dir / "_work"
    segments_dir = work_dir / "segments"
    renders_dir.mkdir(parents=True, exist_ok=True)
    segments_dir.mkdir(parents=True, exist_ok=True)

    project = plan.get("project") or {}
    target = project.get("target") or {"width": 1080, "height": 1920, "fps": 30}
    defaults = plan.get("settings") or {}
    export = plan.get("export") or {}
    clips = normalize_clips(plan)

    segment_paths: list[Path] = []
    report_clips: list[dict] = []
    source_cache: dict[str, dict] = {}
    for clip in clips:
        source_path = resolve_path(str(clip.get("source", "")), project_root)
        if not source_path or not source_path.exists():
            raise SystemExit(f"Clip source not found: {clip.get('source')}")
        cache_key = str(source_path)
        if cache_key not in source_cache:
            source_cache[cache_key] = media_info(source_path)
        out_path = segments_dir / f"seg_{clip['index']:03d}_{source_path.stem}.mp4"
        render_segment(clip, source_path, out_path, target, defaults, export, source_cache[cache_key])
        segment_paths.append(out_path)
        report_clips.append(
            {
                "source": str(source_path),
                "start": clip["start"],
                "end": clip["end"],
                "speed": clip["speed"],
                "output_start": round(clip["output_start"], 3),
                "output_end": round(clip["output_end"], 3),
                "label": clip.get("label"),
            }
        )

    total_duration = sum(float(c["output_duration"]) for c in clips)
    base_path = work_dir / "base_concat.mp4"
    concat_segments(segment_paths, base_path, work_dir)

    ass_path = write_ass(plan, clips, project_root, project_root / "captions" / "final_captions.ass", total_duration)
    output_name = defaults.get("output_name", "final_tiktok.mp4")
    out_path = args.output.expanduser().resolve() if args.output else renders_dir / output_name
    final_render(base_path, ass_path, plan, project_root, out_path, total_duration)

    info = media_info(out_path)
    report = {
        "plan": str(plan_path),
        "project_root": str(project_root),
        "output": str(out_path),
        "duration": round(total_duration, 3),
        "media_info": info,
        "clips": report_clips,
        "captions": str(ass_path) if ass_path else None,
    }
    (renders_dir / "render_report.json").write_text(
        json.dumps(report, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

    if not args.keep_work:
        shutil.rmtree(segments_dir, ignore_errors=True)
    print(json.dumps(report, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Optional local transcription helper for captions and cut decisions.

Uses faster-whisper when installed; otherwise falls back to the `whisper` CLI.
"""

from __future__ import annotations

import argparse
import json
import shutil
from pathlib import Path

from media_common import require_binary, run


def srt_time(seconds: float) -> str:
    seconds = max(0.0, seconds)
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int(round((seconds - int(seconds)) * 1000))
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"


def write_srt(segments: list[dict], out_path: Path) -> None:
    lines: list[str] = []
    for index, seg in enumerate(segments, 1):
        text = str(seg.get("text", "")).strip()
        if not text:
            continue
        lines.extend(
            [
                str(index),
                f"{srt_time(float(seg['start']))} --> {srt_time(float(seg['end']))}",
                text,
                "",
            ]
        )
    out_path.write_text("\n".join(lines), encoding="utf-8")


def extract_audio(source: Path, out_wav: Path) -> None:
    require_binary("ffmpeg")
    out_wav.parent.mkdir(parents=True, exist_ok=True)
    run(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(source),
            "-vn",
            "-ac",
            "1",
            "-ar",
            "16000",
            str(out_wav),
        ]
    )


def transcribe_faster_whisper(audio: Path, model_name: str, language: str | None) -> dict:
    try:
        from faster_whisper import WhisperModel  # type: ignore
    except ImportError as exc:
        raise RuntimeError("faster-whisper is not installed") from exc

    model = WhisperModel(model_name, device="auto", compute_type="auto")
    kwargs = {"word_timestamps": True, "vad_filter": True}
    if language and language != "auto":
        kwargs["language"] = language
    segments_iter, info = model.transcribe(str(audio), **kwargs)
    segments = []
    for idx, seg in enumerate(segments_iter):
        words = []
        for word in seg.words or []:
            words.append({"start": word.start, "end": word.end, "word": word.word})
        segments.append(
            {
                "id": idx,
                "start": seg.start,
                "end": seg.end,
                "text": seg.text.strip(),
                "words": words,
            }
        )
    return {
        "engine": "faster-whisper",
        "model": model_name,
        "language": getattr(info, "language", language or "auto"),
        "duration": getattr(info, "duration", None),
        "segments": segments,
    }


def transcribe_whisper_cli(audio: Path, model_name: str, language: str | None, out_dir: Path) -> dict:
    if shutil.which("whisper") is None:
        raise RuntimeError("Neither faster-whisper nor whisper CLI is available.")
    cmd = [
        "whisper",
        str(audio),
        "--model",
        model_name,
        "--word_timestamps",
        "True",
        "--output_format",
        "json",
        "--output_dir",
        str(out_dir),
    ]
    if language and language != "auto":
        cmd.extend(["--language", language])
    run(cmd)
    json_path = out_dir / f"{audio.stem}.json"
    data = json.loads(json_path.read_text(encoding="utf-8"))
    data.setdefault("engine", "whisper-cli")
    data.setdefault("model", model_name)
    return data


def main() -> None:
    parser = argparse.ArgumentParser(description="Transcribe a video/audio file for bggg-tiktok-cut.")
    parser.add_argument("source", type=Path)
    parser.add_argument("--out-dir", type=Path, help="Output directory. Defaults to sibling transcripts/.")
    parser.add_argument("--model", default="small", help="Whisper model name.")
    parser.add_argument("--language", default="auto", help="Language code, or auto.")
    args = parser.parse_args()

    source = args.source.expanduser().resolve()
    if not source.exists():
        raise SystemExit(f"Source not found: {source}")
    out_dir = args.out_dir.expanduser().resolve() if args.out_dir else source.parent.parent / "transcripts"
    out_dir.mkdir(parents=True, exist_ok=True)
    audio = out_dir / f"{source.stem}.wav"
    extract_audio(source, audio)

    language = args.language if args.language != "auto" else None
    try:
        result = transcribe_faster_whisper(audio, args.model, language)
    except RuntimeError:
        result = transcribe_whisper_cli(audio, args.model, language, out_dir)

    json_path = out_dir / f"{source.stem}.json"
    srt_path = out_dir / f"{source.stem}.srt"
    json_path.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    write_srt(result.get("segments", []), srt_path)
    print(json.dumps({"json": str(json_path), "srt": str(srt_path)}, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()