from __future__ import annotations

import threading
import time

from content_agent.business_modules.content_discovery.pattern_recall import recall_decision
from content_agent.business_modules.content_discovery.pattern_recall import media_pipeline
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.gemini_helpers import FakeGeminiVideoClient, fake_gemini_fail, fake_gemini_pool, fake_gemini_review


def test_pattern_recall_marks_all_judged_videos_for_oss_archive(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    items = [
        _item(run_id, policy_run_id, "content_pool", "d_001"),
        _item(run_id, policy_run_id, "content_review", "d_002"),
        _item(run_id, policy_run_id, "content_failed", "d_003"),
    ]
    media = [_media(run_id, policy_run_id, item["platform_content_id"]) for item in items]
    bundles = [_bundle(item) for item in items]
    pool_result = fake_gemini_pool()
    pool_result["timing_metrics"] = {
        "video_fetch": {"download_duration_ms": 100, "ffmpeg_duration_ms": 200},
        "gemini_request": {"total_duration_ms": 300},
    }
    gemini = FakeGeminiVideoClient(
        result_by_content_id={
            "content_pool": pool_result,
            "content_review": fake_gemini_review(),
            "content_failed": fake_gemini_fail("gemini_client_timeout"),
        }
    )

    result = recall_decision.run(
        run_id,
        policy_run_id,
        items,
        media,
        bundles,
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
    )

    assert {
        row["platform_content_id"]: row["content_media_status"]
        for row in result["content_media_records"]
    } == {
        "content_pool": "oss_upload_pending",
        "content_review": "oss_upload_pending",
        "content_failed": "oss_upload_pending",
    }
    for row in result["content_media_records"]:
        assert row["raw_payload"]["oss_archive_status"] == "pending"
    media_by_id = {row["platform_content_id"]: row for row in result["content_media_records"]}
    assert media_by_id["content_pool"]["raw_payload"]["gemini_timing_metrics"]["video_fetch"] == {
        "download_duration_ms": 100,
        "ffmpeg_duration_ms": 200,
    }
    evidence_by_id = {row["platform_content_id"]: row for row in result["pattern_recall_evidence"]}
    assert evidence_by_id["content_pool"]["evidence_summary"]["timing_metrics"]["gemini_request"] == {
        "total_duration_ms": 300,
    }


def test_pattern_recall_submits_oss_archive_before_gemini_judgment(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = RecordingArchiveDispatcher()
    gemini = ArchiveAwareGeminiVideoClient(dispatcher)

    recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert dispatcher.submitted_content_ids == ["content_pool"]
    assert gemini.calls


def test_pattern_recall_waits_for_oss_archive_before_gemini_judgment(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = SynchronousArchiveDispatcher()
    gemini = OssAwareGeminiVideoClient()

    result = recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert dispatcher.archived_content_ids == ["content_pool"]
    assert gemini.calls[0]["media"]["oss_url"] == "https://res.example/content_pool.mp4"
    [row] = result["content_media_records"]
    assert row["content_media_status"] == "oss_uploaded"
    assert row["oss_url"] == "https://res.example/content_pool.mp4"
    assert row["raw_payload"]["gemini_timing_metrics"]["video_fetch"]["gemini_video_source"] == "oss_url"


def test_pattern_recall_pipelines_qwen_after_each_oss_completion(tmp_path, monkeypatch):
    monkeypatch.setattr(recall_decision, "_resolve_max_workers", lambda: 4)
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    items = [
        _item(run_id, policy_run_id, "slow", "d_001"),
        _item(run_id, policy_run_id, "fast", "d_002"),
    ]
    media = [_media(run_id, policy_run_id, item["platform_content_id"]) for item in items]
    bundles = [_bundle(item) for item in items]
    events: list[str] = []
    lock = threading.Lock()
    dispatcher = DelayedArchiveDispatcher(events, lock)
    gemini = EventGeminiVideoClient(events, lock)

    recall_decision.run(
        run_id,
        policy_run_id,
        items,
        media,
        bundles,
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert events.index("qwen:fast") < events.index("oss_end:slow")
    tasks = runtime.read_media_pipeline_tasks(run_id)
    assert {task["task_type"] for task in tasks} == {"oss_upload", "qwen_judgment"}
    assert len(tasks) == 4


def test_qwen_retry_wait_does_not_occupy_worker(tmp_path, monkeypatch):
    monkeypatch.setattr(recall_decision, "_resolve_max_workers", lambda: 1)
    monkeypatch.setattr(media_pipeline, "_base_qwen_retry_delay", lambda judgment, attempt_count: 0.03)
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    items = [
        _item(run_id, policy_run_id, "retry", "d_001"),
        _item(run_id, policy_run_id, "other", "d_002"),
    ]
    media = [_media(run_id, policy_run_id, item["platform_content_id"]) for item in items]
    bundles = [_bundle(item) for item in items]
    events: list[str] = []
    dispatcher = SynchronousArchiveDispatcher()
    qwen = RetryOnceQwenClient(events)

    result = recall_decision.run(
        run_id,
        policy_run_id,
        items,
        media,
        bundles,
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        qwen,
        archive_dispatcher=dispatcher,
    )

    assert events[:3] == ["qwen:retry:1", "qwen:other:1", "qwen:retry:2"]
    evidence = {row["platform_content_id"]: row for row in result["pattern_recall_evidence"]}
    assert evidence["retry"]["recall_status"] == "judged"
    tasks = [
        task
        for task in runtime.read_media_pipeline_tasks(run_id)
        if task["task_type"] == "qwen_judgment"
    ]
    retry_task = next(task for task in tasks if task["platform_content_id"] == "retry")
    assert retry_task["attempt_count"] == 2
    assert retry_task["next_retry_at"] is None
    assert retry_task["input_payload"]["qwen_retry_policy_version"] == media_pipeline.QWEN_RETRY_POLICY_VERSION


def test_qwen_rate_limit_lowers_dynamic_worker_limit():
    now = time.monotonic()
    assert media_pipeline._dynamic_qwen_limit(5, now + 30, now) == 2
    assert media_pipeline._dynamic_qwen_limit(5, now - 1, now) == 5


def test_pattern_recall_skips_qwen_when_oss_archive_is_pending(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = PendingArchiveDispatcher()
    gemini = PlayUrlAwareGeminiVideoClient()

    result = recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert gemini.calls == []
    [row] = result["content_media_records"]
    assert row["content_media_status"] == "oss_upload_pending"
    assert row["raw_payload"]["oss_archive_last_error"] == "oss_upload_http_error"
    [evidence] = result["pattern_recall_evidence"]
    assert evidence["recall_status"] == "failed"
    assert evidence["evidence_summary"]["failure_type"] == "oss_upload_http_error"


def test_pattern_recall_enqueues_qwen_after_oss_fallback_candidate_succeeds(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    media["raw_payload"]["video_url_candidates"] = [
        {
            "url": "https://source.example/content_pool.mp4",
            "host": "source.example",
            "path": "$.search.video_url_list[0].video_url",
            "source": "search",
            "candidate_index": 0,
        },
        {
            "url": "https://backup.example/content_pool.mp4",
            "host": "backup.example",
            "path": "$.detail.video_url_list[0].video_url",
            "source": "detail",
            "candidate_index": 1,
        },
    ]
    bundle = _bundle(item)
    dispatcher = FallbackUploadDispatcher()
    gemini = OssAwareGeminiVideoClient()

    result = recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert dispatcher.calls == [
        "https://source.example/content_pool.mp4",
        "https://backup.example/content_pool.mp4",
    ]
    assert gemini.calls
    [row] = result["content_media_records"]
    assert row["content_media_status"] == "oss_uploaded"
    assert row["raw_payload"]["oss_archive_selected_video_url_host"] == "backup.example"
    assert row["raw_payload"]["oss_url_attempts"][0]["failure_type"] == "oss_upload_http_error"
    tasks = runtime.read_media_pipeline_tasks(run_id)
    assert {task["task_type"] for task in tasks} == {"oss_upload", "qwen_judgment"}


class RecordingArchiveDispatcher:
    def __init__(self) -> None:
        self.submitted_content_ids: list[str] = []

    def archive_records(self, records: list[dict]) -> list[dict]:
        self.submitted_content_ids.extend(row["platform_content_id"] for row in records)
        archived = []
        for row in records:
            content_id = row["platform_content_id"]
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_uploaded",
                "oss_url": f"https://res.example/{content_id}.mp4",
                "oss_archive_status": "uploaded",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_uploaded",
                    "oss_url": f"https://res.example/{content_id}.mp4",
                    "raw_payload": raw_payload,
                }
            )
        return archived


class SynchronousArchiveDispatcher:
    def __init__(self) -> None:
        self.archived_content_ids: list[str] = []

    def archive_records(self, records: list[dict]) -> list[dict]:
        archived = []
        for row in records:
            content_id = row["platform_content_id"]
            self.archived_content_ids.append(content_id)
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_uploaded",
                "oss_url": f"https://res.example/{content_id}.mp4",
                "oss_archive_status": "uploaded",
                "oss_object_key": f"crawler/video/{content_id}.mp4",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_uploaded",
                    "oss_url": f"https://res.example/{content_id}.mp4",
                    "raw_payload": raw_payload,
                }
            )
        return archived


class PendingArchiveDispatcher:
    def archive_records(self, records: list[dict]) -> list[dict]:
        archived = []
        for row in records:
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_upload_pending",
                "oss_archive_status": "pending",
                "oss_archive_last_error": "oss_upload_http_error",
                "oss_archive_last_exception_type": "HTTPStatusError",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_upload_pending",
                    "oss_url": None,
                    "raw_payload": raw_payload,
                }
            )
        return archived


class FallbackUploadDispatcher:
    def __init__(self) -> None:
        self.calls: list[str] = []

    def upload_fn(self, src_url: str, **kwargs) -> dict:
        self.calls.append(src_url)
        if src_url.startswith("https://source.example/"):
            return {
                "status": "failed",
                "failure_type": "oss_upload_http_error",
                "exception_type": "ReadTimeout",
                "response_absent": True,
            }
        return {
            "status": "ok",
            "oss_url": "https://res.example/content_pool.mp4",
            "oss_object_key": "crawler/video/content_pool.mp4",
        }


class DelayedArchiveDispatcher:
    def __init__(self, events: list[str], lock: threading.Lock) -> None:
        self.events = events
        self.lock = lock

    def archive_records(self, records: list[dict]) -> list[dict]:
        archived = []
        for row in records:
            content_id = row["platform_content_id"]
            with self.lock:
                self.events.append(f"oss_start:{content_id}")
            if content_id == "slow":
                time.sleep(0.08)
            else:
                time.sleep(0.01)
            with self.lock:
                self.events.append(f"oss_end:{content_id}")
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_uploaded",
                "oss_url": f"https://res.example/{content_id}.mp4",
                "oss_archive_status": "uploaded",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_uploaded",
                    "oss_url": f"https://res.example/{content_id}.mp4",
                    "raw_payload": raw_payload,
                }
            )
        return archived


class ArchiveAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self, dispatcher: RecordingArchiveDispatcher) -> None:
        super().__init__(default_result=fake_gemini_pool())
        self.dispatcher = dispatcher

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert self.dispatcher.submitted_content_ids == [content["platform_content_id"]]
        return super().analyze(content, media, source_context)


class PlayUrlAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self) -> None:
        result = fake_gemini_pool()
        result["timing_metrics"] = {
            "video_fetch": {"gemini_video_source": "play_url"},
        }
        super().__init__(default_result=result)

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert media.get("oss_url") is None
        assert media["play_url"].startswith("https://source.example/")
        return super().analyze(content, media, source_context)


class OssAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self) -> None:
        result = fake_gemini_pool()
        result["timing_metrics"] = {
            "video_fetch": {"gemini_video_source": "oss_url"},
        }
        super().__init__(default_result=result)

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert media["oss_url"] == f"https://res.example/{content['platform_content_id']}.mp4"
        assert media["content_media_status"] == "oss_uploaded"
        return super().analyze(content, media, source_context)


class EventGeminiVideoClient(OssAwareGeminiVideoClient):
    def __init__(self, events: list[str], lock: threading.Lock) -> None:
        super().__init__()
        self.events = events
        self.lock = lock

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        with self.lock:
            self.events.append(f"qwen:{content['platform_content_id']}")
        return super().analyze(content, media, source_context)


class RetryOnceQwenClient:
    def __init__(self, events: list[str]) -> None:
        self.events = events
        self.attempts_by_content_id: dict[str, int] = {}

    def analyze_once(
        self,
        content: dict,
        media: dict,
        source_context: dict,
        *,
        attempt_number: int = 1,
        previous_attempts: list[dict] | None = None,
        request_started_at=None,
    ) -> dict:
        content_id = content["platform_content_id"]
        self.attempts_by_content_id[content_id] = self.attempts_by_content_id.get(content_id, 0) + 1
        attempt = self.attempts_by_content_id[content_id]
        self.events.append(f"qwen:{content_id}:{attempt}")
        attempts = [
            *(previous_attempts or []),
            {
                "attempt": attempt_number,
                "status": "failed" if content_id == "retry" and attempt == 1 else "ok",
                "failure_type": "qwen_http_error" if content_id == "retry" and attempt == 1 else None,
                "http_status_code": 429 if content_id == "retry" and attempt == 1 else 200,
            },
        ]
        if content_id == "retry" and attempt == 1:
            result = fake_gemini_fail("qwen_http_error")
            return {
                **result,
                "http_status_code": 429,
                "qwen_retryable": True,
                "qwen_retry_after_seconds": 0.03,
                "timing_metrics": {"qwen_request": {"attempts": attempts}},
            }
        result = fake_gemini_pool()
        return {
            **result,
            "timing_metrics": {"qwen_request": {"attempts": attempts}},
        }

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        return self.analyze_once(content, media, source_context)


def _item(run_id: str, policy_run_id: str, content_id: str, discovery_id: str) -> dict:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "content_discovery_id": discovery_id,
        "search_query_id": "q_001",
        "platform": "douyin",
        "platform_content_id": content_id,
        "platform_content_format": "video",
        "description": "stub",
        "platform_author_id": "author_001",
        "author_display_name": "author",
        "statistics": {"digg_count": 100, "comment_count": 10, "share_count": 5},
        "tags": [],
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": "query_direct",
        "raw_payload": {"platform_content_id": content_id},
    }


def _media(run_id: str, policy_run_id: str, content_id: str) -> dict:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "platform": "douyin",
        "platform_content_id": content_id,
        "content_media_status": "metadata_only",
        "content_metadata_source": "douyin_keyword_search",
        "play_url": f"https://source.example/{content_id}.mp4",
        "local_path": None,
        "oss_url": None,
        "raw_payload": {"platform_content_id": content_id},
    }


def _bundle(item: dict) -> dict:
    return {
        "source_evidence": {"discovered_platform_content_id": item["platform_content_id"]},
        "content": {
            "decision_target_type": "content",
            "decision_target_id": item["platform_content_id"],
            "content_discovery_id": item["content_discovery_id"],
            "search_query_id": item["search_query_id"],
            "platform": item["platform"],
            "platform_content_id": item["platform_content_id"],
            "platform_content_format": "video",
            "description": item["description"],
        },
        "pattern_match_result": {},
        "content_engagement_metrics": {},
        "run_context": {
            "decision_input_snapshot_id": f"evidence_bundle:{item['platform_content_id']}",
        },
    }