Video Assemble

"""Optional 剪映 / JianYing (CapCut) draft exporter — decoupled, stdlib + ffprobe only.

Reads a backend-neutral `timeline.json` (see timeline.py) and writes a 剪映 draft
folder (`draft_content.json` + `draft_info.json` + `draft_meta_info.json`) that the
desktop app can open and the user can keep editing: video clips on the main track,
the narration and BGM as their own audio tracks, the recap lines as a subtitle
track, and the gap-fill ducking carried as native volume keyframes.

The public entrypoints stay small (`build_draft`, `export_timeline_to_jianying`,
`_us`, CLI). Internally the exporter is split into schema/templates, a thin
normalized build context, material/segment builders, track layout metadata, and
a safe writer/bundler. This mirrors the useful schema boundaries from duo-video
while ffmpeg remains the canonical renderer and JianYing export remains an
optional sidecar.

Schema and the draft skeleton are reimplemented from the open-source
pyJianYingDraft (© GuanYixuan, Apache-2.0) and capcut-mate (© Hommy, Apache-2.0);
see ACKNOWLEDGEMENTS / 致谢 in the README. No third-party code is vendored — the
draft JSON is built directly here so the bundle stays stdlib-only.
"""

import json
import subprocess
import uuid

from jianying_builders import build_timeline_track as _build_timeline_track
from jianying_model import DraftBuildContext as _DraftBuildContext
from jianying_schema import draft_content_skeleton as _draft_content_skeleton
from jianying_schema import meta_info as _meta_info
from jianying_schema import us as _us
from jianying_writer import write_draft as _write_draft

__all__ = ["_us", "build_draft", "export_timeline_to_jianying", "main"]


def _default_id():
    return str(uuid.uuid4()).upper()


def _probe_media(path):
    """Return (duration_us, width, height) via ffprobe; zeros on any failure."""
    try:
        r = subprocess.run(
            [
                "ffprobe", "-v", "error", "-of", "json",
                "-show_entries", "format=duration:stream=width,height,codec_type", str(path),
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )
        data = json.loads(r.stdout or "{}")
        dur_us = _us(float(data.get("format", {}).get("duration") or 0))
        width = height = 0
        for stream in data.get("streams", []):
            if stream.get("codec_type") == "video":
                width, height = int(stream.get("width") or 0), int(stream.get("height") or 0)
                break
        return dur_us, width, height
    except (OSError, ValueError, json.JSONDecodeError, subprocess.TimeoutExpired):
        return 0, 0, 0


def build_draft(timeline, new_id=None, probe=None):
    """Build the 剪映 draft_content dict and companion meta from a timeline."""
    new_id = new_id or _default_id
    probe = probe or _probe_media
    ctx = _DraftBuildContext.from_timeline(timeline, new_id, probe)

    for timeline_track in timeline.get("tracks", []):
        _build_timeline_track(ctx, timeline_track)

    draft_id = new_id()
    content = _draft_content_skeleton(
        draft_id,
        ctx.width,
        ctx.height,
        ctx.fps,
        ctx.total_us,
        ctx.materials,
        ctx.tracks,
    )
    meta = _meta_info(draft_id, ctx.total_us)
    return content, meta, ctx.notes


def export_timeline_to_jianying(timeline, out_dir, draft_name="recap", new_id=None,
                                probe=None, bundle_media=False):
    """Write a 剪映 draft folder under out_dir/draft_name. Returns (folder, notes).

    bundle_media=True copies the referenced media into the draft folder so it is
    self-contained and portable to another machine.
    """
    content, meta, notes = build_draft(timeline, new_id=new_id, probe=probe)
    return _write_draft(content, meta, notes, out_dir, draft_name, bundle_media_enabled=bundle_media)


def main():
    import argparse
    ap = argparse.ArgumentParser(description="Export a timeline.json to a 剪映/JianYing draft folder.")
    ap.add_argument("timeline", help="path to timeline.json")
    ap.add_argument("--out-dir", required=True, help="parent dir to create the draft folder in")
    ap.add_argument("--name", default="recap", help="draft folder name")
    ap.add_argument("--bundle-media", action="store_true",
                    help="copy referenced media into the draft folder (portable, self-contained)")
    args = ap.parse_args()
    with open(args.timeline, encoding="utf-8") as f:
        timeline = json.load(f)
    draft_dir, notes = export_timeline_to_jianying(
        timeline,
        args.out_dir,
        args.name,
        bundle_media=args.bundle_media,
    )
    for note in notes:
        print(f"  注意: {note}")
    print(json.dumps({"status": "exported", "draft_dir": draft_dir}, ensure_ascii=False))


if __name__ == "__main__":
    main()

"""Material and segment builders for the milestone-1 JianYing exporter."""

import json
import os

from jianying_schema import us, validate_material_category
from jianying_tracks import RI_BGM, RI_NARRATION, RI_TEXT, RI_VIDEO


def timerange(start_us, dur_us):
    return {"start": int(start_us), "duration": int(dur_us)}


def speed_material(new_id):
    return {"id": new_id(), "speed": 1.0, "type": "speed", "mode": 0, "curve_speed": None}


def volume_keyframes(keyframes, seg_start_s, new_id):
    """Build one KFTypeVolume keyframe list from timeline-absolute points."""
    if not keyframes:
        return []
    kfs = []
    for kf in keyframes:
        kfs.append({
            "curveType": "Line",
            "graphID": "",
            "left_control": {"x": 0.0, "y": 0.0},
            "right_control": {"x": 0.0, "y": 0.0},
            "id": new_id(),
            "time_offset": max(0, us(kf["t"] - seg_start_s)),
            "values": [round(float(kf["gain"]), 4)],
        })
    return [{
        "id": new_id(),
        "keyframe_list": kfs,
        "material_id": "",
        "property_type": "KFTypeVolume",
    }]


def windowed_volume_keyframes(keyframes, seg_start_s, seg_end_s, default_gain, new_id):
    """Window timeline-absolute keyframes for one split/looped segment."""
    if not keyframes:
        return []
    start = float(seg_start_s)
    end = float(seg_end_s)
    default_gain = float(default_gain)
    ordered = sorted((
        {"t": float(kf["t"]), "gain": float(kf["gain"])}
        for kf in keyframes
        if "t" in kf and "gain" in kf
    ), key=lambda kf: kf["t"])
    if not ordered or end <= start:
        return []

    start_gain = default_gain
    for kf in ordered:
        if kf["t"] <= start:
            start_gain = kf["gain"]
        else:
            break
    inner = [kf for kf in ordered if start <= kf["t"] <= end]
    if not inner and abs(start_gain - default_gain) < 1e-4:
        return []

    selected = [{"t": start, "gain": start_gain}]
    for kf in inner:
        if abs(kf["t"] - start) < 1e-4:
            selected[-1] = {"t": start, "gain": kf["gain"]}
        else:
            selected.append(kf)
    if all(abs(kf["gain"] - default_gain) < 1e-4 for kf in selected):
        return []
    return volume_keyframes(selected, start, new_id)


def clip_default():
    return {
        "alpha": 1.0,
        "flip": {"horizontal": False, "vertical": False},
        "rotation": 0.0,
        "scale": {"x": 1.0, "y": 1.0},
        "transform": {"x": 0.0, "y": 0.0},
    }


def base_segment(material_id, target_start_us, target_dur_us, render_index, volume, keyframes, new_id):
    return {
        "enable_adjust": True,
        "enable_color_correct_adjust": False,
        "enable_color_curves": True,
        "enable_color_match_adjust": False,
        "enable_color_wheels": True,
        "enable_lut": True,
        "enable_smart_color_adjust": False,
        "last_nonzero_volume": 1.0,
        "reverse": False,
        "track_attribute": 0,
        "track_render_index": 0,
        "visible": True,
        "id": new_id(),
        "material_id": material_id,
        "target_timerange": timerange(target_start_us, target_dur_us),
        "common_keyframes": keyframes,
        "keyframe_refs": [],
        "speed": 1.0,
        "volume": round(float(volume), 4),
        "is_tone_modify": False,
        "render_index": render_index,
    }


def audio_segment_piece(material_id, target_start_us, target_dur_us, source_start_us,
                        source_dur_us, render_index, volume, keyframes, new_id):
    seg = base_segment(material_id, target_start_us, target_dur_us, render_index, volume, keyframes, new_id)
    seg["source_timerange"] = timerange(source_start_us, source_dur_us)
    seg["extra_material_refs"] = []
    seg["clip"] = None
    seg["hdr_settings"] = None
    return seg


def track(name, track_type, segments, new_id):
    return {
        "attribute": 0,
        "flag": 0,
        "id": new_id(),
        "is_default_name": False,
        "name": name,
        "segments": segments,
        "type": track_type,
    }


def unsupported_track_note(kind):
    info = validate_material_category(kind)
    if info["supported"]:
        return None
    return info.get("note")


def build_video_track(ctx, timeline_track):
    segs = []
    for clip in timeline_track.get("clips", []):
        ts, te = float(clip["timeline_start"]), float(clip["timeline_end"])
        ss, se = float(clip["source_start"]), float(clip["source_end"])
        path = clip["source_path"]
        src_dur_us, width, height = ctx.media_duration(path, us(se))
        mat_id = ctx.new_id()
        ctx.materials["videos"].append({
            "audio_fade": None,
            "category_id": "",
            "category_name": "local",
            "check_flag": 63487,
            "crop": {
                "upper_left_x": 0.0,
                "upper_left_y": 0.0,
                "upper_right_x": 1.0,
                "upper_right_y": 0.0,
                "lower_left_x": 0.0,
                "lower_left_y": 1.0,
                "lower_right_x": 1.0,
                "lower_right_y": 1.0,
            },
            "crop_ratio": "free",
            "crop_scale": 1.0,
            "duration": int(src_dur_us),
            "height": height or ctx.height,
            "id": mat_id,
            "local_material_id": mat_id,
            "material_id": mat_id,
            "material_name": os.path.basename(path),
            "media_path": "",
            "path": path,
            "type": "video",
            "width": width or ctx.width,
        })
        speed = speed_material(ctx.new_id)
        ctx.materials["speeds"].append(speed)
        audio = clip.get("audio", {})
        keyframes = volume_keyframes(audio.get("volume_keyframes"), ts, ctx.new_id)
        volume = audio.get("base_gain", 1.0) if not keyframes else 1.0
        seg = base_segment(mat_id, us(ts), us(te - ts), RI_VIDEO, volume, keyframes, ctx.new_id)
        seg["source_timerange"] = timerange(us(ss), us(se - ss))
        seg["extra_material_refs"] = [speed["id"]]
        seg["clip"] = clip_default()
        seg["uniform_scale"] = {"on": True, "value": 1.0}
        seg["hdr_settings"] = {"intensity": 1.0, "mode": 1, "nits": 1000}
        segs.append(seg)
    ctx.tracks.append(track("video", "video", segs, ctx.new_id))


def build_audio_track(ctx, timeline_track):
    role = timeline_track.get("role", timeline_track.get("name", "audio"))
    render_index = RI_BGM if role == "bgm" else RI_NARRATION
    segs = []
    for segment in timeline_track.get("segments", []):
        ts, te = float(segment["timeline_start"]), float(segment["timeline_end"])
        path = segment["source_path"]
        mat_dur_us, _width, _height = ctx.media_duration(path, us(te - ts))
        want_us = us(te - ts)
        place_us = want_us
        if mat_dur_us and want_us > mat_dur_us:
            if role == "bgm" and timeline_track.get("loop"):
                place_us = want_us
            else:
                place_us = mat_dur_us
            if role == "bgm" and not timeline_track.get("loop"):
                ctx.note(
                    f"BGM 素材({mat_dur_us/1e6:.1f}s) 短于时间线({(te - ts):.1f}s)，"
                    "剪映中未循环铺满（可在剪映里手动复制延长）"
                )
        mat_id = ctx.new_id()
        ctx.materials["audios"].append({
            "app_id": 0,
            "category_id": "",
            "category_name": "local",
            "check_flag": 3,
            "copyright_limit_type": "none",
            "duration": int(mat_dur_us or want_us),
            "effect_id": "",
            "formula_id": "",
            "id": mat_id,
            "local_material_id": mat_id,
            "music_id": mat_id,
            "name": os.path.basename(path),
            "path": path,
            "source_platform": 0,
            "type": "extract_music",
            "wave_points": [],
        })
        keyframes = volume_keyframes(segment.get("volume_keyframes"), ts, ctx.new_id)
        volume = segment.get("gain", 1.0) if not keyframes else 1.0
        if role == "bgm" and timeline_track.get("loop") and mat_dur_us and want_us > mat_dur_us:
            cursor = 0
            while cursor < want_us:
                piece = min(mat_dur_us, want_us - cursor)
                if piece <= 0:
                    break
                piece_start_s = ts + (cursor / 1_000_000)
                piece_end_s = ts + ((cursor + piece) / 1_000_000)
                speed = speed_material(ctx.new_id)
                ctx.materials["speeds"].append(speed)
                piece_kfs = windowed_volume_keyframes(
                    segment.get("volume_keyframes"), piece_start_s, piece_end_s,
                    segment.get("gain", 1.0), ctx.new_id)
                piece_volume = segment.get("gain", 1.0) if not piece_kfs else 1.0
                piece_seg = audio_segment_piece(
                    mat_id, us(ts) + cursor, piece, 0, piece, render_index,
                    piece_volume, piece_kfs, ctx.new_id)
                piece_seg["extra_material_refs"] = [speed["id"]]
                segs.append(piece_seg)
                cursor += piece
        else:
            speed = speed_material(ctx.new_id)
            ctx.materials["speeds"].append(speed)
            audio_seg = audio_segment_piece(
                mat_id, us(ts), place_us, 0, place_us, render_index, volume, keyframes, ctx.new_id)
            audio_seg["extra_material_refs"] = [speed["id"]]
            segs.append(audio_seg)
    ctx.tracks.append(track(role, "audio", segs, ctx.new_id))


def build_text_track(ctx, timeline_track):
    segs = []
    for segment in timeline_track.get("segments", []):
        ts, te = float(segment["timeline_start"]), float(segment["timeline_end"])
        text = segment.get("text", "")
        mat_id = ctx.new_id()
        content = {"text": text, "styles": [{
            "fill": {
                "alpha": 1.0,
                "content": {"render_type": "solid", "solid": {"alpha": 1.0, "color": [1.0, 1.0, 1.0]}},
            },
            "range": [0, len(text.encode("utf-16-le")) // 2],
            "size": 8.0,
            "bold": False,
            "italic": False,
            "underline": False,
            "strokes": [],
        }]}
        ctx.materials["texts"].append({
            "id": mat_id,
            "content": json.dumps(content, ensure_ascii=False),
            "type": "text",
            "typesetting": 0,
            "alignment": 1,
            "letter_spacing": 0.0,
            "line_spacing": 0.02,
            "font_size": 8.0,
            "text_color": "#FFFFFF",
            "add_type": 0,
            "check_flag": 7,
        })
        seg = base_segment(mat_id, us(ts), us(te - ts), RI_TEXT, 1.0, [], ctx.new_id)
        seg["source_timerange"] = None
        seg["extra_material_refs"] = []
        seg["clip"] = clip_default()
        seg["uniform_scale"] = {"on": True, "value": 1.0}
        seg["type"] = "text_effect"
        seg["source_platform"] = 1
        seg["clip"]["transform"]["y"] = -0.72
        segs.append(seg)
    ctx.tracks.append(track("subtitle", "text", segs, ctx.new_id))


def build_timeline_track(ctx, timeline_track):
    kind = timeline_track.get("kind")
    if kind in ("audio", "text") and not timeline_track.get("segments"):
        return
    if kind == "video":
        build_video_track(ctx, timeline_track)
    elif kind == "audio":
        build_audio_track(ctx, timeline_track)
    elif kind == "text":
        build_text_track(ctx, timeline_track)
    else:
        note = unsupported_track_note(kind)
        if note:
            ctx.note(note)

"""Thin internal model used at the JianYing adapter boundary."""

import os
from dataclasses import dataclass, field
from typing import Callable

from jianying_schema import us


ProbeFn = Callable[[str], tuple[int, int, int]]
NewIdFn = Callable[[], str]


@dataclass
class DraftBuildContext:
    """Normalized draft build state.

    Public `timeline.json` stays backend-neutral (seconds/gains). This context is
    the adapter-local place where canvas values and durations are normalized into
    JianYing-friendly integers and material/track arrays are accumulated.
    """

    width: int
    height: int
    fps: float
    total_us: int
    new_id: NewIdFn
    probe: ProbeFn
    materials: dict[str, list] = field(
        default_factory=lambda: {"videos": [], "audios": [], "texts": [], "speeds": []}
    )
    tracks: list[dict] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)

    @classmethod
    def from_timeline(cls, timeline, new_id, probe):
        canvas = timeline.get("canvas", {})
        return cls(
            width=int(canvas.get("width", 1920)),
            height=int(canvas.get("height", 1080)),
            fps=float(canvas.get("fps", 30)),
            total_us=us(timeline.get("duration", 0)),
            new_id=new_id,
            probe=probe,
        )

    def media_duration(self, path, fallback_us):
        if path and os.path.exists(path):
            duration_us, width, height = self.probe(path)
            return duration_us or fallback_us, width, height
        return fallback_us, 0, 0

    def note(self, message):
        self.notes.append(message)

"""Schema constants, skeleton factories, and capability registries for JianYing export.

This module is intentionally data-oriented: it owns draft version metadata, the
full `materials` parallel-array shape, and the duo-video-inspired distinction
between material categories and cross-cutting exporter capabilities.
"""

DRAFT_VERSION = 360000
NEW_VERSION = "111.0.0"
APP = {"app_id": 3704, "app_source": "lv", "app_version": "5.9.5-beta1", "os": "mac"}

# The full 剪映 materials object: ~45 parallel arrays, only a few are populated
# by the milestone-1 exporter. Keeping every array preserves draft compatibility.
MATERIAL_KEYS = (
    "ai_translates audio_balances audio_effects audio_fades audio_track_indexes audios "
    "beats canvases chromas color_curves digital_humans drafts effects flowers green_screens "
    "handwrites hsl images log_color_wheels loudnesses manual_deformations masks common_mask "
    "material_animations material_colors multi_language_refs placeholders plugin_effects "
    "primary_color_wheels realtime_denoises shapes smart_crops smart_relights "
    "sound_channel_mappings speeds stickers tail_leaders text_templates texts time_marks "
    "transitions video_effects video_trackings videos vocal_beautifys vocal_separations"
).split()


def us(seconds):
    """Seconds (float) -> integer microseconds. The single seconds->µs boundary."""
    return int(round(float(seconds) * 1_000_000))


def full_materials(filled):
    """Return a complete JianYing `materials` object with all known arrays."""
    out = {k: [] for k in MATERIAL_KEYS}
    out.update(filled)
    return out


def draft_content_skeleton(draft_id, width, height, fps, total_us, materials, tracks):
    """Build the root `draft_content.json` / `draft_info.json` skeleton."""
    return {
        "canvas_config": {"width": width, "height": height, "ratio": "original"},
        "color_space": 0,
        "config": {
            "adjust_max_index": 1,
            "attachment_info": [],
            "combination_max_index": 1,
            "export_range": None,
            "extract_audio_last_index": 1,
            "lyrics_recognition_id": "",
            "lyrics_sync": True,
            "lyrics_taskinfo": [],
            "maintrack_adsorb": True,
            "material_save_mode": 0,
            "multi_language_current": "none",
            "multi_language_list": [],
            "multi_language_main": "none",
            "multi_language_mode": "none",
            "original_sound_last_index": 1,
            "record_audio_last_index": 1,
            "sticker_max_index": 1,
            "subtitle_keywords_config": None,
            "subtitle_recognition_id": "",
            "subtitle_sync": True,
            "subtitle_taskinfo": [],
            "system_font_list": [],
            "video_mute": False,
            "zoom_info_params": None,
        },
        "cover": None,
        "create_time": 0,
        "duration": int(total_us),
        "extra_info": None,
        "fps": fps,
        "free_render_index_mode_on": False,
        "group_container": None,
        "id": draft_id,
        "keyframe_graph_list": [],
        "keyframes": {
            "adjusts": [],
            "audios": [],
            "effects": [],
            "filters": [],
            "handwrites": [],
            "stickers": [],
            "texts": [],
            "videos": [],
        },
        "last_modified_platform": dict(APP),
        "platform": dict(APP),
        "materials": full_materials(materials),
        "mutable_config": None,
        "name": "",
        "new_version": NEW_VERSION,
        "relationships": [],
        "render_index_track_mode_on": False,
        "retouch_cover": None,
        "source": "default",
        "static_cover_image_path": "",
        "time_marks": None,
        "tracks": tracks,
        "update_time": 0,
        "version": DRAFT_VERSION,
    }


def meta_info(draft_id, total_us):
    """Build the companion `draft_meta_info.json` skeleton."""
    return {
        "cloud_package_completed_time": "",
        "draft_cloud_capcut_purchase_info": "",
        "draft_cloud_last_action_download": False,
        "draft_cloud_materials": [],
        "draft_cloud_purchase_info": "",
        "draft_cloud_template_id": "",
        "draft_cloud_tutorial_info": "",
        "draft_cloud_videocut_purchase_info": "",
        "draft_cover": "",
        "draft_deeplink_url": "",
        "draft_enterprise_info": {
            "draft_enterprise_extra": "",
            "draft_enterprise_id": "",
            "draft_enterprise_name": "",
            "enterprise_material": [],
        },
        "draft_fold_path": "",
        "draft_id": draft_id,
        "draft_is_ai_packaging_used": False,
        "draft_is_ai_shorts": False,
        "draft_is_ai_translate": False,
        "draft_is_article_video_draft": False,
        "draft_is_from_deeplink": "false",
        "draft_is_invisible": False,
        "draft_materials": [{"type": t, "value": []} for t in (0, 1, 2, 3, 6, 7, 8)],
        "draft_materials_copied_info": [],
        "draft_name": "",
        "draft_new_version": "",
        "draft_removable_storage_device": "",
        "draft_root_path": "",
        "draft_segment_extra_info": [],
        "draft_type": "",
        "tm_draft_cloud_completed": "",
        "tm_draft_cloud_modified": 0,
        "tm_draft_removed": 0,
        "tm_duration": int(total_us),
    }


def material_category_registry():
    """Material category support table inspired by duo-video's MaterialTypeEnum.

    Keep this separate from feature capabilities: keyframes, bundling, and path
    rewrite are exporter capabilities, not material categories.
    """
    return {
        "video": {"status": "supported", "materials_key": "videos", "track_type": "video"},
        "audio": {"status": "supported", "materials_key": "audios", "track_type": "audio"},
        "text": {"status": "supported", "materials_key": "texts", "track_type": "text"},
        "subtitle": {"status": "supported", "materials_key": "texts", "track_type": "text"},
        "speed": {"status": "supported_auxiliary", "materials_key": "speeds", "track_type": None},
        "image": {"status": "reserved", "materials_key": "images", "track_type": "video"},
        "sticker": {"status": "reserved", "materials_key": "stickers", "track_type": "sticker"},
        "sound": {"status": "reserved", "materials_key": "audios", "track_type": "audio"},
        "text_template": {"status": "reserved", "materials_key": "text_templates", "track_type": "text"},
        "lut": {"status": "reserved", "materials_key": "effects", "track_type": "video"},
        "transition": {"status": "reserved", "materials_key": "transitions", "track_type": "video"},
        "video_effect": {"status": "reserved", "materials_key": "video_effects", "track_type": "video"},
        "face_effect": {"status": "reserved", "materials_key": "effects", "track_type": "video"},
        "mask": {"status": "reserved", "materials_key": "masks", "track_type": "video"},
        "style": {"status": "reserved", "materials_key": "material_colors", "track_type": "video"},
    }


def feature_capabilities():
    """Cross-cutting exporter capabilities, deliberately not material categories."""
    return {
        "volume_automation": {
            "status": "supported",
            "property_type": "KFTypeVolume",
            "description": "timeline-absolute gain points become segment-relative JianYing keyframes",
        },
        "bgm_loop_splitting": {
            "status": "supported",
            "description": "looped BGM is split into repeated JianYing audio segments",
        },
        "media_bundling": {
            "status": "supported",
            "description": "referenced media can be copied into the draft materials folder",
        },
        "path_rewrite": {
            "status": "supported",
            "description": "bundled media paths are rewritten after atomic move",
        },
        "collision_safe_write": {
            "status": "supported",
            "description": "non-empty draft folders are never overwritten",
        },
        "lazy_export_isolation": {
            "status": "supported",
            "description": "assemble imports JianYing modules only when export is requested",
        },
    }


def validate_material_category(category):
    """Return deterministic support metadata for a public or future material kind."""
    normalized = (category or "").strip().lower()
    registry = material_category_registry()
    info = dict(registry.get(normalized, {"status": "unsupported", "materials_key": None, "track_type": None}))
    info["category"] = normalized
    info["supported"] = info["status"] in {"supported", "supported_auxiliary"}
    if not info["supported"]:
        status = "保留但暂未实现" if info["status"] == "reserved" else "未知"
        info["note"] = f"暂不支持的 JianYing material category: {normalized}（{status}），已跳过以避免生成无效草稿"
    return info

"""Track layout bands and deterministic overlap-safe allocation for JianYing export."""

from dataclasses import dataclass


@dataclass(frozen=True)
class TrackBand:
    kind: str
    track_type: str
    render_index: int
    description: str


# Parity values for existing output are preserved: video at 0, narration/BGM at
# 1/2 via explicit role handling, and subtitles in JianYing's high text band.
RI_VIDEO = 0
RI_NARRATION = 1
RI_BGM = 2
RI_TEXT = 15000


TRACK_LAYOUT_BANDS = {
    "audio": TrackBand("audio", "audio", RI_NARRATION, "narration and general audio below text"),
    "sound": TrackBand("sound", "audio", RI_BGM, "sound/BGM bed lane"),
    "video": TrackBand("video", "video", RI_VIDEO, "base video track"),
    "image": TrackBand("image", "video", 1000, "future image/overlay lane"),
    "overlay": TrackBand("overlay", "video", 1000, "future video/image overlay lane"),
    "mask": TrackBand("mask", "video", 4000, "future mask/effect lane"),
    "effect": TrackBand("effect", "video", 5000, "future effects lane"),
    "video_effect": TrackBand("video_effect", "video", 5000, "future video effects lane"),
    "sticker": TrackBand("sticker", "sticker", 10000, "future sticker lane"),
    "subtitle": TrackBand("subtitle", "text", RI_TEXT, "subtitle text lane"),
    "text": TrackBand("text", "text", RI_TEXT, "plain text lane"),
    "text_template": TrackBand("text_template", "text", RI_TEXT + 100, "future text template lane"),
}


@dataclass(frozen=True)
class AllocatedTrack:
    kind: str
    name: str
    track_type: str
    render_index: int


class TrackAllocator:
    """Allocate deterministic suffix tracks when same-name segments overlap."""

    def __init__(self):
        self._occupied = {}

    @staticmethod
    def _overlaps(start_us, duration_us, existing):
        end_us = int(start_us) + int(duration_us)
        return any(int(start_us) < old_end and end_us > old_start for old_start, old_end in existing)

    def allocate(self, kind, base_name, start_us, duration_us):
        band = TRACK_LAYOUT_BANDS.get(kind, TRACK_LAYOUT_BANDS["video"])
        base_name = base_name or kind
        suffix = 0
        while True:
            name = base_name if suffix == 0 else f"{base_name}-{suffix}"
            key = (kind, name)
            occupied = self._occupied.setdefault(key, [])
            if not self._overlaps(start_us, duration_us, occupied):
                occupied.append((int(start_us), int(start_us) + int(duration_us)))
                return AllocatedTrack(kind, name, band.track_type, band.render_index + suffix)
            suffix += 1

"""Safe writer and media bundler for JianYing draft folders."""

import json
import os
import shutil
import tempfile


def validate_draft_name(draft_name):
    """Reject draft names that could escape or alias the requested parent dir."""
    if not isinstance(draft_name, str):
        raise TypeError("draft_name must be a string")
    if not draft_name or not draft_name.strip():
        raise ValueError("draft_name must not be empty")
    if os.path.isabs(draft_name):
        raise ValueError("draft_name must be a plain folder name, not an absolute path")
    if "/" in draft_name or "\\" in draft_name:
        raise ValueError("draft_name must not contain path separators")
    if draft_name in {".", ".."}:
        raise ValueError("draft_name must not be '.' or '..'")


def bundle_media(content, draft_dir):
    """Copy referenced media into `<draft_dir>/materials/` and rewrite paths."""
    mats_dir = os.path.join(draft_dir, "materials")
    os.makedirs(mats_dir, exist_ok=True)
    copied, used, notes = {}, set(), []
    for arr in (content["materials"]["videos"], content["materials"]["audios"]):
        for material in arr:
            src = material.get("path")
            if not src:
                continue
            if src in copied:
                material["path"] = copied[src]
                continue
            if not os.path.exists(src):
                notes.append(f"素材缺失，未打包: {src}")
                continue
            base = os.path.basename(src)
            name, stem_ext = base, os.path.splitext(base)
            i = 1
            while name in used:
                name = f"{stem_ext[0]}_{i}{stem_ext[1]}"
                i += 1
            used.add(name)
            dest = os.path.join(mats_dir, name)
            shutil.copy2(src, dest)
            copied[src] = dest
            material["path"] = dest
    return notes


def draft_dir_has_user_content(draft_dir):
    """Return True when writing here could overwrite an existing draft/material."""
    if not os.path.exists(draft_dir):
        return False
    try:
        return any(os.scandir(draft_dir))
    except OSError:
        return True


def collision_safe_draft_dir(out_dir, draft_name):
    """Pick a fresh draft folder instead of overwriting an existing non-empty one."""
    validate_draft_name(draft_name)
    base = os.path.join(out_dir, draft_name)
    if not draft_dir_has_user_content(base):
        return base, draft_name
    idx = 2
    while True:
        candidate_name = f"{draft_name}_{idx}"
        candidate = os.path.join(out_dir, candidate_name)
        if not draft_dir_has_user_content(candidate):
            return candidate, candidate_name
        idx += 1


def rewrite_material_prefix(content, old_prefix, new_prefix):
    old_prefix = os.path.abspath(old_prefix)
    new_prefix = os.path.abspath(new_prefix)
    for arr in (content["materials"]["videos"], content["materials"]["audios"]):
        for material in arr:
            path = material.get("path")
            if path and os.path.abspath(path).startswith(old_prefix + os.sep):
                material["path"] = new_prefix + os.path.abspath(path)[len(old_prefix):]


def write_draft(content, meta, notes, out_dir, draft_name, bundle_media_enabled=False):
    """Atomically write the three JianYing draft JSON files and optional bundle."""
    validate_draft_name(draft_name)
    out_dir = os.path.abspath(out_dir)
    os.makedirs(out_dir, exist_ok=True)
    draft_dir, actual_name = collision_safe_draft_dir(out_dir, draft_name)
    notes = list(notes)
    if actual_name != draft_name:
        notes.append(f"草稿目录已存在，改写为 {actual_name} 以避免覆盖")

    tmp_parent = tempfile.mkdtemp(prefix=f".{actual_name}.", dir=out_dir)
    tmp_dir = os.path.join(tmp_parent, actual_name)
    try:
        os.makedirs(tmp_dir, exist_ok=False)
        if bundle_media_enabled:
            notes = notes + bundle_media(content, tmp_dir)
            rewrite_material_prefix(content, tmp_dir, draft_dir)
        meta["draft_name"] = actual_name
        meta["draft_fold_path"] = draft_dir
        content["name"] = actual_name
        for fname in ("draft_content.json", "draft_info.json"):
            with open(os.path.join(tmp_dir, fname), "w", encoding="utf-8") as f:
                json.dump(content, f, ensure_ascii=False, indent=2)
        with open(os.path.join(tmp_dir, "draft_meta_info.json"), "w", encoding="utf-8") as f:
            json.dump(meta, f, ensure_ascii=False, indent=2)
        if os.path.isdir(draft_dir) and not draft_dir_has_user_content(draft_dir):
            os.rmdir(draft_dir)
        os.replace(tmp_dir, draft_dir)
    except Exception:
        shutil.rmtree(tmp_parent, ignore_errors=True)
        raise
    finally:
        if os.path.exists(tmp_parent):
            shutil.rmtree(tmp_parent, ignore_errors=True)
    return draft_dir, notes

"""Self-contained config + utilities for this skill (no cross-skill imports)."""
import os
import subprocess
from pathlib import Path


# ── 配置 ──────────────────────────────────────────────────────────────

DEFAULT_MIMO_API_URL = "https://api.xiaomimimo.com/v1"
DEFAULT_MIMO_TOKEN_PLAN_CLUSTER = "cn"
MIMO_TOKEN_PLAN_API_URLS = {
    "cn": "https://token-plan-cn.xiaomimimo.com/v1",
    "sgp": "https://token-plan-sgp.xiaomimimo.com/v1",
    "ams": "https://token-plan-ams.xiaomimimo.com/v1",
}
DEFAULT_MIMO_MODEL = "mimo-v2.5"          # VLM / chat (vision understanding)
DEFAULT_MIMO_ASR_MODEL = "mimo-v2.5-asr"  # speech-to-text
DEFAULT_MIMO_TTS_MODEL = "mimo-v2.5-tts"  # text-to-speech


def normalize_api_url(raw_url):
    """Normalize a MiMo (OpenAI-compatible) base URL or chat/completions endpoint."""
    url = (raw_url or DEFAULT_MIMO_API_URL).rstrip("/")
    if url.endswith("/chat/completions"):
        return url
    return f"{url}/chat/completions"


def is_mimo_token_plan_key(api_key):
    """Return True for Xiaomi MiMo Token Plan keys, which use token-plan base URLs."""
    return str(api_key or "").strip().startswith("tp-")


def default_mimo_api_url(api_key="", cluster=None):
    """Pick the correct MiMo base URL for pay-as-you-go vs Token Plan keys.

    MiMo uses independent credentials for pay-as-you-go (`sk-*`) and Token Plan
    (`tp-*`). Token Plan keys must be sent to the Token Plan cluster base URL,
    not the pay-as-you-go `api.xiaomimimo.com` endpoint.
    """
    if is_mimo_token_plan_key(api_key):
        cluster_name = (cluster or os.environ.get("MIMO_TOKEN_PLAN_CLUSTER") or DEFAULT_MIMO_TOKEN_PLAN_CLUSTER)
        cluster_name = str(cluster_name).strip().lower()
        return MIMO_TOKEN_PLAN_API_URLS.get(cluster_name, MIMO_TOKEN_PLAN_API_URLS[DEFAULT_MIMO_TOKEN_PLAN_CLUSTER])
    return DEFAULT_MIMO_API_URL


def env_int(name, default, *, minimum=None):
    """Read an integer env var; ignore malformed values instead of crashing import."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        value = int(raw)
    except (TypeError, ValueError):
        return default
    if minimum is not None:
        value = max(minimum, value)
    return value


def env_bool(name, default=False):
    """Read common boolean env var forms."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def env_float(name, default, *, minimum=None):
    """Read a float env var; ignore malformed values instead of crashing import."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return default
    if minimum is not None:
        value = max(minimum, value)
    return value


# Single MiMo credential powers ASR + VLM + TTS. Per-capability overrides
# (MIMO_VIDEO_API_KEY / MIMO_TTS_API_KEY / MIMO_ASR_API_KEY and their *_API_URL forms)
# are optional and fall back to MIMO_API_KEY / MIMO_API_URL. Token-Plan keys (tp-*) auto-
# route to the Token-Plan cluster base URL; pay-as-you-go keys use api.xiaomimimo.com.
_mimo_api_key = os.environ.get("MIMO_API_KEY", "")
_mimo_video_api_key = os.environ.get("MIMO_VIDEO_API_KEY", "") or _mimo_api_key
_mimo_tts_api_key = os.environ.get("MIMO_TTS_API_KEY", "") or _mimo_api_key
_mimo_asr_api_key = os.environ.get("MIMO_ASR_API_KEY", "") or _mimo_api_key
_raw_api_url = os.environ.get("MIMO_API_URL") or default_mimo_api_url(_mimo_api_key)
_raw_mimo_video_api_url = (
    os.environ.get("MIMO_VIDEO_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_video_api_key)
)
_raw_mimo_tts_api_url = (
    os.environ.get("MIMO_TTS_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_tts_api_key)
)
_raw_mimo_asr_api_url = (
    os.environ.get("MIMO_ASR_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_asr_api_key)
)

# Cross-language source: when the original audio is in a language the narration is NOT in
# (e.g. a Japanese drama recapped in Chinese), the original speech bleeding under the narration
# is just noise the viewer can't parse — it reads as 怪音. In that mode the original is ducked to
# near-silent UNDER narration; it still plays full-volume in the original-audio gap blocks, where
# a single language is fine. Explicit SPEECH_DUCKING_VOLUME / ZONE_DUCKING_VOLUME still override.
_foreign_source_audio = env_bool("FOREIGN_SOURCE_AUDIO", False)
_foreign_under_narration_volume = 0.05  # original volume under narration when source audio is foreign

CONFIG = {
    "api_provider": "mimo",
    "api_provider_source": "default",
    "api_url": normalize_api_url(_raw_api_url),
    "api_url_source": "env" if os.environ.get("MIMO_API_URL") else "default",
    "api_key": _mimo_api_key,
    "api_key_source": "MIMO_API_KEY",
    "mimo_api_url": normalize_api_url(_raw_api_url),
    "mimo_api_url_source": "env" if os.environ.get("MIMO_API_URL") else "default",
    "mimo_api_key": _mimo_api_key,
    "mimo_api_key_source": "MIMO_API_KEY",
    "mimo_video_api_url": normalize_api_url(_raw_mimo_video_api_url),
    "mimo_video_api_url_source": "env" if (
        os.environ.get("MIMO_VIDEO_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_video_api_key": _mimo_video_api_key,
    "mimo_video_api_key_source": "MIMO_VIDEO_API_KEY" if os.environ.get("MIMO_VIDEO_API_KEY") else "MIMO_API_KEY",
    "mimo_tts_api_url": normalize_api_url(_raw_mimo_tts_api_url),
    "mimo_tts_api_url_source": "env" if (
        os.environ.get("MIMO_TTS_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_tts_api_key": _mimo_tts_api_key,
    "mimo_tts_api_key_source": "MIMO_TTS_API_KEY" if os.environ.get("MIMO_TTS_API_KEY") else "MIMO_API_KEY",
    "mimo_asr_api_url": normalize_api_url(_raw_mimo_asr_api_url),
    "mimo_asr_api_url_source": "env" if (
        os.environ.get("MIMO_ASR_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_asr_api_key": _mimo_asr_api_key,
    "mimo_asr_api_key_source": "MIMO_ASR_API_KEY" if os.environ.get("MIMO_ASR_API_KEY") else "MIMO_API_KEY",
    "mimo_model": os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "mimo_model_source": "env" if os.environ.get("MIMO_MODEL") else "default",
    "mimo_video_model": os.environ.get("MIMO_VIDEO_MODEL") or os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "mimo_video_model_source": "env" if (
        os.environ.get("MIMO_VIDEO_MODEL") or os.environ.get("MIMO_MODEL")
    ) else "default",
    "vlm_model": os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "vlm_model_source": "env" if os.environ.get("MIMO_MODEL") else "default",
    "mimo_asr_model": os.environ.get("MIMO_ASR_MODEL", DEFAULT_MIMO_ASR_MODEL),
    "mimo_asr_model_source": "env" if os.environ.get("MIMO_ASR_MODEL") else "default",
    "mimo_asr_language": os.environ.get("MIMO_ASR_LANGUAGE", "auto"),  # auto | zh | en
    "mimo_asr_base64_max_mb": env_float("MIMO_ASR_BASE64_MAX_MB", 10.0, minimum=1.0),
    # ASR 分段窗口秒数。越小 → 长视频的对白时间戳越精细（默认 15s）。旧值 180s 会把 >3min
    # 视频的对白塌缩成一个时间戳，既让 brief 无法定位对白，又触发 detect.py 的粗粒度跳过，
    # 使 overlaps_speech/安静窗口判断失真。代价是更多 ASR 调用；ASR 慢时可调大。
    "asr_segment_seconds": env_float("ASR_SEGMENT_SECONDS", 15.0, minimum=5.0),
    "scene_threshold": 0.1,
    "scene_threshold_source": "default",
    "mimo_tts_model": os.environ.get("MIMO_TTS_MODEL", DEFAULT_MIMO_TTS_MODEL),
    "mimo_tts_model_source": "env" if os.environ.get("MIMO_TTS_MODEL") else "default",
    "mimo_tts_voice": os.environ.get("MIMO_TTS_VOICE", "冰糖"),
    "mimo_tts_voice_source": "env" if os.environ.get("MIMO_TTS_VOICE") else "default",
    "mimo_tts_style": os.environ.get(
        "MIMO_TTS_STYLE",
        "自然、清晰、有感染力，像在给观众讲故事；随剧情起伏，该紧张时紧张、该动情时动情，不平铺直叙。",
    ),
    "mimo_tts_style_source": "env" if os.environ.get("MIMO_TTS_STYLE") else "default",
    "mimo_media_resolution": os.environ.get("MIMO_MEDIA_RESOLUTION", "default"),
    "mimo_media_resolution_source": "env" if os.environ.get("MIMO_MEDIA_RESOLUTION") else "default",
    "mimo_video_overview": env_bool("MIMO_VIDEO_OVERVIEW", False),  # opt-in (--mimo-video-overview / =1); when on it becomes the PRIMARY per-scene description, frames stay the anchor/fallback
    "mimo_video_overview_source": "env" if os.environ.get("MIMO_VIDEO_OVERVIEW") else "default",
    "mimo_video_fps": env_float("MIMO_VIDEO_FPS", 3.0, minimum=0.1),
    "mimo_video_fps_source": "env" if os.environ.get("MIMO_VIDEO_FPS") else "default",
    "mimo_video_chunk_max_seconds": env_float("MIMO_VIDEO_CHUNK_MAX_SECONDS", 20.0, minimum=1.0),
    "mimo_video_chunk_min_seconds": env_float("MIMO_VIDEO_CHUNK_MIN_SECONDS", 1.0, minimum=0.2),
    "mimo_video_chunk_timeout": env_int("MIMO_VIDEO_CHUNK_TIMEOUT", 180, minimum=1),
    "mimo_video_base64_max_mb": env_float("MIMO_VIDEO_BASE64_MAX_MB", 45.0, minimum=1.0),
    # Per-scene frame VLM sampling — scale frames with scene length instead of a hard cap of 6
    "vlm_seconds_per_frame": env_float("VLM_SECONDS_PER_FRAME", 4.0, minimum=0.5),
    "vlm_max_frames": env_int("VLM_MAX_FRAMES", 16, minimum=3),
    "vlm_max_tokens": env_int("VLM_MAX_TOKENS", 1500, minimum=200),
    "mimo_video_prompt": os.environ.get(
        "MIMO_VIDEO_PROMPT",
        "请用中文分析这个视频分片的主要人物、场景变化、关键动作、情绪走向和剧情冲突，"
        "重点提取适合写短视频解说的故事线索。不要泛泛复述画面，要标出对后续写稿有用的信息。",
    ),
    "mimo_disable_thinking": env_bool("MIMO_DISABLE_THINKING", True),
    "mimo_disable_thinking_source": "env" if os.environ.get("MIMO_DISABLE_THINKING") else "default",
    "fps": 0,  # 0 = 自动（≤60s→2fps, ≤5min→1.5fps, >5min→1fps）
    # TTS 语速（字符/秒）。实测 mimo-tts 冰糖音色中位 ~3.9 字/秒，可用 SPEECH_RATE 覆盖
    # 生成解说时使用 speech_rate * safety_margin 作为约束
    "speech_rate": env_float("SPEECH_RATE", 3.9, minimum=0.5),  # 旧值 3.5 系统性偏低 ~10-17%
    "speech_safety_margin": env_float("SPEECH_SAFETY_MARGIN", 0.85, minimum=0.1),  # 保守系数：TTS 实际语速有 ±20% 波动
    # Block-coverage lint thresholds — promoted from inline .get() literals to real CONFIG keys (tunable; defaults unchanged)
    "narration_coverage_target": 0.7,   # aim ~70% narrated:original (7:3)
    "narration_coverage_min": 0.5,      # below this coverage → under_narrated
    "narration_block_seconds": 9.0,     # block cadence used to derive target block count
    "original_block_min_seconds": 2.5,  # a deliberate original-audio gap must be at least this long
    "narration_block_min_chars": 16,    # below this avg block size → fragmented_beats
    "fade_ms": env_int("FADE_MS", 120, minimum=0),  # 每段 TTS 淡入淡出(ms)；过大会让紧凑的句子一顿一顿，120ms 防爆音又不发闷
    "breath_ms": 250,  # 段间呼吸空间(ms)；block recap 块内连贯、块间留原声呼吸
    # Legacy single-pass cut mapping density fields; current writing uses block coverage controls below.
    "target_segments_per_minute": 9.6,   # legacy single-pass cut mapping report only; block recap uses narration_coverage_*
    "min_segments_per_minute": 6.24,     # legacy single-pass cut mapping report only
    "max_narration_gap_seconds": 11.0,   # legacy single-pass cut mapping report only
    "ducking_mode": "fixed",  # fixed | sidechaincompress | none
    "ducking_threshold": 0.15,
    "ducking_ratio": 3,
    "ducking_attack": 10,
    "ducking_release": 300,
    "ducking_level_sc": 2.0,
    "ducking_makeup": 1.2,
    "ducking_narr_weight": 1.5,
    "ducking_orig_volume": env_float("DUCKING_ORIG_VOLUME", 0.3, minimum=0.0),  # 解说时原声基准音量
    "foreign_source_audio": _foreign_source_audio,  # 原声语言≠解说语言：解说下原声压到近静音(消除"怪音"双语重叠)
    "zone_ducking_volume": env_float("ZONE_DUCKING_VOLUME",
        _foreign_under_narration_volume if _foreign_source_audio else 0.12, minimum=0.0),  # 解说时原声压低到的音量
    "zone_fade_seconds": 0.5,      # 解说/原声切换的淡入淡出时长(秒)
    "idle_orig_volume": env_float("IDLE_ORIG_VOLUME", 1.0, minimum=0.0),  # 解说块之间的"原声块"音量：默认满音量(1.0)，让精彩原声整段放出来，不被压低（用户要求解说成块、原声也成块）
    "duck_fade_seconds": env_float("DUCK_FADE_SECONDS", 0.3, minimum=0.0),  # 解说块/原声块切换的淡入淡出(秒)，略放宽到 0.3 让满音量↔压低的过渡更顺
    "duck_bridge_seconds": env_float("DUCK_BRIDGE_SECONDS", 1.5, minimum=0.0),  # 仅把间隔小于此值的相邻解说窗口并成一段压低；超过则视为作者特意留的"原声块"，原声放回满音量。默认 1.5s：解说块内部连续压低，块与块之间的留白放出满音量原声（约 7:3 的解说/原声节奏）。调大→更连续铺底、原声块更少；调小→更碎
    "bgm_path": os.environ.get("BGM_PATH", "").strip(),  # 背景音乐文件(可选)，留空则不加 BGM
    "source_video": os.environ.get("SOURCE_VIDEO", "").strip(),  # 剪辑模式下的原始视频(可选)，用于时间线/剪映导出引用原片片段
    "export_jianying": env_bool("EXPORT_JIANYING", False),  # 渲染后可选导出剪映草稿(默认关；与核心解耦)
    "jianying_draft_dir": os.environ.get("JIANYING_DRAFT_DIR", "").strip(),  # 剪映草稿输出父目录(留空=work_dir)
    "jianying_bundle_media": env_bool("JIANYING_BUNDLE_MEDIA", True),  # 默认开：macOS 剪映沙箱读不到外部路径，须把素材拷进草稿目录
    "bgm_volume": env_float("BGM_VOLUME", 0.18, minimum=0.0),  # BGM 铺底音量
    "bgm_ducking_volume": env_float("BGM_DUCKING_VOLUME", 0.10, minimum=0.0),  # 旁白时 BGM 压低到的音量
    "narration_speed": env_float("NARRATION_SPEED", 1.3, minimum=0.5),  # 解说整体提速(atempo)，默认偏快适配短视频；长片可设 1.0
    "mask_source_subtitles": env_bool("MASK_SOURCE_SUBTITLES", True),  # 遮挡原片烧录字幕（默认开；无烧录字幕素材设 false）
    "source_subtitle_mask_ratio": env_float("SOURCE_SUBTITLE_MASK_RATIO", 0.14, minimum=0.0),  # 底部遮挡比例
    "narration_delay_seconds": 1.5,  # 解说延迟放置秒数，让画面先出现再解说（仅用于段落起点）
    "narration_tighten": env_bool("NARRATION_TIGHTEN", True),  # 段落内把句子紧贴上一句实际收尾播放，句间间隔稳定≤tight_pause，杜绝"一句解说一段空白"的卡顿
    "narration_run_gap_seconds": env_float("NARRATION_RUN_GAP_SECONDS", 1.6, minimum=0.0),  # 作者留白超过此值=新段落（让精彩原声透出）；小于则视为同一连续段落
    "narration_tight_pause_seconds": env_float("NARRATION_TIGHT_PAUSE_SECONDS", 0.35, minimum=0.0),  # 段落内句间固定间隔(秒)
    "narration_max_pull_seconds": env_float("NARRATION_MAX_PULL_SECONDS", 1.2, minimum=0.0),  # 收紧时一句最多比作者标注提前的秒数（漂移上限，越小越贴画面）
    "narration_tail_pad_seconds": 0.1,  # 解说尾部最少留白；短 slot 会自动压低 delay 避免截断
    "quiet_overlap_min_ratio": 0.8,  # 解说段至少多少比例落在安静窗口内才标记为非对白重叠
    "visual_beat_max_seconds": 18.0,  # 单段解说超过该时长且跨多个帧锚点时给 lint 提醒
    "visual_beat_max_facts": 3,  # 单段解说最多建议覆盖的 frame_facts 锚点数量
    "asr_chunk_min_chars": env_int("ASR_CHUNK_MIN_CHARS", 500, minimum=1),  # brief 中 ASR 写作分块最小字数/词数
    "asr_chunk_max_chars": env_int("ASR_CHUNK_MAX_CHARS", 800, minimum=1),  # brief 中 ASR 写作分块最大字数/词数
    "speech_ducking_volume": env_float("SPEECH_DUCKING_VOLUME",
        _foreign_under_narration_volume if _foreign_source_audio else 0.2, minimum=0.0),    # 解说与对白重叠时原声音量
    "silence_noise_threshold": "-25dB",  # ffmpeg silencedetect 噪声阈值
    "silence_min_duration": 0.3,     # 静音最短持续秒数
    "quiet_window_min": 1.0,         # 可放解说的安静窗口最短秒数
    "silence_merge_gap": 0.5,        # 相邻静音段间隔<此值时合并
    "scene_merge_min": 4.0,         # 场景合并最短时长，<此值的场景合并到相邻场景
    "scene_junk_filter": env_bool("SCENE_JUNK_FILTER", True),  # 过滤连续黑/白帧无效过渡场景
    "scene_junk_dark_luma": env_float("SCENE_JUNK_DARK_LUMA", 8.0, minimum=0.0),
    "scene_junk_bright_luma": env_float("SCENE_JUNK_BRIGHT_LUMA", 245.0, minimum=0.0),
    "scene_junk_pixel_ratio": env_float("SCENE_JUNK_PIXEL_RATIO", 0.995, minimum=0.0),
    "context_info": "",              # 额外上下文（节目名、角色名等）
    "context_info_source": "default",
    "fps_source": "default",
    "style": "纪录片",               # 解说风格（resume 时随 run_settings 持久化/恢复）
    "style_source": "default",
    "tts_dynamic_params": True,  # 启用动态语速调节
    "vlm_workers": env_int("VLM_WORKERS", 8, minimum=1),  # VLM 并行分析线程数
    "tts_workers": env_int("TTS_WORKERS", 4, minimum=1),  # TTS 并行合成线程数
    "tts_timeout": env_int("TTS_TIMEOUT", 90, minimum=1),  # 单段 TTS 命令超时秒数
    "tts_retries": env_int("TTS_RETRIES", 3, minimum=1),  # 单段 TTS 失败重试次数
    "allow_partial_tts": env_bool("ALLOW_PARTIAL_TTS", False),
    "edit_mode": os.environ.get("EDIT_MODE", "full"),  # full | cut
    "edit_mode_source": "env" if os.environ.get("EDIT_MODE") else "default",
    "target_duration": os.environ.get("TARGET_DURATION", ""),  # cut 模式目标成片时长，如 10m
    "target_duration_source": "env" if os.environ.get("TARGET_DURATION") else "default",
    "clip_padding": env_float("CLIP_PADDING", 0.0, minimum=0.0),  # cut 模式片段两端扩展秒数
    "clip_padding_source": "env" if os.environ.get("CLIP_PADDING") else "default",
    "allow_clip_overlap": env_bool("ALLOW_CLIP_OVERLAP", False),  # cut 模式是否允许重复/重叠使用原片
    "burn_subtitles": env_bool("BURN_SUBTITLES", True),  # 烧录解说字幕（默认开；遮挡原字幕后需自带字幕，否则字幕区空白）
    "subtitle_original_in_gaps": env_bool("SUBTITLE_ORIGINAL_IN_GAPS", True),  # 原声留白处补烧原声台词字幕（来自 ASR）
    "force_video_reencode": env_bool("FORCE_VIDEO_REENCODE", False),  # 组装时重编码视频，修复部分容器时间戳问题
    # 成片压制（仅在重编码时生效：烧字幕/遮罩/缩放/FORCE_VIDEO_REENCODE 任一触发重编码）。
    "output_crf": env_int("OUTPUT_CRF", 18, minimum=0),          # x264 CRF；越大文件越小、画质越低（18≈视觉无损，23~26 体积更小）
    "output_preset": os.environ.get("OUTPUT_PRESET", "veryfast"),  # x264 preset；slow/slower 同 CRF 下体积更小但更慢
    "output_max_height": env_int("OUTPUT_MAX_HEIGHT", 0, minimum=0),  # >0 时把成片高度上限缩到该值(保持宽高比、偶数宽)；0=不缩放
    # 成片末端整体响度归一（默认混音偏轻，归一后更接近常见短视频响度；样片约 -11.9，默认取更安全的 -14）
    "final_loudnorm": env_bool("FINAL_LOUDNORM", True),  # 组装末端做一次整体响度归一
    "target_lufs": env_float("TARGET_LUFS", -14.0),       # 目标综合响度 (LUFS)
    "target_true_peak": env_float("TARGET_TRUE_PEAK", -1.0),  # 目标真峰值 (dBTP)
    "target_lra": env_float("TARGET_LRA", 11.0),          # 目标响度范围 (LU)
    "subtitle_font_name": os.environ.get("SUBTITLE_FONT_NAME", "Arial"),
    "subtitle_font_size": env_int("SUBTITLE_FONT_SIZE", 42, minimum=8),
    "subtitle_primary_color": os.environ.get("SUBTITLE_PRIMARY_COLOR", "&H00FFFFFF"),
    "subtitle_outline_color": os.environ.get("SUBTITLE_OUTLINE_COLOR", "&H00000000"),
    "subtitle_outline": env_float("SUBTITLE_OUTLINE", 2.0, minimum=0.0),
    "subtitle_shadow": env_float("SUBTITLE_SHADOW", 1.0, minimum=0.0),
    "subtitle_margin_v": env_int("SUBTITLE_MARGIN_V", 48, minimum=0),
    "subtitle_margin_l": env_int("SUBTITLE_MARGIN_L", 40, minimum=0),
    "subtitle_margin_r": env_int("SUBTITLE_MARGIN_R", 40, minimum=0),
    "subtitle_alignment": env_int("SUBTITLE_ALIGNMENT", 2, minimum=1),
    "subtitle_max_chars": env_int("SUBTITLE_MAX_CHARS", 20, minimum=6),
    "subtitle_play_res_x": env_int("SUBTITLE_PLAY_RES_X", 1280, minimum=1),
    "subtitle_play_res_y": env_int("SUBTITLE_PLAY_RES_Y", 720, minimum=1),
}

SCRIPT_DIR = Path(__file__).parent
PROMPTS_DIR = SCRIPT_DIR.parent / "references"

def log(msg):
    print(f"[video-recap] {msg}", flush=True)

def run_cmd(cmd, **kwargs):
    """运行命令，返回 CompletedProcess"""
    if isinstance(cmd, list):
        display_parts = []
        for part in cmd:
            text = str(part)
            display_parts.append(text if len(text) <= 240 else text[:237] + "...")
        display = " ".join(display_parts)
    else:
        display = str(cmd)
        if len(display) > 2000:
            display = display[:1997] + "..."
    log(f"运行: {display}")
    return subprocess.run(cmd, capture_output=True, text=True, **kwargs)

def get_video_duration(video_path):
    """获取视频时长（秒）"""
    cmd = ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
           "-of", "csv=p=0", str(video_path)]
    result = run_cmd(cmd)
    if result.returncode != 0:
        return 0.0
    try:
        return float(result.stdout.strip())
    except (TypeError, ValueError):
        return 0.0

"""Multi-track timeline model for the recap (backend-neutral, stdlib only).

A `Timeline` is a small, serializable representation of the finished recap as a
set of tracks — exactly like a cut-tool project:

  - one **video** track: the source clip(s), each carrying its own *original
    audio* with a per-clip volume automation (the ducking: a continuous low bed
    under narration, held across short inter-sentence gaps, back up only at the
    lead-in/out and genuine long gaps);
  - one **narration** audio track: the placed TTS beats;
  - an optional **bgm** audio track: a looped music bed with its own ducking;
  - one **subtitle** (text) track: the narration lines.

The canonical renderer is still ffmpeg (`assemble.py`); this model is what it
emits as `timeline.json` and what the *optional* 剪映 exporter consumes. The
model itself knows nothing about ffmpeg or 剪映 — times are plain seconds and
volumes are plain gains, so any backend can read it.
"""

import json

SCHEMA_VERSION = 1


def _kf(t_s, gain):
    return {"t": round(float(t_s), 4), "gain": round(float(gain), 4)}


def ducking_keyframes(windows, idle, duck, fade, span_start, span_end, bridge=None):
    """Volume automation for a track that holds at `idle` and dips to `duck`
    under each narration window, with `fade`-second linear ramps.

    `windows` is a list of (start_s, end_s) narration spans (timeline-absolute).
    Returns timeline-absolute [{t, gain}] keyframes clamped to [span_start,
    span_end]; empty when there is nothing to automate (caller uses a flat gain).
    """
    if bridge is None:
        bridge = 2 * fade
    rel = sorted((max(span_start, w[0]), min(span_end, w[1]))
                 for w in windows if w[1] > span_start and w[0] < span_end and w[1] > w[0])
    if not rel:
        return []
    # Coalesce windows whose gap is below `bridge`: with no room to ramp back to idle and
    # down again between them, the duck must stay held — otherwise the original would pump
    # up and back down between sentences. Genuine gaps (>= bridge) survive as a plateau, so
    # the original still swells there.
    merged = [list(rel[0])]
    for s, e in rel[1:]:
        if s - merged[-1][1] < bridge:
            merged[-1][1] = max(merged[-1][1], e)
        else:
            merged.append([s, e])
    pts = [(span_start, idle)]
    for s, e in merged:
        # hold the duck across the merged window; ramp in just before, release just after
        pts.append((max(span_start, s - fade), idle))
        pts.append((s, duck))
        pts.append((e, duck))
        pts.append((min(span_end, e + fade), idle))
    pts.append((span_end, idle))
    # sort by time, collapse points at the same instant (last wins for a clean step)
    pts.sort(key=lambda p: p[0])
    out = []
    for t, g in pts:
        if out and abs(out[-1][0] - t) < 1e-4:
            out[-1] = (t, g)
        else:
            out.append((t, g))
    return [_kf(t, g) for t, g in out]


def _coalesce_windows(rel, bridge):
    """Merge sorted (start, end, level) windows whose gap is below `bridge` into one held
    span at the most-ducked (min) level. This is the SAME coalescing the ffmpeg render path
    uses (assemble._coalesce_duck_windows), so the exported 剪映 draft matches the rendered
    mix exactly even when a bridged span mixes speech (louder) and quiet (deeper) beats."""
    if not rel:
        return []
    merged = [list(rel[0])]
    for s, e, level in rel[1:]:
        if s - merged[-1][1] < bridge:
            merged[-1][1] = max(merged[-1][1], e)
            merged[-1][2] = min(merged[-1][2], level)
        else:
            merged.append([s, e, level])
    return merged


def variable_ducking_keyframes(windows, idle, fade, span_start, span_end, bridge=None):
    """Volume automation with a per-window duck gain.

    `windows` is [(start_s, end_s, duck_gain)]. Windows whose gap is below `bridge` coalesce
    into one held span at the most-ducked level (matching the renderer), so the original stays
    ducked across short inter-sentence gaps instead of swelling back to idle; only the
    lead-in/out and genuine gaps >= bridge return to idle. Defaults bridge to 2*fade.
    """
    if bridge is None:
        bridge = 2 * fade
    rel = sorted(
        (max(span_start, float(w[0])), min(span_end, float(w[1])), float(w[2]))
        for w in windows
        if float(w[1]) > span_start and float(w[0]) < span_end and float(w[1]) > float(w[0])
    )
    merged = _coalesce_windows(rel, bridge)
    if not merged:
        return []

    pts = [(span_start, idle)]
    for s, e, level in merged:
        # hold the duck across the merged span; ramp in just before, release just after
        pts.append((max(span_start, s - fade), idle))
        pts.append((s, level))
        pts.append((e, level))
        pts.append((min(span_end, e + fade), idle))
    pts.append((span_end, idle))

    pts.sort(key=lambda p: p[0])
    out = []
    for t, g in pts:
        if out and abs(out[-1][0] - t) < 1e-4:
            # coincident points (overlapping ramps): prefer the lower gain, no swell
            out[-1] = (t, min(out[-1][1], g))
        else:
            out.append((t, g))
    return [_kf(t, g) for t, g in out]


def build_timeline(canvas, duration_s, video_clips, narration_segments,
                   bgm=None, ducking=None, subtitle_segments=None):
    """Assemble a Timeline dict from resolved placement data.

    canvas: {"width", "height", "fps"}
    duration_s: total output length (seconds)
    video_clips: ordered [{"source_path", "source_start", "source_end",
                 "timeline_start", "timeline_end"}] (cut mode: one per clip;
                 full mode: a single clip spanning the whole video).
    narration_segments: placed beats [{"source_path", "timeline_start",
                 "timeline_end", "text", "overlaps_speech", "gain"?}].
    subtitle_segments: optional display-ready text cues [{"text", "timeline_start",
                 "timeline_end"}]. When present, this is authoritative for the
                 subtitle/text track; narration segment text remains raw editor metadata.
    bgm: optional {"source_path", "volume", "ducking_volume"}.
    ducking: {"idle", "speech", "quiet", "fade", "bridge"?} for the original-audio
             automation; None disables original ducking (flat original). `bridge` holds
             the duck across inter-beat gaps shorter than it (defaults to 2*fade).
    """
    windows = [(float(s["timeline_start"]), float(s["timeline_end"]))
               for s in narration_segments
               if s.get("timeline_end", 0) > s.get("timeline_start", 0)]
    duck_windows = [
        (
            float(s["timeline_start"]),
            float(s["timeline_end"]),
            float((ducking or {}).get("speech" if s.get("overlaps_speech", True) else "quiet", 1.0)),
        )
        for s in narration_segments
        if s.get("timeline_end", 0) > s.get("timeline_start", 0)
    ]

    # --- video track: each clip carries its original audio + ducking automation
    video_clip_objs = []
    for c in video_clips:
        ts, te = float(c["timeline_start"]), float(c["timeline_end"])
        audio = {"role": "original", "volume_keyframes": []}
        if ducking is not None:
            audio["volume_keyframes"] = variable_ducking_keyframes(
                duck_windows, ducking["idle"], ducking["fade"], ts, te,
                bridge=ducking.get("bridge"))
            audio["base_gain"] = round(float(ducking["idle"]), 4)
        else:
            audio["base_gain"] = 1.0
        video_clip_objs.append({
            "source_path": c["source_path"],
            "source_start": round(float(c["source_start"]), 4),
            "source_end": round(float(c["source_end"]), 4),
            "timeline_start": round(ts, 4),
            "timeline_end": round(te, 4),
            "audio": audio,
        })

    tracks = [{"kind": "video", "name": "video", "clips": video_clip_objs}]

    # --- narration track
    narr_segs = []
    for s in narration_segments:
        ts, te = float(s["timeline_start"]), float(s["timeline_end"])
        if te <= ts:
            continue
        narr_segs.append({
            "source_path": s["source_path"],
            "timeline_start": round(ts, 4),
            "timeline_end": round(te, 4),
            "gain": round(float(s.get("gain", 1.0)), 4),
            "text": s.get("text", ""),
            "overlaps_speech": bool(s.get("overlaps_speech", True)),
        })
    if narr_segs:
        tracks.append({"kind": "audio", "name": "narration", "role": "narration",
                       "segments": narr_segs})

    # --- bgm track (optional, looped, ducked under narration)
    if bgm and bgm.get("source_path"):
        base = float(bgm.get("volume", 0.18))
        duck = float(bgm.get("ducking_volume", 0.10))
        fade = float(bgm.get("fade") or (ducking or {}).get("fade", 0.25))
        kfs = ducking_keyframes(windows, base, duck, fade, 0.0, duration_s,
                                bridge=(ducking or {}).get("bridge"))
        tracks.append({
            "kind": "audio", "name": "bgm", "role": "bgm", "loop": True,
            "segments": [{
                "source_path": bgm["source_path"],
                "timeline_start": 0.0,
                "timeline_end": round(float(duration_s), 4),
                "gain": round(base, 4),
                "volume_keyframes": kfs,
            }],
        })

    # --- subtitle (text) track
    text_source = subtitle_segments if subtitle_segments is not None else narration_segments
    text_segs = []
    for s in text_source or []:
        if not isinstance(s, dict) or not s.get("text"):
            continue
        try:
            ts = float(s["timeline_start"])
            te = float(s["timeline_end"])
        except (KeyError, TypeError, ValueError):
            continue
        if te <= ts:
            continue
        text_segs.append({
            "text": s.get("text", ""),
            "timeline_start": round(ts, 4),
            "timeline_end": round(te, 4),
        })
    if text_segs:
        tracks.append({"kind": "text", "name": "subtitle", "segments": text_segs})

    return {
        "schema_version": SCHEMA_VERSION,
        "canvas": {"width": int(canvas["width"]), "height": int(canvas["height"]),
                   "fps": float(canvas.get("fps", 30))},
        "duration": round(float(duration_s), 4),
        "tracks": tracks,
    }


def save_timeline(timeline, path):
    with open(path, "w", encoding="utf-8") as f:
        json.dump(timeline, f, ensure_ascii=False, indent=2)
    return path


def load_timeline(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)