"""Explicit media-valued fields and bounded, source-bound execution timings."""
from __future__ import annotations

import copy
import json

import pytest
from pydantic import ValidationError

from apps.server.learning_checks import output_assertion_checks
from apps.server.learning_contracts import Action
from apps.server.reading import detail_contract
from apps.server.tests.test_discovery import item, search_candidate
from apps.server.tests.test_learning import client, initialize
from apps.server.tests.test_postgres import pg_client
from apps.server.tests.test_reading import COMMENTS, NOTE_ID, TITLE, comment_output, learned_detail
from apps.server.worker import WorkerError, safe_verify_timings

FIELD = {"name": "body", "source": "content", "text_selector": ".body", "media_selector": ".picture", "required": True, "max_media": 4}


def content_contract():
    contract = copy.deepcopy(COMMENTS)
    contract["fields"][-1] = copy.deepcopy(FIELD)
    return contract


def test_content_is_explicit_required_typed_and_never_matches_media_as_text():
    contract = content_contract()
    assert Action.model_validate(contract).model_dump(exclude_none=True)["fields"][-1] == FIELD
    for invalid_field in ({**FIELD, "required": False}, {**FIELD, "attribute": "id"}, {**FIELD, "text_selector": "body .body"}, {**FIELD, "media_selector": ""},
        {**FIELD, "max_media": True}, {**FIELD, "max_media": 5}, {**FIELD, "source": "text"}, {key: value for key, value in FIELD.items() if key != "text_selector"}):
        with pytest.raises(ValidationError):
            Action.model_validate({**contract, "fields": [*contract["fields"][:-1], invalid_field]})
    for kind, text, media in (("text", "装修需求", []), ("media", None, [{"kind": "image", "selector": ".picture"}]), ("mixed", "装修需求", [{"kind": "video", "selector": "video"}])):
        output = comment_output()
        output["items"][0]["body"] = {"kind": kind, "text": text, "media": media, "media_truncated": False}
        def checks(expectation, current=output):
            return output_assertion_checks({"comments": current}, {"comments": expectation}, {"comments": contract})
        assert all(check["passed"] for check in checks({"state": "nonempty"}))
        assert not all(check["passed"] for check in checks(".picture"))
        assert not all(check["passed"] for check in checks("image"))
        assert all(check["passed"] for check in checks("装修需求")) == (text is not None)
        for bad in ({"kind": "media", "text": None, "media": [], "media_truncated": False}, {"kind": "mixed", "text": "", "media": media, "media_truncated": False},
            {"kind": "media", "text": None, "media": [{"kind": "image", "selector": ".picture", "src": "private-fixture"}], "media_truncated": False},
            {"kind": "media", "text": None, "media": [{"kind": [], "selector": ".picture"}], "media_truncated": False}):
            altered = copy.deepcopy(output)
            altered["items"][0]["body"] = bad
            assert not checks({"state": "nonempty"}, altered)[0]["passed"]
        assert not output_assertion_checks({"comments": output}, {"comments": {"state": "nonempty"}}, {"comments": COMMENTS})[0]["passed"]


def test_media_hints_are_finite_safe_and_never_preserve_urls_or_alt():
    capture = {"url": "http://127.0.0.1/demo", "screenshot_path": "fixture.png", "elements": [{"selector": "#comments", "read_hints": [{"selector": ".picture", "tag": "img", "label": "图片元素（未解释）", "source": "media", "media_kind": "image", "loaded": True, "src": "private-fixture", "alt": "private-fixture", "in_viewport": True, "link": False}]}]}
    hint = WorkerError("fixture", before=capture).before["elements"][0]["read_hints"][0]
    assert hint["source"] == "media" and hint["media_kind"] == "image" and hint["loaded"] is True
    assert "private-fixture" not in json.dumps(hint)
    for kind, loaded in (([], True), ("arbitrary", True), ("image", "true")):
        capture["elements"][0]["read_hints"][0].update(media_kind=kind, loaded=loaded)
        hint = WorkerError("fixture", before=capture).before["elements"][0]["read_hints"][0]
        assert "media_kind" not in hint and "loaded" not in hint and hint.get("source") != "media"


def test_timing_protocol_has_exact_steps_phases_prefix_and_no_private_metadata():
    steps = [{"kind": "fill"}, {"kind": "extract"}]
    timings = [{"step_position": index, "kind": step["kind"], "status": "passed", "duration_ms": 10, "phase_ms": {"execute": 5, "capture_after": 5}} for index, step in enumerate(steps, 1)]
    total = {"duration_ms": 40, "entry_ms": 10, "close_ms": None}
    assert safe_verify_timings(timings, total, steps, passed=True)
    assert safe_verify_timings(timings[:1], total, steps, passed=True) is None
    interrupted = [timings[0], {**timings[1], "status": "interrupted"}]
    assert safe_verify_timings(interrupted, {**total, "close_ms": 4}, steps)
    assert safe_verify_timings([], total, steps)
    for changed in ({**timings[0], "step_position": True}, {**timings[0], "kind": "click"}, {**timings[0], "duration_ms": -1}, {**timings[0], "status": "unknown"},
        {**timings[0], "phase_ms": {"execute": float("inf")}}, {**timings[0], "phase_ms": {"private-fixture": 1}}, {**timings[0], "extra": "private-fixture"}, {**timings[0], "phase_ms": {"execute": 30}}):
        assert safe_verify_timings([changed, timings[1]], total, steps) is None
    assert safe_verify_timings([{**timings[0], "status": "failed"}, timings[1]], total, steps) is None
    overlapping = [{**item, "duration_ms": 80000, "phase_ms": {"execute": 80000}} for item in timings]
    assert safe_verify_timings(overlapping, {"duration_ms": 100000, "entry_ms": 10, "close_ms": None}, steps, passed=True) is None
    overlapping = [{**item, "duration_ms": 10000, "phase_ms": {"execute": 10000}} for item in timings]
    assert safe_verify_timings(overlapping, {"duration_ms": 100000, "entry_ms": 80000, "close_ms": 20000}, steps) is None
    for changed in ({**total, "close_ms": True}, {**total, "entry_ms": 100}, {**total, "duration_ms": 120001}, {**total, "query": "private-fixture"}):
        assert safe_verify_timings(timings, changed, steps) is None


@pytest.fixture(params=["client", "pg_client"])
def content_client(request):
    return request.getfixturevalue(request.param)


def test_content_reading_and_timings_freeze_and_preserve_legacy(content_client):
    client = content_client
    initialize(client)
    _, search = search_candidate(client, [item(TITLE)])
    run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "装修"}).json()
    _, legacy, _ = learned_detail(client)
    _, candidate, state = learned_detail(client, content=True)
    assert detail_contract(candidate["manifest"])
    assert candidate["manifest"]["steps"][-1]["fields"][-1] == FIELD
    endpoint = "/api/discovery/runs/" + run["id"] + "/readings"
    payload = {"detail_candidate_id": candidate["id"], "note_id": NOTE_ID}
    results = []
    for mode, expected_kind in (("normal", "text"), ("media", "media"), ("mixed", "mixed")):
        state["mode"] = mode
        response = client.post(endpoint, json=payload)
        assert response.status_code == 201, response.text
        reading = response.json()
        assert reading["status"] == "completed" and reading["comments"]["items"][0]["body"]["kind"] == expected_kind
        assert len(reading["step_timings"]) == 7 and reading["verify_timing"]["close_ms"] is None
        assert reading["evidence"] and all(client.get(item["screenshot_url"]).status_code == 200 for item in reading["evidence"])
        results.append(reading)
    state["mode"] = "media"
    text_only = client.post("/api/learning/candidates/" + candidate["id"] + "/verify", json={"expected_outputs": {"comments": ".picture"}, "note_binding": {"note_id": NOTE_ID}}).json()
    assert text_only["status"] == "failed"
    state["mode"] = "bad_timing"
    partial = client.post(endpoint, json=payload).json()
    assert partial["status"] == "partial" and "TIMING_TRACE_INVALID" in partial["error_codes"]
    assert "step_timings" not in partial and "verify_timing" not in partial and "private-fixture" not in json.dumps(partial)
    old = client.get("/api/learning/candidates/" + legacy["id"]).json()
    assert old["manifest"] == legacy["manifest"] and old["digest"] == legacy["digest"]
    for reading in results:
        assert client.get("/api/discovery/readings/" + reading["id"]).json() == reading