content-find-agent-new/content_agent/run_service.py

from __future__ import annotations

import os
import socket
import threading
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4

from content_agent.constants import RUNTIME_SCHEMA_VERSION
from content_agent.business_modules import run_record
from content_agent.errors import ContentAgentError, ErrorCode, error_from_exception
from content_agent.graph import RunDependencies, build_run_graph
from content_agent.integrations.composite_runtime import CompositeRuntimeStore
from content_agent.integrations.database_runtime import ContentSupplyDbConfig, DatabaseRuntimeStore
from content_agent.integrations.demand_source import DemandSourceService
from content_agent.integrations.douyin import CrawapiDouyinClient
from content_agent.integrations.gemini_video import GeminiVideoClient as RealGeminiVideoClient
from content_agent.integrations.qwen_video import QwenVideoClient
from content_agent.integrations.kuaishou import CrawapiKuaishouClient
from content_agent.integrations.mock_platform import MockPlatformClient
from content_agent.integrations import oss_archive
from content_agent.integrations.shipinhao import CrawapiShipinhaoClient
from content_agent.integrations.policy_json import JsonPolicyBundleStore
from content_agent.integrations.query_variant import (
    MissingQueryVariantClient,
    query_variant_client_from_env,
)
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from content_agent.interfaces import (
    GeminiVideoClient,
    PlatformSearchClient,
    PolicyBundleStore,
    QueryVariantClient,
    RuntimeStore,
)
from content_agent.models import RunState
from content_agent.schemas import RunStartRequest


class RunService:
    def __init__(
        self,
        runtime_root: Path | str = Path("runtime/v1"),
        runtime: RuntimeStore | None = None,
        policy_store: PolicyBundleStore | None = None,
        demand_source: DemandSourceService | None = None,
        query_variant_client: QueryVariantClient | None = None,
        gemini_video_client: GeminiVideoClient | None = None,
        pattern_pg: Any | None = None,
    ) -> None:
        self.runtime = runtime or LocalRuntimeFileStore(runtime_root)
        self.policy_store = policy_store or JsonPolicyBundleStore(Path("."))
        self.demand_source = demand_source
        # M9C Gate 1:None → 真实 run 时惰性 from_env;测试可注入 fake。
        self.pattern_pg = pattern_pg
        self.query_variant_client = query_variant_client or MissingQueryVariantClient(
            "query variant client is not configured"
        )
        self._gemini_video_client = gemini_video_client or _DeterministicGeminiVideoClient()

    @classmethod
    def from_env(cls, runtime_root: Path | str = Path("runtime/v1")) -> "RunService":
        local_runtime = LocalRuntimeFileStore(runtime_root)
        env = _merged_project_env()
        query_variant_client = query_variant_client_from_env(env)
        gemini_video_client = _gemini_video_client_from_env(env)
        db_runtime_enabled = _env_enabled("CONTENT_AGENT_DB_RUNTIME_ENABLED")
        try:
            config = ContentSupplyDbConfig.from_env()
        except Exception as exc:
            if not db_runtime_enabled:
                return cls(
                    runtime_root=runtime_root,
                    runtime=local_runtime,
                    query_variant_client=query_variant_client,
                    gemini_video_client=gemini_video_client,
                )
            raise ContentAgentError(
                ErrorCode.DB_CONFIG_MISSING,
                "content supply db config is missing",
                {"exception_type": type(exc).__name__},
            ) from exc

        demand_source = DemandSourceService(config)
        if not db_runtime_enabled:
            return cls(
                runtime_root=runtime_root,
                runtime=local_runtime,
                demand_source=demand_source,
                query_variant_client=query_variant_client,
                gemini_video_client=gemini_video_client,
            )

        db_runtime = DatabaseRuntimeStore(config)
        return cls(
            runtime_root=runtime_root,
            runtime=CompositeRuntimeStore(db_runtime, local_runtime),
            demand_source=demand_source,
            query_variant_client=query_variant_client,
            gemini_video_client=gemini_video_client,
        )

    def start_run(self, request: RunStartRequest) -> RunState:
        run_id = request.run_id or f"v1_run_{uuid4().hex[:12]}"
        policy_run_id = f"policy_run_{run_id.removeprefix('v1_run_')}"
        source_ref = self._source_ref_from_request(request)
        initial_state: RunState = {
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "schema_version": RUNTIME_SCHEMA_VERSION,
            "platform": request.platform,
            "platform_mode": request.platform_mode,
            "source": request.source,
            "strategy_version": request.strategy_version,
            "current_step": "start",
            "status": "running",
            "errors": [],
        }

        try:
            self.runtime.prepare_run(run_id)
            self._create_run_record(run_id, request, source_ref)
            source = self._resolve_source(request)
            resolved_source_ref = self._source_ref_from_resolved_source(source, source_ref)
            if resolved_source_ref != source_ref:
                self.runtime.update_run_record(
                    run_id,
                    {
                        "demand_content_id": _demand_content_id_from_source(source),
                        "source_ref": resolved_source_ref,
                    },
                )
            initial_state["source"] = source
            self._append_lifecycle_event(
                run_id,
                policy_run_id,
                event_id="lifecycle_start",
                event_type="run_started",
                status="running",
                message="run started",
                raw_payload={"source_ref": resolved_source_ref, "executor": _executor_identity()},
            )

            # 先验证/构造平台 client(拒绝非法平台),再过 Gate 1(不为非法平台浪费 PG 调用)。
            platform_client = self._platform_client(request.platform, request.platform_mode)

            gate1_reason = self._gate1_block_reason(request, source)
            if gate1_reason is not None:
                return self._gate1_blocked_state(initial_state, gate1_reason)

            deps = RunDependencies(
                runtime=self.runtime,
                platform_client=platform_client,
                policy_store=self.policy_store,
                query_variant_client=self.query_variant_client,
                gemini_video_client=self._gemini_video_client,
            )
            graph = build_run_graph(deps)
            state = graph.invoke(initial_state)
            self._record_success_metadata(state)
            self._trigger_post_run_oss_archive(state, request)
            return state
        except Exception as exc:
            error = self._classify_error(exc)
            failed_state = self._failed_state(initial_state, error)
            self._record_failure_metadata(
                failed_state,
                request,
                source_ref,
                error,
            )
            return failed_state

    def _source_ref_from_request(self, request: RunStartRequest) -> dict[str, Any]:
        if request.demand_content_id is not None:
            return {
                "source_type": "demand_content",
                "demand_content_id": request.demand_content_id,
            }
        if request.run_label:
            return {
                "source_type": "demand_content",
                "run_label": request.run_label,
            }
        if request.source:
            return {"source_type": "local_source", "source": request.source}
        return {
            "source_type": "demand_content_default",
            "selector": "pg_pattern_v2_passed_first",
        }

    def _resolve_source(self, request: RunStartRequest) -> str | dict[str, Any] | None:
        if request.demand_content_id is not None:
            if not self.demand_source:
                raise ContentAgentError(
                    ErrorCode.DB_CONFIG_MISSING,
                    "demand source db is not configured",
                    {"selector": "demand_content_id"},
                )
            source = self.demand_source.get_by_id(request.demand_content_id)
            return source
        if request.run_label:
            if not self.demand_source:
                raise ContentAgentError(
                    ErrorCode.DB_CONFIG_MISSING,
                    "demand source db is not configured",
                    {"selector": "run_label"},
                )
            source = self.demand_source.get_by_run_label(request.run_label)
            return source
        if request.source:
            return request.source
        if not self.demand_source:
            raise ContentAgentError(
                ErrorCode.DB_CONFIG_MISSING,
                "demand source db is not configured",
                {"selector": "default_pg_pattern_v2_passed"},
            )
        return self.demand_source.get_default_pg_pattern_source()

    def _source_ref_from_resolved_source(
        self,
        source: str | dict[str, Any] | None,
        source_ref: dict[str, Any],
    ) -> dict[str, Any]:
        if source_ref.get("source_type") != "demand_content_default" or not isinstance(
            source,
            dict,
        ):
            return source_ref
        demand_content_id = _demand_content_id_from_source(source)
        resolved = dict(source_ref)
        if demand_content_id is not None:
            resolved["demand_content_id"] = demand_content_id
        return resolved

    def _create_run_record(
        self,
        run_id: str,
        request: RunStartRequest,
        source_ref: dict[str, Any],
    ) -> None:
        self.runtime.create_run_record(
            {
                "run_id": run_id,
                "demand_content_id": request.demand_content_id,
                "run_label": request.run_label,
                "platform": request.platform,
                "platform_mode": request.platform_mode,
                "strategy_version": request.strategy_version,
                "status": "running",
                "current_step": "start",
                "source_ref": source_ref,
                "started_at": _utc_now(),
            }
        )

    def _record_success_metadata(self, state: RunState) -> None:
        final_status = "partial_success" if state.get("query_failures") else "success"
        state["status"] = final_status
        self.runtime.record_policy_run(_policy_run_record_from_state(state))
        validation = self.validate_run(state["run_id"])
        self._update_final_output_validation(state["run_id"], validation)
        self.runtime.update_run_record(
            state["run_id"],
            {
                "status": final_status,
                "current_step": state.get("current_step", "review_strategy"),
                "validation_status": validation["status"],
                "completed_at": _utc_now(),
            },
        )
        self._append_lifecycle_event(
            state["run_id"],
            state["policy_run_id"],
            event_id="lifecycle_success",
            event_type="run_succeeded",
            status=final_status,
            message="run succeeded"
            if final_status == "success"
            else "run partially succeeded",
            output_ref="final_output.json",
            raw_payload={
                "validation_status": validation["status"],
                "policy_bundle_id": state.get("policy_bundle_id"),
                "query_failures": state.get("query_failures", []),
            },
        )

    def _trigger_post_run_oss_archive(self, state: RunState, request: RunStartRequest) -> None:
        if request.platform_mode != "real":
            return
        run_id = state["run_id"]
        policy_run_id = state["policy_run_id"]
        try:
            records = self.runtime.read_jsonl(run_id, "content_media_records.jsonl")
            pending_count = sum(
                1
                for row in records
                if row.get("content_media_status") == "oss_upload_pending" and row.get("play_url")
            )
            if pending_count <= 0:
                return
            self._append_lifecycle_event(
                run_id,
                policy_run_id,
                event_id="oss_archive_post_run_started",
                event_type="oss_archive_post_run",
                status="running",
                message="post-run oss archive started",
                raw_payload={"pending_due_count": pending_count},
            )
        except Exception:
            return

        def _archive() -> None:
            try:
                archived = oss_archive.archive_pending_for_run(self.runtime, run_id)
                status_counts = Counter(row.get("content_media_status") for row in archived)
                self._append_lifecycle_event(
                    run_id,
                    policy_run_id,
                    event_id="oss_archive_post_run_completed",
                    event_type="oss_archive_post_run",
                    status="success",
                    message="post-run oss archive completed",
                    raw_payload={
                        "pending_due_count": pending_count,
                        "content_media_status_counts": dict(status_counts),
                    },
                )
            except Exception as exc:  # noqa: BLE001
                self._append_lifecycle_event(
                    run_id,
                    policy_run_id,
                    event_id="oss_archive_post_run_failed",
                    event_type="oss_archive_post_run",
                    status="failed",
                    message="post-run oss archive failed",
                    error_code="OSS_ARCHIVE_POST_RUN_FAILED",
                    raw_payload={
                        "pending_due_count": pending_count,
                        "exception_type": type(exc).__name__,
                        "error": str(exc)[:500],
                    },
                )

        self._start_background_thread(_archive, name=f"oss-archive-{run_id}")

    def _start_background_thread(self, target: Any, *, name: str) -> None:
        thread = threading.Thread(target=target, name=name, daemon=True)
        thread.start()

    def _update_final_output_validation(self, run_id: str, validation: dict[str, Any]) -> None:
        final_output = self.runtime.read_json(run_id, "final_output.json")
        validation_status = validation["status"]
        findings = validation.get("findings", [])
        final_output["validation_status"] = validation_status
        summary = final_output.setdefault("summary", {})
        summary["run_path_complete"] = validation_status == "pass"
        summary["trace_complete"] = validation_status == "pass"
        summary["validation_findings_summary"] = [
            finding.get("message", str(finding)) if isinstance(finding, dict) else str(finding)
            for finding in findings
        ]
        self.runtime.update_json(run_id, "final_output.json", final_output)

    def _record_failure_metadata(
        self,
        state: RunState,
        request: RunStartRequest,
        source_ref: dict[str, Any],
        error: ContentAgentError,
    ) -> None:
        run_id = state["run_id"]
        policy_run_id = state["policy_run_id"]
        try:
            self.runtime.update_run_record(
                run_id,
                {
                    "status": "failed",
                    "current_step": state.get("current_step", "failed"),
                    "error_code": error.error_code.value,
                    "error_message": error.message,
                    "error_detail": error.detail,
                    "completed_at": _utc_now(),
                },
            )
            self._record_platform_query_failure_details(run_id, policy_run_id, error)
            self._append_lifecycle_event(
                run_id,
                policy_run_id,
                event_id="lifecycle_failed",
                event_type="run_failed",
                status="failed",
                message=error.message,
                error_code=error.error_code.value,
                raw_payload={
                    "error_detail": error.detail,
                    "source_ref": source_ref,
                    "platform_mode": request.platform_mode,
                },
            )
        except Exception:
            # Preserve the original run failure; DB failure here is already reflected by the state.
            return

    def _record_platform_query_failure_details(
        self,
        run_id: str,
        policy_run_id: str,
        error: ContentAgentError,
    ) -> None:
        query_failures = _query_failures_from_error(error)
        if not query_failures:
            return
        try:
            search_queries = self.runtime.read_jsonl(run_id, "search_queries.jsonl")
        except FileNotFoundError:
            return
        if not search_queries:
            return
        records = run_record.build_platform_query_failure_records(
            run_id,
            policy_run_id,
            search_queries,
            query_failures,
        )
        self.runtime.append_jsonl(run_id, "search_queries.jsonl", records["search_queries"])
        self.runtime.append_jsonl(run_id, "search_clues.jsonl", records["search_clues"])
        self.runtime.append_jsonl(run_id, "run_events.jsonl", records["run_events"])

    def _append_lifecycle_event(
        self,
        run_id: str,
        policy_run_id: str,
        *,
        event_id: str,
        event_type: str,
        status: str,
        message: str,
        input_ref: str | None = None,
        output_ref: str | None = None,
        error_code: str | None = None,
        raw_payload: dict[str, Any] | None = None,
    ) -> None:
        self.runtime.append_run_event_records(
            run_id,
            policy_run_id,
            [
                {
                    "event_id": event_id,
                    "event_type": event_type,
                    "status": status,
                    "input_ref": input_ref,
                    "output_ref": output_ref,
                    "error_code": error_code,
                    "message": message,
                    "raw_payload": raw_payload or {},
                    "created_at": _utc_now(),
                }
            ],
        )

    def _gate1_block_reason(self, request: RunStartRequest, source: Any) -> str | None:
        """M9C Gate 1:仅非抖音真实 run;需求 itemset_items 至少含一个分类树终端元素才做。

        无法判定(非 dict 源 / 缺 execution_id / 缺 category_id)→ 不阻断(交给后续)。
        PG 不可达 → has_terminal_element 抛 ContentAgentError(被 start_run except 归类),不静默放过。
        """
        if request.platform == "douyin" or request.platform_mode != "real":
            return None
        if not isinstance(source, dict):
            return None
        evidence_pack = (source.get("ext_data") or {}).get("evidence_pack") or {}
        execution_id = evidence_pack.get("pattern_execution_id")
        category_ids = [
            item.get("category_id")
            for item in evidence_pack.get("itemset_items", [])
            if isinstance(item, dict) and item.get("category_id") is not None
        ]
        if execution_id is None or not category_ids:
            return None
        client = self.pattern_pg
        if client is None:
            from content_agent.integrations.pattern_pg import PatternPgClient

            client = PatternPgClient.from_env()
        if client.has_terminal_element(execution_id, category_ids):
            return None
        return "gate1_no_terminal_element"

    def _gate1_blocked_state(self, initial_state: RunState, reason: str) -> RunState:
        run_id = initial_state["run_id"]
        policy_run_id = initial_state["policy_run_id"]
        blocked: RunState = {
            **initial_state,
            "current_step": "blocked_gate1",
            "status": "blocked",
            "error_code": reason,
            "error_message": "需求未来自分类树终端元素,非抖音 Gate 1 不放行",
            "errors": [],
        }
        try:
            self.runtime.update_run_record(
                run_id,
                {
                    "status": "blocked",
                    "current_step": "blocked_gate1",
                    "error_code": reason,
                    "completed_at": _utc_now(),
                },
            )
            self._append_lifecycle_event(
                run_id,
                policy_run_id,
                event_id="lifecycle_blocked_gate1",
                event_type="run_blocked",
                status="blocked",
                message=blocked["error_message"],
                error_code=reason,
            )
        except Exception:
            pass
        return blocked

    def _failed_state(self, initial_state: RunState, error: ContentAgentError) -> RunState:
        return {
            **initial_state,
            "current_step": "failed",
            "status": "failed",
            "error_code": error.error_code.value,
            "error_message": error.message,
            "error_detail": error.detail,
            "http_status_code": error.status_code,
            "errors": [error.message],
        }

    def _classify_error(self, exc: Exception) -> ContentAgentError:
        if isinstance(exc, ContentAgentError):
            return exc
        if isinstance(exc, FileNotFoundError):
            return ContentAgentError(
                ErrorCode.INVALID_SOURCE,
                "source file not found",
                {"exception_type": type(exc).__name__},
                status_code=400,
            )
        if isinstance(exc, ValueError) and "unknown strategy_version" in str(exc):
            return ContentAgentError(
                ErrorCode.POLICY_BUNDLE_NOT_FOUND,
                "policy bundle not found",
                {"exception_type": type(exc).__name__},
                status_code=400,
            )
        if isinstance(exc, ValueError) and "evidence_pack" in str(exc):
            return ContentAgentError(
                ErrorCode.INVALID_SOURCE,
                "invalid source",
                {"exception_type": type(exc).__name__},
                status_code=400,
            )
        return error_from_exception(exc, detail={"exception_type": type(exc).__name__})

    def _platform_client(self, platform: str, platform_mode: str) -> PlatformSearchClient:
        if platform_mode == "mock":
            return MockPlatformClient()
        if platform_mode == "real":
            real_clients = {
                "douyin": CrawapiDouyinClient.from_env,
                "kuaishou": CrawapiKuaishouClient.from_env,
                "shipinhao": CrawapiShipinhaoClient.from_env,
            }
            builder = real_clients.get(platform)
            if builder is None:
                raise ContentAgentError(
                    ErrorCode.INVALID_REQUEST,
                    "unsupported real platform",
                    {"platform": platform},
                    status_code=400,
                )
            try:
                return builder()
            except Exception as exc:
                raise ContentAgentError(
                    ErrorCode.PLATFORM_CONFIG_MISSING,
                    "platform config missing",
                    {"exception_type": type(exc).__name__},
                ) from exc
        raise ContentAgentError(
            ErrorCode.INVALID_REQUEST,
            "unsupported platform_mode",
            {"platform_mode": platform_mode},
            status_code=400,
        )

    def get_summary(self, run_id: str) -> dict:
        final_output_exists = (self.runtime.run_dir(run_id) / "final_output.json").exists()
        status = self._summary_status(run_id, final_output_exists)
        validation = self.validate_run(run_id) if final_output_exists else {"status": "fail"}
        policy_run_id = None
        if final_output_exists:
            policy_run_id = self.runtime.read_json(run_id, "final_output.json").get("policy_run_id")
        return {
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "status": status,
            "current_step": "review_strategy" if final_output_exists else "unknown",
            "output_dir": str(self.runtime.run_dir(run_id)),
            "files": self.runtime.file_status(run_id),
            "validation_status": validation["status"],
            "errors": [],
        }

    def _summary_status(self, run_id: str, final_output_exists: bool) -> str:
        try:
            run_events = self.runtime.read_jsonl(run_id, "run_events.jsonl")
            lifecycle_events = [
                event for event in run_events if str(event.get("event_id", "")).startswith("lifecycle_")
            ]
        except Exception:
            run_events = []
            lifecycle_events = []
        if lifecycle_events:
            return lifecycle_events[-1].get("status") or (
                "success" if final_output_exists else "failed"
            )
        if final_output_exists and any(
            event.get("event_type") == "platform_query_failed" for event in run_events
        ):
            return "partial_success"
        return "success" if final_output_exists else "failed"

    def read_jsonl(self, run_id: str, filename: str) -> list[dict]:
        return self.runtime.read_jsonl(run_id, filename)

    def read_json(self, run_id: str, filename: str) -> dict:
        return self.runtime.read_json(run_id, filename)

    def strategy_review(self, run_id: str) -> dict:
        try:
            return self.runtime.read_json(run_id, "strategy_review.json")
        except FileNotFoundError:
            policy_run_id = None
            try:
                policy_run_id = self.runtime.read_json(run_id, "final_output.json").get(
                    "policy_run_id"
                )
            except FileNotFoundError:
                pass
            return {
                "schema_version": RUNTIME_SCHEMA_VERSION,
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "review_status": "not_generated",
            }

    def regenerate_strategy_review(self, run_id: str) -> dict:
        from content_agent.business_modules.learning_review import run

        final_output = self.runtime.read_json(run_id, "final_output.json")
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
        review_id = f"review_{final_output['policy_run_id']}_{timestamp}"
        return run(run_id, final_output["policy_run_id"], self.runtime, review_id=review_id)

    def validate_run(self, run_id: str) -> dict:
        from content_agent.business_modules.run_record import validate_run

        return validate_run(run_id, self.runtime)


def _policy_run_record_from_state(state: RunState) -> dict[str, Any]:
    policy_bundle = state.get("policy_bundle") or {}
    decisions = state.get("rule_decisions") or []
    return {
        "run_id": state["run_id"],
        "policy_run_id": state["policy_run_id"],
        "run_role": "primary",
        "policy_bundle_id": state.get("policy_bundle_id"),
        "rule_pack_id": policy_bundle.get("rule_pack_id"),
        "strategy_id": policy_bundle.get("strategy_id"),
        "strategy_version": state.get("strategy_version"),
        "rule_pack_version": policy_bundle.get("rule_pack_version"),
        "walk_strategy_version": policy_bundle.get("walk_strategy_version"),
        "policy_bundle_hash": policy_bundle.get("policy_bundle_hash"),
        "strategy_source_ref": policy_bundle.get("strategy_source_ref"),
        "rule_pack_source_ref": policy_bundle.get("rule_pack_source_ref"),
        "evidence_bundle_schema_version": policy_bundle.get("evidence_bundle_schema_version"),
        "runtime_record_schema_version": policy_bundle.get("runtime_record_schema_version"),
        "status": state.get("status", "success"),
        "metrics": (state.get("final_output") or {}).get("summary", {}),
        "decision_summary": _decision_summary(decisions),
        "raw_payload": {
            "current_step": state.get("current_step"),
            "search_query_count": len(state.get("search_queries") or []),
            "discovered_content_count": len(state.get("discovered_content_items") or []),
            "decision_count": len(decisions),
            "final_output_ref": "final_output.json",
            "dispatch": policy_bundle.get("dispatch"),
            "runtime_status_contract": policy_bundle.get("runtime_status_contract", {}),
            "effect_status_mapping": policy_bundle.get("effect_status_mapping", []),
            "query_effect_aggregation": policy_bundle.get("query_effect_aggregation", []),
            "decision_reason_codes": policy_bundle.get("decision_reason_codes", []),
            "policy_bundle_hash": policy_bundle.get("policy_bundle_hash"),
        },
    }


def _decision_summary(decisions: list[dict[str, Any]]) -> dict[str, Any]:
    return {
        "decision_action_counts": dict(Counter(item.get("decision_action") for item in decisions)),
        "decision_reason_code_counts": dict(
            Counter(item.get("decision_reason_code") for item in decisions)
        ),
        "effect_status_counts": dict(
            Counter(item.get("search_query_effect_status") for item in decisions)
        ),
    }


def _query_failures_from_error(error: ContentAgentError) -> list[dict[str, Any]]:
    query_failures = error.detail.get("query_failures") if isinstance(error.detail, dict) else None
    if not isinstance(query_failures, list):
        return []
    return [failure for failure in query_failures if isinstance(failure, dict)]


def _gemini_video_client_from_env(env: dict[str, str]) -> GeminiVideoClient:
    # 视频理解 provider 开关:dashscope=通义千问(国内,video_url 直取 oss);否则 OpenRouter/Gemini(海外)。
    provider = (env.get("CONTENT_AGENT_VIDEO_LLM_PROVIDER") or "openrouter").strip().lower()
    if provider in {"dashscope", "qwen", "bailian", "tongyi"}:
        return QwenVideoClient.from_env(env)
    return RealGeminiVideoClient.from_env(env)


class _DeterministicGeminiVideoClient:
    """mock/默认判定 client:固定返回 V4 高相关结果,供本地/smoke 无网跑通。"""

    def analyze(
        self,
        content: dict[str, Any],
        media: dict[str, Any],
        source_context: dict[str, Any],
    ) -> dict[str, Any]:
        return {
            "schema_version": "v4_gemini_query_relevance.v1",
            "query_text": "deterministic query",
            "query_relevance_score": 80,
            "query_relevance_reason": "deterministic stub relevance",
            "final_status": "ok",
            "retry_count": 0,
        }


def _env_enabled(key: str) -> bool:
    value = _load_project_env().get(key)
    if os.environ.get(key) is not None:
        value = os.environ[key]
    return (value or "").lower() in {"1", "true", "yes", "on"}


def _merged_project_env() -> dict[str, str]:
    return {
        **_load_project_env(),
        **dict(os.environ),
    }


def _demand_content_id_from_source(source: str | dict[str, Any] | None) -> int | None:
    if not isinstance(source, dict):
        return None
    value = source.get("id") or source.get("demand_content_id")
    if value in (None, ""):
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _proc_start_time(pid: int) -> str | None:
    # Linux /proc/<pid>/stat 第 22 字段 starttime(自启动以来的时钟滴答),作为 PID 复用的指纹;
    # comm(第 2 字段)在括号里且可能含空格,故按最后一个 ')' 切分。非 Linux/取不到 → None。
    try:
        with open(f"/proc/{pid}/stat", encoding="utf-8") as handle:
            content = handle.read()
        return content[content.rindex(")") + 2:].split()[19]
    except Exception:
        return None


def _executor_identity() -> dict[str, Any]:
    # 记录"谁在跑这条 run":pid + 主机 + 进程启动时刻。读状态时据此判进程是否还活着(无时间阈值)。
    pid = os.getpid()
    return {"pid": pid, "host": socket.gethostname(), "proc_start": _proc_start_time(pid)}


def _load_project_env(env_file: Path | str = ".env") -> dict[str, str]:
    path = Path(env_file)
    if not path.exists():
        return {}
    result: dict[str, str] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        key, value = stripped.split("=", 1)
        result[key.strip()] = value.strip().strip('"').strip("'")
    return result