app.py

"""AutoCut Studio: a presentation-ready UI over the existing agent pipeline."""
from __future__ import annotations

import json
import os
import re
import shutil
import threading
import uuid
from pathlib import Path
from typing import Any

import requests

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

ROOT = Path(__file__).resolve().parent
RUNS = ROOT / "runs"
RUNS.mkdir(exist_ok=True)
app = FastAPI(title="AutoCut Studio")
app.mount("/static", StaticFiles(directory=ROOT / "static"), name="static")
jobs: dict[str, dict[str, Any]] = {}
# Runtime data caches are intentionally shared by the legacy agents.  Serialize
# jobs so a new cleanup cannot delete an in-flight job's temporary artifacts.
JOB_RUN_LOCK = threading.Lock()
TOKEN_PLAN_BASE = "https://token-plan.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
TOKEN_PLAN_MODEL = "deepseek-v3.2"

_REGENERABLE_CACHE_DIRS = (
    "audio",
    "subtitles",
    "parse_results",
    "highlight_results",
    "verification_frames",
)
_REGENERABLE_CACHE_FILES = (
    "latest_verification_job.json",
    "miss_verification_job.json",
    "two_point_verification_job.json",
)


def _clear_regenerable_cache() -> dict[str, int]:
    """Remove stale derived data before a job without deleting user outputs.

    The original uploads under ``runs/``, completed videos under ``output/``,
    static assets and downloaded model weights are deliberately preserved.
    Only files that every run can recreate are cleared.
    """
    temp_root = ROOT / "temp"
    cleared_dirs = 0
    cleared_files = 0
    for name in _REGENERABLE_CACHE_DIRS:
        path = temp_root / name
        if path.exists():
            shutil.rmtree(path)
            cleared_dirs += 1
        path.mkdir(parents=True, exist_ok=True)
    for name in _REGENERABLE_CACHE_FILES:
        path = temp_root / name
        if path.is_file():
            path.unlink()
            cleared_files += 1
    # MoviePy occasionally leaves a temporary audio mux file beside the app.
    for path in ROOT.glob("*_TEMP_MPY_*"):
        if path.is_file():
            path.unlink()
            cleared_files += 1
    return {"directories": cleared_dirs, "files": cleared_files}


def _reweight_candidates_with_scoreboard(
    candidates: list[dict[str, Any]], evidence_by_candidate: dict[str, dict[str, Any]],
) -> list[dict[str, Any]]:
    """Make confirmed score changes dominate ranking; keep sound auxiliary."""
    from config import settings

    weighted: list[dict[str, Any]] = []
    for original in candidates:
        candidate = dict(original)
        evidence = evidence_by_candidate.get(str(candidate.get("candidate_id")), {})
        audio = max(0.0, min(1.0, float(candidate.get("audio_score", 0.0) or 0.0)))
        text = max(0.0, min(1.0, float(candidate.get("text_score", 0.0) or 0.0)))
        score_confidence = max(0.0, min(1.0, float(evidence.get("confidence", 0.0) or 0.0)))
        if evidence.get("score_verified"):
            combined = (
                settings.verified_score_weight * score_confidence
                + settings.verified_score_text_weight * text
                + settings.verified_score_audio_weight * audio
            )
            candidate["ranking_evidence"] = "stable_scoreboard"
        elif evidence.get("score_supported"):
            # A nearby legal score change with one readable after-state is
            # weaker than a stable OCR transition, but it still takes priority
            # over incomplete subtitles or crowd-noise guesses.  It remains
            # visibly labelled as pending review in the timeline.
            combined = 0.78 * score_confidence + 0.16 * text + 0.06 * audio
            candidate["ranking_evidence"] = "scoreboard_supported"
        else:
            # For misses and non-scoring actions there is no score delta.  Text
            # remains useful, while a crowd swell stays a minor signal.
            combined = settings.candidate_text_weight * text + settings.candidate_audio_weight * audio
            candidate["ranking_evidence"] = "text_with_auxiliary_audio"
        candidate["combined_score"] = round(max(0.0, min(1.0, combined)), 3)
        weighted.append(candidate)
    return sorted(
        weighted,
        key=lambda item: (
            bool(evidence_by_candidate.get(str(item.get("candidate_id")), {}).get("score_verified")),
            bool(evidence_by_candidate.get(str(item.get("candidate_id")), {}).get("score_supported")),
            float(item.get("combined_score", 0.0) or 0.0),
        ),
        reverse=True,
    )


def _relative(path: str | Path) -> str:
    return str(Path(path).resolve().relative_to(ROOT.resolve())).replace("\\", "/")


def _chat_completion(api_key: str, messages: list[dict[str, str]], temperature: float = 0.35) -> str:
    """Call Token Plan transiently; the supplied API key is never written to disk."""
    response = requests.post(
        f"{TOKEN_PLAN_BASE}/chat/completions",
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        json={"model": TOKEN_PLAN_MODEL, "messages": messages, "temperature": temperature, "max_tokens": 700},
        timeout=45,
    )
    response.raise_for_status()
    return str(response.json()["choices"][0]["message"]["content"]).strip()


def _request_excludes(text: str, aliases: tuple[str, ...]) -> bool:
    """Return whether a natural-language brief explicitly excludes an event.

    A phrase such as ``不包括罚球`` is not a request *for* free throws.  Keep
    this parsing local and deterministic so an exclusion never turns an
    otherwise populated score pool into an empty one.
    """
    exclusion_prefix = r"(?:不包括|不含|排除|去除|去掉|不要|除外|without|excluding|exclude|no)"
    for alias in aliases:
        pattern = rf"{exclusion_prefix}[\s，,、:：()（）的]*(?:任何|所有|the\s*)?[\s，,、:：()（）的]*{re.escape(alias)}"
        if re.search(pattern, text, re.I):
            return True
    return False


def _structured_event_filter(candidates: list[dict[str, Any]], request: str) -> list[dict[str, Any]]:
    """Apply unambiguous event requests before asking the LLM to rank clips."""
    text = request.lower()
    types: set[str] = set()
    excluded_types: set[str] = set()
    outcomes: set[str] = set()
    candidate_sports = {
        str(candidate.get("sport")) for candidate in candidates
        if str(candidate.get("sport")) in {"basketball", "football"}
    }
    basketball_request = "篮球" in text or "basketball" in text or candidate_sports == {"basketball"}
    football_request = "足球" in text or "football" in text or candidate_sports == {"football"}
    if "三分" in text or "three" in text:
        types.add("three_point")
    if "两分" in text or "two point" in text:
        types.add("two_point")
    if "罚球" in text or "free throw" in text:
        if _request_excludes(text, ("罚球", "free throw", "free throws")):
            excluded_types.add("free_throw")
        else:
            types.add("free_throw")
    if "扣篮" in text or "dunk" in text:
        types.add("dunk")
    if "进球" in text:
        # “进球” is sport-dependent in Chinese.  In a basketball task it
        # means a made basket, not a football goal event.
        if basketball_request and not football_request:
            outcomes.add("made")
        else:
            types.update({"goal", "penalty_goal"})
    if ("得分" in text or "scoring" in text) and basketball_request:
        outcomes.add("made")
    if "goal" in text:
        types.update({"goal", "penalty_goal"})
    if "点球" in text or "penalty" in text:
        types.update({"penalty_goal", "penalty_missed"})
    if "扑救" in text or "save" in text:
        types.add("shot_saved")
    if "射门未进" in text or "射偏" in text or "off target" in text:
        types.update({"shot_missed", "penalty_missed"})
    if "红牌" in text or "red card" in text:
        types.add("red_card")
    if "黄牌" in text or "yellow card" in text:
        types.add("yellow_card")
    if "越位" in text or "offside" in text:
        types.add("offside")
    if any(term in text for term in ("打铁", "没进", "不中", "未命中", "miss", "no good", "brick", "airball", "short")):
        outcomes.add("missed")
    if any(term in text for term in ("命中", "投进", "打进", "made", "makes", "it's good", "its good", "good!")):
        outcomes.add("made")
    if not types and not outcomes and not excluded_types:
        return []
    # For a narrow request such as “only made threes”, require a stable score
    # transition.  For “all basketball scores”, keep clearly-commentated made
    # baskets too; verified score changes still outrank them in the planner.
    strict_make_terms = ("只", "仅", "命中", "投进", "打进", "made", "makes", "it's good", "its good", "good!")
    requires_verified_make = outcomes == {"made"} and any(term in text for term in strict_make_terms)
    return [
        candidate for candidate in candidates
        if (not types or str(candidate.get("event_type")) in types)
        and str(candidate.get("event_type")) not in excluded_types
        and (not outcomes or str(candidate.get("event_outcome")) in outcomes)
        and (not requires_verified_make or bool(candidate.get("score_verified")))
    ]


def _discover_penalty_goal_fallback(
    subtitle_items: list[dict[str, Any]], audio_segments: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], str]:
    """Create a conservative long-form penalty candidate when OCR misses +1.

    Some broadcast overlays animate during the first penalty score, making the
    0→1 transition unreadable even when the later score is clear.  A penalty
    setup phrase followed by the actual taker's commentary and a cluster of
    crowd-energy peaks is strong enough to retain that sequence for a goal
    montage, while being explicitly marked as lower-confidence evidence.
    """
    penalty_re = re.compile(r"\b(penalty|spot kick)\b|点球", re.I)
    taking_re = re.compile(
        r"\b(convert|takes? it|what will he do|top corner|from the spot|steps up)\b|主罚|罚球|操刀",
        re.I,
    )
    penalty_times = [float(item.get("start_sec", 0.0)) for item in subtitle_items if penalty_re.search(str(item.get("text", "")))]
    surges = [
        item for item in audio_segments
        if bool(item.get("loudness_surge_flag"))
    ]
    if not penalty_times or not surges:
        return [], {}, ""

    candidates: list[dict[str, Any]] = []
    evidence: dict[str, dict[str, Any]] = {}
    total_sec = max((float(item.get("end_sec", 0.0)) for item in audio_segments), default=0.0)
    for item in subtitle_items:
        text = str(item.get("text", ""))
        setup_start = float(item.get("start_sec", 0.0))
        if not taking_re.search(text) or not any(0.0 <= setup_start - penalty_time <= 150.0 for penalty_time in penalty_times):
            continue
        nearby = [
            segment for segment in surges
            if setup_start - 3.0 <= float(segment.get("start_sec", 0.0)) <= setup_start + 35.0
        ]
        if len(nearby) < 2:
            continue
        start = max(0.0, min(setup_start, float(nearby[0]["start_sec"])) - 2.0)
        end = min(total_sec, max(float(nearby[-1]["end_sec"]) + 4.0, float(item.get("end_sec", 0.0)) + 12.0))
        if end - start < 14.0:
            continue
        candidate_id = f"HL_PENALTY_{len(candidates) + 1:03d}"
        related = [
            int(sub.get("index", 0)) for sub in subtitle_items
            if float(sub.get("end_sec", 0.0)) >= start and float(sub.get("start_sec", 0.0)) <= end
        ]
        candidates.append({
            "candidate_id": candidate_id,
            "start_sec": round(start, 3), "end_sec": round(end, 3),
            "duration_sec": round(end - start, 3), "trigger": "fused",
            "audio_score": 0.92, "text_score": 0.90, "combined_score": 0.93,
            "matched_keywords": ["penalty", "goal_sequence"],
            "related_subtitle_indices": related, "llm_validated": True,
            "llm_reason": "点球解说链与连续现场高能反应形成的进球序列。",
        })
        evidence[candidate_id] = {
            "team": "待确认", "confidence": 0.62, "source": "penalty_commentary_fallback",
            "sport": "football", "event_hint": "penalty_goal",
            "reason": "点球准备、主罚解说与后续连续高能反应，比分 0→1 OCR 未连续读到。",
        }
        # The same penalty may be mentioned in several adjacent subtitle lines.
        break
    return candidates, evidence, f"点球序列回退补充 {len(candidates)} 个候选。" if candidates else ""


def _discover_basketball_miss_fallback(
    subtitle_items: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], str]:
    """Separate a missed attempt from a made basket that immediately follows it.

    A fast-break miss followed seconds later by a put-back can otherwise become
    one broad audio candidate.  Scoreboard OCR then correctly sees the later
    +2 but incorrectly labels the whole combined window as a made basket,
    hiding the preceding miss from a "打铁" request.
    """
    miss_re = re.compile(
        r"\b(miss(?:es|ed)?|no good|comes up short|won'?t go|off the rim|"
        r"rimmed out|brick|air ?ball)\b|打铁|不中|没进",
        re.I,
    )
    made_re = re.compile(
        r"\b(up and good|it'?s good|good!|makes?|scores?|lays? it in|finishes|"
        r"gets? it to go|it'?s good)\b|命中|进了|好球",
        re.I,
    )
    ordered = sorted(subtitle_items, key=lambda item: float(item.get("start_sec", 0.0)))
    candidates: list[dict[str, Any]] = []
    evidence: dict[str, dict[str, Any]] = {}
    for position, item in enumerate(ordered):
        text = str(item.get("text", ""))
        if not miss_re.search(text):
            continue
        miss_start = float(item.get("start_sec", 0.0))
        miss_end = float(item.get("end_sec", miss_start))
        next_made = next(
            (
                later for later in ordered[position + 1:]
                if miss_end <= float(later.get("start_sec", 0.0)) <= miss_end + 8.0
                and made_re.search(str(later.get("text", "")))
            ),
            None,
        )
        # This fallback is only for two distinct plays packed into one short
        # sequence.  An isolated miss is already handled by the normal flow.
        if next_made is None:
            continue
        start = max(0.0, miss_start - 4.5)
        end = max(miss_end + 2.0, float(next_made.get("start_sec", miss_end)) - 0.2)
        if end - start < 4.0:
            continue
        candidate_id = f"HL_MISS_{len(candidates) + 1:03d}"
        related = [
            int(sub.get("index", 0)) for sub in ordered
            if float(sub.get("end_sec", 0.0)) >= start and float(sub.get("start_sec", 0.0)) <= end
        ]
        candidates.append({
            "candidate_id": candidate_id,
            "start_sec": round(start, 3), "end_sec": round(end, 3),
            "duration_sec": round(end - start, 3), "trigger": "fused",
            "audio_score": 0.78, "text_score": 0.94, "combined_score": 0.86,
            "matched_keywords": ["miss", "separated_play"],
            "related_subtitle_indices": related, "llm_validated": True,
            "llm_reason": "相邻得分回合前的明确未命中，已从连续镜头中拆分。",
        })
        evidence[candidate_id] = {
            "team": "待确认", "confidence": 0.78, "source": "commentary_sequence",
            "sport": "basketball", "event_hint": "shot_missed",
            "reason": "字幕明确说明未命中，且后续短时间内出现另一回合的命中解说。",
        }
    return candidates, evidence, f"篮球连续回合拆分 {len(candidates)} 个打铁候选。" if candidates else ""


_BASKETBALL_FIELD_GOAL_ACTION = re.compile(
    r"\b(floater|layup|jumper|hook shot|drive|dunk|slam|three|3[- ]?pointer)\b|"
    r"三分|两分|上篮|跳投|抛投|扣篮",
    re.I,
)
_BASKETBALL_MADE_CALL = re.compile(
    r"\b(it'?s good|up and good|lays? it in|finishes|gets? it to go|makes?|scores?|"
    r"buries|drills|knocks? (?:it )?down)\b|命中|进了|好球",
    re.I,
)
_BASKETBALL_MISS_CALL = re.compile(
    r"\b(miss(?:es|ed)?|no good|comes up short|won'?t go|off the rim|rimmed out|"
    r"brick|air ?ball)\b|打铁|不中|没进",
    re.I,
)
_BASKETBALL_FREE_THROW_CALL = re.compile(r"\bfree throws?\b|罚球", re.I)


def _has_visible_basketball_field_goal(text: str) -> bool:
    """Return whether local commentary supports a made *field goal*.

    A score overlay alone cannot tell a two-pointer from two free throws.
    Attempt words such as ``three`` or ``floater`` are not makes by
    themselves: the last outcome call must be a make, and free-throw
    commentary is deliberately excluded from this field-goal path.
    """
    if _BASKETBALL_FREE_THROW_CALL.search(text):
        return False
    made_matches = list(_BASKETBALL_MADE_CALL.finditer(text))
    if not made_matches:
        return False
    miss_matches = list(_BASKETBALL_MISS_CALL.finditer(text))
    last_made = made_matches[-1]
    last_miss = miss_matches[-1] if miss_matches else None
    if last_miss is not None and last_miss.start() > last_made.start():
        return False
    # "The floater's up and good" already contains attempt and result.  For
    # shorter calls, require an explicit action so generic praise cannot pass
    # as a basket.
    return bool(
        _BASKETBALL_FIELD_GOAL_ACTION.search(text)
        or re.search(r"\b(up and good|lays? it in|finishes|gets? it to go)\b|命中|进了", text, re.I)
    )


def _attach_basketball_score_context(
    candidates: list[dict[str, Any]],
    evidence_by_candidate: dict[str, dict[str, Any]],
    subtitle_items: list[dict[str, Any]],
) -> None:
    """Link score changes to nearby calls and flag unseen +2 updates.

    A scorebug can prove that the score changed even when the source recording
    jumps over the two free throws themselves.  A normal highlight reel should
    not present that discontinuity as a made field goal.  We therefore mark a
    score update as ``event_visible=False`` when its local commentary lacks a
    made field-goal call.  An explicit request for all scoreboard changes can
    still opt into such a candidate later.
    """
    for candidate in candidates:
        candidate_id = str(candidate.get("candidate_id"))
        evidence = evidence_by_candidate.get(candidate_id)
        if not evidence or not (evidence.get("score_verified") or evidence.get("score_supported")):
            continue
        start, end = float(candidate.get("start_sec", 0.0)), float(candidate.get("end_sec", 0.0))
        before_time = float(evidence.get("before_time", start))
        after_time = float(evidence.get("after_time", end))
        # Score candidates intentionally have wide render windows.  Restrict
        # the commentary check to the score transition itself so an unrelated
        # later play cannot turn a score-only cut into a made field goal.  The
        # lead-in still covers ordinary scorebug animation lag.
        context_start = max(start, before_time - 4.5)
        context_end = min(end, after_time + 2.5)
        related = [
            int(item.get("index", 0)) for item in subtitle_items
            if float(item.get("end_sec", 0.0)) >= context_start
            and float(item.get("start_sec", 0.0)) <= context_end
        ]
        candidate["related_subtitle_indices"] = related
        nearby_text = " ".join(
            str(item.get("text", "")) for item in subtitle_items
            if float(item.get("end_sec", 0.0)) >= context_start
            and float(item.get("start_sec", 0.0)) <= context_end
        )
        free_throw_context = bool(_BASKETBALL_FREE_THROW_CALL.search(nearby_text))
        visible_make = _has_visible_basketball_field_goal(nearby_text)
        evidence["event_visible"] = visible_make
        evidence["score_context_start"] = round(context_start, 3)
        evidence["score_context_end"] = round(context_end, 3)
        # Keep the field on both objects.  ``enrich_candidate_events`` also
        # copies it from evidence, but this makes pre-classification selection
        # and debugging deterministic.
        candidate["event_visible"] = visible_make
        if not visible_make:
            # A local free-throw call is enough to label the score sequence as
            # foul shots even when a coarse OCR interval observed it as +2.
            # Other score-only transitions remain available only when the user
            # explicitly asks to prioritise raw scoreboard changes.
            if free_throw_context or evidence.get("event_hint") == "free_throw_pair":
                evidence["event_hint"] = "free_throw_pair"
            else:
                evidence["event_hint"] = "scoreboard_unseen_score"
            evidence["reason"] = (
                f"{evidence.get('reason', '比分牌确认得分')}；本地时间窗口没有可确认的两分/三分命中过程，"
                "默认不作为连续进球镜头。"
            )


def _apply_edit_request(
    candidates: list[dict[str, Any]], request: str, api_key: str | None, target_seconds: float,
    team_evidence: dict[str, dict[str, Any]] | None = None,
) -> tuple[list[dict[str, Any]], str]:
    """Select existing candidates according to an optional natural-language edit brief."""
    if not request.strip() or not candidates:
        return candidates, ""
    structured = _structured_event_filter(candidates, request)
    normalized_request = request.lower()
    strict_terms = (
        "三分", "three", "两分", "two point", "罚球", "free throw", "扣篮", "dunk", "进球", "得分", "scoring", "goal", "点球", "penalty",
        "扑救", "save", "射门未进", "射偏", "off target", "红牌", "red card", "黄牌", "yellow card",
        "越位", "offside", "打铁", "没进", "不中", "未命中", "miss", "no good", "brick", "airball",
        "short", "命中", "投进", "打进", "made", "makes", "it's good", "its good", "good!",
    )
    has_strict_intent = any(term in normalized_request for term in strict_terms)
    known_teams = {
        str(item.get("team", "")).strip().lower()
        for item in (team_evidence or {}).values()
        if str(item.get("team", "")).strip() not in {"", "待确认"}
    }
    has_team_focus = any(team in normalized_request for team in known_teams)
    made_score_terms = (
        "进球", "得分", "scoring", "命中", "投进", "打进", "made", "makes",
        "it's good", "its good", "good!", "三分", "three", "两分", "two point",
        "罚球", "free throw", "扣篮", "dunk",
    )
    score_priority_request = any(term in normalized_request for term in made_score_terms)
    # Scoreboard evidence is the primary recall signal.  A short source often
    # has no clean commentary around a real basket, so a legal +1/+2/+3 must
    # stay in the candidate pool even when ``event_visible`` is false.  The
    # only deterministic exclusion is an explicitly identified/requested
    # free-throw sequence.
    scorebacked_all = [
        candidate for candidate in candidates
        if bool(candidate.get("score_verified") or candidate.get("score_supported"))
        and str(candidate.get("event_outcome")) == "made"
        and not (
            _request_excludes(normalized_request, ("罚球", "free throw", "free throws"))
            and str(candidate.get("event_type")) == "free_throw"
        )
    ]
    scorebacked_structured = [
        candidate for candidate in structured
        if bool(candidate.get("score_verified") or candidate.get("score_supported"))
    ]
    # A score scan is a strong recall signal, but it is not the only valid
    # evidence: an end-of-video scorebug can be readable only once while the
    # commentator clearly calls "the floater's up and good".  Keep those
    # visible, typed basketball makes together with visible score-backed
    # candidates instead of allowing OCR to erase the play.
    visible_basketball_actions = [
        candidate for candidate in structured
        if str(candidate.get("sport")) == "basketball"
        and str(candidate.get("event_outcome")) == "made"
        and candidate.get("event_visible") is not False
        and str(candidate.get("event_type")) in {
            "two_point", "three_point", "dunk", "free_throw", "shot",
        }
    ]

    # Do not let a semantic type mismatch turn a real score change into an
    # EmptyHighlightPool.  For example, a hard cut from 92 to 94 is correctly
    # labelled ``confirmed_score`` instead of a fabricated two-point shot.
    # It remains a better result than returning no video at all when the user
    # asks to prioritize actual scoreboard changes.
    if has_strict_intent and not structured:
        if score_priority_request and scorebacked_all and not has_team_focus:
            return scorebacked_all, "未能从原片确认具体出手类型；已保留比分牌确认的得分变化。"
        return [], "未找到符合剪辑指令的镜头；没有用无关画面替代。"

    pool = structured if has_strict_intent else candidates
    structured_note = "已按结构化剪辑指令严格筛选。" if has_strict_intent else ""

    # For a broad basketball-scoring request, bypass the LLM so it cannot
    # delete a verified score *or* a clearly called field goal that OCR missed.
    # Pacing has already collapsed duplicate representations of one play.
    if has_strict_intent and score_priority_request and not has_team_focus:
        selected_by_id: dict[str, dict[str, Any]] = {}
        for candidate in [*scorebacked_structured, *visible_basketball_actions]:
            selected_by_id.setdefault(str(candidate.get("candidate_id")), candidate)
        if selected_by_id:
            selected = sorted(
                selected_by_id.values(), key=lambda candidate: float(candidate.get("start_sec", 0.0))
            )
            return selected, "已按比分变化优先保留得分候选；字幕与声音用于解释事件、定位出手和识别罚球。"

    # An edit brief such as "所有篮球进球" is a recall request, not an
    # invitation for the language model to delete evidence-backed plays.
    recall_terms = ("所有", "全部", "全场", "all", "every", "each")
    keep_all_structured = (
        has_strict_intent
        and bool(structured)
        and any(term in normalized_request for term in recall_terms)
        and not has_team_focus
    )
    if keep_all_structured:
        return structured, "已保留所有符合指令的已确认事件；AI 文案不会再删减高光。"

    if not api_key:
        return pool, structured_note
    compact = [
        {
            "id": c["candidate_id"],
            "time": f"{c['start_sec']:.1f}-{c['end_sec']:.1f}s",
            "keywords": c.get("matched_keywords", []),
            "reason": c.get("llm_reason", ""),
            "sport": c.get("sport_label", c.get("sport", "通用")),
            "event": c.get("event_label", "待判定"),
            "team_evidence": (team_evidence or {}).get(str(c["candidate_id"]), {}),
        }
        for c in pool
    ]
    prompt = (
        "你是体育视频剪辑助理。根据用户要求，从候选片段中选出最匹配的片段。"
        "sport 与 event 字段均为结构化证据：用户说“三分”仅选篮球三分，说“两分”仅选篮球两分，说“打铁/没进”仅选篮球打铁。"
        "足球中，用户说“进球”只选 event 为进球/点球命中的候选；说“点球”只选点球；说“射门未进/扑救”只选对应足球事件；说“红牌/黄牌/越位”只选对应判罚。"
        "尽可能多选择有证据的同类比分事件来接近目标时长，绝不能只留一段而忽略其他个同类候选。"
        "如果要求指定球队，优先使用 team_evidence：其中 source=scoreboard_ocr 且 confidence 不小于 0.55 时是最高优先级。足球比分牌中的球队缩写与中文队名等价，例如 RMA=皇家马德里、BAR=巴塞罗那。"
        "比分牌证据不足时才参考球衣识别；不能把明确为对方队的候选选入。"
        "只可以返回候选列表中已有的 id；若证据不足则保留全部候选，避免误删。"
        "严格只返回 JSON：{\"candidate_ids\":[\"HL_001\"],\"note\":\"不超过30字的中文说明\"}。\n"
        f"目标成片时长：{target_seconds:.0f}秒。\n用户要求：{request}\n候选：{json.dumps(compact, ensure_ascii=False)}"
    )
    try:
        answer = _chat_completion(api_key, [{"role": "user", "content": prompt}], temperature=0.0)
        match = re.search(r"\{.*\}", answer, re.S)
        result = json.loads(match.group(0) if match else answer)
        ids = {str(item) for item in result.get("candidate_ids", [])}
        selected = [candidate for candidate in pool if candidate["candidate_id"] in ids]
        if not selected:
            return pool, structured_note or "AI 未能可靠区分候选，已保留全部片段。"
        return selected, str(result.get("note", structured_note or "已按你的要求筛选高光。"))
    except Exception:
        return pool, structured_note or "AI 筛选暂不可用，已保留全部候选。"


def _run_job(
    job_id: str, source: Path, target_seconds: float, api_key: str | None, editing_request: str,
) -> None:
    """Run jobs one at a time because their derived cache is intentionally reset."""
    with JOB_RUN_LOCK:
        _run_job_exclusive(job_id, source, target_seconds, api_key, editing_request)


def _run_job_exclusive(
    job_id: str, source: Path, target_seconds: float, api_key: str | None, editing_request: str,
) -> None:
    job = jobs[job_id]
    try:
        job.update(status="running", stage="清理本次剪辑缓存", progress=3)
        job["cache_cleanup"] = _clear_regenerable_cache()
        job.update(status="running", stage="字幕解析与音频特征提取", progress=16)
        # Agent configuration is created at import time, so define the root first.
        os.environ["SPORT_CLIP_ROOT"] = str(ROOT)
        from media_parse_agent import MediaParseAgent
        from highlight_filter_agent import HighlightFilterAgent
        from edit_planner import EditPlannerAgent
        from render_manager import RenderManagerAgent
        from config import settings

        job["compute_device"] = settings.device

        if api_key:
            # Token Plan is OpenAI-compatible.  The pipeline has two optional
            # LLM stages (gap filling and highlight validation); wire the same
            # transient UI token to both without persisting it to disk.
            object.__setattr__(settings, "deepseek_api_key", api_key)
            object.__setattr__(settings, "deepseek_api_base", TOKEN_PLAN_BASE)
            object.__setattr__(settings, "llm_api_key", api_key)
            object.__setattr__(settings, "llm_api_base", TOKEN_PLAN_BASE)
            object.__setattr__(settings, "llm_model_name", settings.deepseek_model)
            os.environ["DEEPSEEK_API_KEY"] = api_key
            os.environ["LLM_API_KEY"] = api_key
        object.__setattr__(settings, "default_montage_duration_sec", target_seconds)

        parsed = MediaParseAgent().process(str(source))
        with open(parsed, encoding="utf-8") as f:
            parsed_data = json.load(f)
        # MediaParseOutput serializes its subtitle list under "subtitles"
        # and uses start_sec/end_sec.  Keep a transcript fallback for older
        # cached parse files so the chat assistant always receives context.
        subtitle_items = parsed_data.get("subtitles") or parsed_data.get("transcript") or []
        subtitle_srt_path = str(parsed_data.get("subtitle_srt_path") or "")
        if not subtitle_srt_path or not Path(subtitle_srt_path).is_file():
            subtitle_srt_path = ""
        subtitle_context = "\n".join(
            f"[{float(item.get('start_sec', item.get('start', 0))):.1f}s-"
            f"{float(item.get('end_sec', item.get('end', 0))):.1f}s] {item.get('text', '')}"
            for item in subtitle_items
            if item.get("text")
        )[:6000]
        job.update(stage="高光候选检测（音频 + 字幕）", progress=42)
        candidates_path = HighlightFilterAgent().process(parsed)
        with open(candidates_path, encoding="utf-8") as f:
            candidates = json.load(f)
        candidate_list = candidates.get("candidates", [])
        from event_classifier import detect_sport, enrich_candidate_events
        detected_sport, sport_confidence = detect_sport(candidate_list, subtitle_items)
        # A user who explicitly writes “篮球” or “足球” has provided a more
        # reliable sport hint than an incomplete/quiet commentary transcript.
        # This ensures the correct full-video scoreboard scanner still runs
        # when Whisper has little useful sports vocabulary to vote on.
        requested_sport = ""
        request_lower = editing_request.lower()
        if "篮球" in request_lower or "basketball" in request_lower:
            requested_sport = "basketball"
        elif "足球" in request_lower or "football" in request_lower:
            requested_sport = "football"
        if requested_sport and (detected_sport == "generic" or detected_sport != requested_sport):
            detected_sport = requested_sport
            sport_confidence = max(float(sport_confidence), 0.9)
        sport_label = {"football": "足球", "basketball": "篮球"}.get(detected_sport, "通用体育")
        job["sport_type"] = detected_sport
        job["sport_note"] = f"自动识别：{sport_label}（字幕证据 {sport_confidence:.0%}）"
        job.update(stage=f"{sport_label}比分牌 OCR 与得分归属", progress=52)
        from scoreboard_ocr_agent import (
            analyze_scoreboard_evidence,
            discover_basketball_score_events,
            discover_football_score_events,
        )
        scanned_goal_candidates: list[dict[str, Any]] = []
        scanned_goal_evidence: dict[str, dict[str, Any]] = {}
        penalty_fallback_candidates: list[dict[str, Any]] = []
        penalty_fallback_evidence: dict[str, dict[str, Any]] = {}
        basketball_score_candidates: list[dict[str, Any]] = []
        basketball_score_evidence: dict[str, dict[str, Any]] = {}
        basketball_miss_candidates: list[dict[str, Any]] = []
        basketball_miss_evidence: dict[str, dict[str, Any]] = {}
        scan_note = ""
        penalty_note = ""
        basketball_score_note = ""
        basketball_note = ""
        job["candidate_audit"] = {"subtitle_audio_candidates": len(candidate_list)}
        if detected_sport == "football":
            job.update(stage="足球全片比分扫描（定位进球）", progress=52)
            scanned_goal_candidates, scanned_goal_evidence, scan_note = discover_football_score_events(
                source,
                device=settings.device,
                scan_interval_sec=settings.football_score_scan_interval_sec,
            )
            # A live score overlay often animates exactly when the first
            # penalty is converted.  When that makes the +1 transition
            # unreadable, retain the clearly-commentated, crowd-confirmed
            # penalty sequence as a separate goal candidate.
            penalty_fallback_candidates, penalty_fallback_evidence, penalty_note = _discover_penalty_goal_fallback(
                subtitle_items, parsed_data.get("audio_segments") or parsed_data.get("segments") or []
            )
            candidate_list.extend(scanned_goal_candidates)
            candidate_list.extend(penalty_fallback_candidates)
        elif detected_sport == "basketball":
            job.update(stage="篮球全片比分扫描（定位得分）", progress=52)
            basketball_score_candidates, basketball_score_evidence, basketball_score_note = discover_basketball_score_events(
                source,
                device=settings.device,
                scan_interval_sec=settings.basketball_score_scan_interval_sec,
                min_confirmations=settings.basketball_score_scan_confirmations,
                pre_roll_sec=settings.basketball_score_pre_roll_sec,
                post_roll_sec=settings.basketball_score_post_roll_sec,
            )
            _attach_basketball_score_context(
                basketball_score_candidates, basketball_score_evidence, subtitle_items
            )
            basketball_miss_candidates, basketball_miss_evidence, basketball_note = _discover_basketball_miss_fallback(
                subtitle_items
            )
            # Scoreboard candidates are independently created from the entire
            # match, so a quiet or badly transcribed made basket cannot vanish
            # merely because the audio/subtitle detector missed it.
            candidate_list.extend(basketball_score_candidates)
            candidate_list.extend(basketball_miss_candidates)
        job["candidate_audit"].update({
            "basketball_scoreboard_candidates": len(basketball_score_candidates),
            "basketball_scoreboard_verified": sum(
                1 for item in basketball_score_evidence.values() if item.get("score_verified")
            ),
            "basketball_scoreboard_supported": sum(
                1 for item in basketball_score_evidence.values() if item.get("score_supported")
            ),
            "football_scoreboard_candidates": len(scanned_goal_candidates),
            "before_local_scoreboard_check": len(candidate_list),
        })
        scoreboard_evidence, scoreboard_note = analyze_scoreboard_evidence(
            source, candidate_list, device=settings.device, sport_hint=detected_sport
        )
        # The continuous scan has a known before/after pair, so keep it as the
        # stronger evidence for its purpose-built score candidates.
        scoreboard_evidence.update(scanned_goal_evidence)
        scoreboard_evidence.update(penalty_fallback_evidence)
        scoreboard_evidence.update(basketball_score_evidence)
        scoreboard_evidence.update(basketball_miss_evidence)
        resolved_sport, resolved_confidence = enrich_candidate_events(
            candidate_list, subtitle_items, scoreboard_evidence, sport_hint=detected_sport
        )
        # A genuine before/after score change is stronger evidence than a loud
        # crowd.  Re-rank after OCR rather than letting audio peaks decide the
        # clips that survive the duration budget.
        candidate_list = _reweight_candidates_with_scoreboard(candidate_list, scoreboard_evidence)
        sport_label = {"football": "足球", "basketball": "篮球"}.get(resolved_sport, "通用体育")
        job["sport_type"] = resolved_sport
        job["sport_note"] = f"自动识别：{sport_label}（字幕证据 {resolved_confidence:.0%}）"

        # Wide candidate windows protect recall during detection, but they are
        # not suitable for the final reel.  A dedicated pacing pass centres
        # every verified play on its commentary / OCR / crowd anchor and
        # collapses overlapping duplicates before team filtering and planning.
        job.update(stage="节奏剪辑 Agent：进球对齐与去重", progress=56)
        from pacing_editor_agent import PacingEditorAgent

        candidate_list, pacing_note = PacingEditorAgent().process(
            candidates=candidate_list,
            subtitles=subtitle_items,
            audio_segments=parsed_data.get("audio_segments") or parsed_data.get("segments") or [],
            event_evidence=scoreboard_evidence,
            video_duration_sec=float(parsed_data.get("total_duration_sec") or 0.0),
        )
        job["pacing_note"] = pacing_note
        job["candidate_audit"]["after_pacing_dedup"] = len(candidate_list)

        # The available jersey-colour fallback was trained only for the Lakers
        # and Celtics.  For football we rely on the more defensible scoreboard
        # result instead of assigning a team from arbitrary kit colours.
        if resolved_sport == "basketball":
            job.update(stage="视觉队伍识别（球衣 + 持球人）", progress=58)
            from team_visual_agent import analyze_team_evidence
            jersey_evidence, jersey_note = analyze_team_evidence(
                source, candidate_list, device=settings.device
            )
        else:
            job.update(stage="足球事件识别（进球 / 射门 / 判罚）", progress=58)
            jersey_evidence = {}
            jersey_note = "足球队伍归属优先由比分牌变化确认；无比分变化的镜头不臆测球队。"
        team_evidence = {}
        for candidate in candidate_list:
            candidate_id = str(candidate["candidate_id"])
            score = scoreboard_evidence.get(candidate_id, {})
            jersey = jersey_evidence.get(candidate_id, {})
            team_evidence[candidate_id] = score if score.get("team") not in {None, "待确认"} else jersey
        job["team_evidence"] = team_evidence
        job["visual_note"] = f"{job['sport_note']}。{pacing_note} {scan_note} {penalty_note} {basketball_score_note} {basketball_note} {scoreboard_note} {jersey_note}"
        candidate_list, selection_note = _apply_edit_request(
            candidate_list, editing_request, api_key, target_seconds, team_evidence
        )
        job["candidate_audit"]["after_edit_instruction"] = len(candidate_list)
        # Agent 1: turn the evidence-backed event list into an explainable
        # creative plan.  The optional model may polish copy only; it never
        # changes event selection or invents footage.
        job.update(stage="AI 体育导演：生成故事板", progress=64)
        from director_agent import DirectorAgent

        creative_provider = None
        if api_key:
            def creative_provider(prompt: str) -> str:
                return _chat_completion(
                    api_key,
                    [
                        {
                            "role": "system",
                            "content": "你是严谨的体育短视频文案助手。只能润色给定证据，绝不编造球员、比分、队伍或镜头。",
                        },
                        {"role": "user", "content": prompt},
                    ],
                    temperature=0.35,
                )

        director_plan = DirectorAgent().process(
            sport=resolved_sport,
            request=editing_request,
            candidates=candidate_list,
            team_evidence=team_evidence,
            target_seconds=target_seconds,
            creative_provider=creative_provider,
        )
        job["director_plan"] = director_plan
        director_path = source.parent / "director_storyboard.json"
        director_path.write_text(
            json.dumps(director_plan, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        job["director_artifact"] = _relative(director_path)
        candidates["candidates"] = candidate_list
        with open(candidates_path, "w", encoding="utf-8") as f:
            json.dump(candidates, f, ensure_ascii=False, indent=2)
        if selection_note:
            job["editing_note"] = selection_note
        job["candidates"] = candidate_list
        job["chat_context"] = (
            f"本次视频项目：{sport_label}。已生成 {len(candidate_list)} 个候选高光。用户剪辑要求：{editing_request or '无'}。"
            f"字幕摘录：{subtitle_context or '未识别到可用字幕'}\n"
            f"视觉队伍识别：{json.dumps(team_evidence, ensure_ascii=False)}\n"
            f"候选信息：{json.dumps(candidate_list, ensure_ascii=False)[:6000]}"
        )
        job["chat_context"] += (
            "\n导演方案："
            + json.dumps(
                {
                    "title": director_plan.get("title"),
                    "summary": director_plan.get("summary"),
                    "storyboard": director_plan.get("storyboard", []),
                },
                ensure_ascii=False,
            )[:3500]
        )

        job.update(stage="语义精剪与镜头编排", progress=68)
        from schema import SportType
        sport_enum = {
            "football": SportType.FOOTBALL,
            "basketball": SportType.BASKETBALL,
        }.get(resolved_sport, SportType.GENERIC)
        script_path = EditPlannerAgent().process(
            candidates_path,
            target_duration_sec=target_seconds,
            sport_type=sport_enum,
            output_filename=f"{job_id}_highlight_montage.mp4",
            script_filename=f"{job_id}_edit_script.json",
        )
        with open(script_path, encoding="utf-8") as f:
            job["edit_script"] = json.load(f)

        job.update(stage="字幕烧录与成片渲染", progress=86)
        video_path, report_path = RenderManagerAgent().process(
            script_path,
            subtitle_srt_path=subtitle_srt_path or None,
            report_filename=f"{job_id}_render_report.json",
        )
        with open(report_path, encoding="utf-8") as f:
            job["report"] = json.load(f)
        # Agent 2: inspect the actual render against the edit plan before it
        # is handed to the user.  It reports actionable warnings rather than
        # silently hiding an imperfect export.
        job.update(stage="审片质检 Agent：验证成片", progress=96)
        from quality_review_agent import QualityReviewAgent

        quality_review = QualityReviewAgent().process(
            target_seconds=target_seconds,
            edit_script=job["edit_script"],
            render_report=job["report"],
            director_plan=director_plan,
            subtitle_path=subtitle_srt_path or None,
        )
        job["quality_review"] = quality_review
        quality_path = source.parent / "quality_review.json"
        quality_path.write_text(
            json.dumps(quality_review, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        job["quality_artifact"] = _relative(quality_path)
        job.update(status="done", stage="成片已生成", progress=100, video=_relative(video_path))
    except Exception as exc:
        job.update(status="error", stage="处理失败", error=f"{type(exc).__name__}: {exc}")


@app.get("/")
def home() -> FileResponse:
    return FileResponse(ROOT / "static" / "index.html")


@app.post("/api/jobs")
async def create_job(
    video: UploadFile = File(...),
    target_seconds: float = Form(60),
    api_key: str = Form(""),
    editing_request: str = Form(""),
) -> dict[str, str]:
    if Path(video.filename or "").suffix.lower() not in {".mp4", ".mkv", ".mov", ".avi"}:
        raise HTTPException(400, "请选择 MP4、MKV、MOV 或 AVI 视频文件。")
    job_id = uuid.uuid4().hex[:10]
    job_dir = RUNS / job_id
    job_dir.mkdir()
    source = job_dir / (video.filename or "input.mp4")
    with source.open("wb") as f:
        shutil.copyfileobj(video.file, f)
    jobs[job_id] = {
        "id": job_id,
        "status": "queued",
        "stage": "等待开始",
        "progress": 0,
        "candidates": [],
        "editing_request": editing_request,
        "director_plan": None,
        "quality_review": None,
    }
    threading.Thread(target=_run_job, args=(job_id, source, target_seconds, api_key or None, editing_request), daemon=True).start()
    return {"job_id": job_id}


@app.get("/api/jobs/{job_id}")
def job_status(job_id: str) -> dict[str, Any]:
    if job_id not in jobs:
        raise HTTPException(404, "任务不存在")
    return jobs[job_id]


class ChatRequest(BaseModel):
    api_key: str = Field(min_length=1)
    messages: list[dict[str, str]]
    job_id: str | None = None


@app.post("/api/chat")
def chat(request: ChatRequest) -> dict[str, str]:
    context = "尚未上传视频。你可以先帮助用户梳理剪辑需求。"
    if request.job_id and request.job_id in jobs:
        context = jobs[request.job_id].get("chat_context", context)
    system = (
        "你是 AutoCut Studio 的体育视频 AI 助理。请用中文简洁回答。"
        "你可以解释候选高光、给出剪辑建议，或基于已提供的字幕和候选评价球员打法；"
        "没有视频证据时必须明确说明不能臆测。\n当前任务上下文：" + context
    )
    safe_messages = [{"role": "system", "content": system}] + request.messages[-12:]
    try:
        return {"answer": _chat_completion(request.api_key, safe_messages)}
    except requests.HTTPError as exc:
        raise HTTPException(502, "大模型服务拒绝了请求，请检查 Token Plan Key。") from exc
    except Exception as exc:
        raise HTTPException(502, f"大模型暂时不可用：{type(exc).__name__}") from exc


@app.get("/runs/{path:path}")
def run_file(path: str) -> FileResponse:
    file = (RUNS / path).resolve()
    if RUNS.resolve() not in file.parents or not file.is_file():
        raise HTTPException(404, "文件不存在")
    return FileResponse(file)


@app.get("/output/{path:path}")
def output_file(path: str) -> FileResponse:
    """Expose only generated deliverables, never arbitrary local files."""
    file = (ROOT / "output" / path).resolve()
    output_root = (ROOT / "output").resolve()
    if output_root not in file.parents or not file.is_file():
        raise HTTPException(404, "文件不存在")
    return FileResponse(file)