"""Source-bound human labels; no browser/provider traffic or real account data."""
from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
import copy
import sqlite3

from fastapi import HTTPException
from fastapi.testclient import TestClient
import pytest

from apps.server.analysis import checked_results, make_sources
from apps.server.analysis_reviews import AXES, model_labels
from apps.server.db import StoreConflict, digest, encoded, identifier, now
from apps.server.main import create_app
from apps.server.tests.test_analysis import answer, seed
from apps.server.tests.test_learning import FakeWorker, client
from apps.server.tests.test_postgres import pg_client


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


def prepared(client, *, region_scope="unrestricted", version="demand-result-v2", status="completed"):
    reading_id, owner = seed(client)
    service = client.app_reference.state.analysis_reviews
    reading = service.analysis.reading.get(reading_id, owner)
    sources, coverage = make_sources(reading)
    conditions = {"service": "装修服务", "target_customer": "有装修需求的业主", "region": "杭州" if region_scope == "specified" else None,
                  "region_scope": region_scope, "exclude_service_promotions": False}
    content = answer({"sources": [source for source in sources if source["text"].strip()]})
    content["results"][0]["region"] = {"value": "杭州", "evidence": "坐标杭州"}
    results = checked_results(encoded(content), sources, "completed", conditions)
    record = {"status": status, "reading_id": reading_id, "sources": sources, "results": results, "coverage": coverage,
              "result_version": version, "context": {"conditions": conditions}}
    run_id = identifier("analysis")
    service.store.execute("INSERT INTO analysis_runs(id,owner_id,reading_run_id,data_json,created_at) VALUES($1,$2,$3,$4,$5)", (run_id, owner, reading_id, encoded(record), now()))
    return service, owner, service.analysis.get(run_id, owner)


def payload(analysis, source_ref="note", **changes):
    result = next(item for item in analysis["results"] if item["source_ref"] == source_ref)
    return {"source_ref": source_ref, "previous_review_id": None, "decision": "confirmed", "labels": model_labels(result), "evidence": [], "note": "", **changes}


def test_review_confirm_correct_chain_latest_summary_and_frozen_sources(review_client):
    client = review_client
    service, owner, analysis = prepared(client)
    endpoint = "/api/discovery/analyses/" + analysis["id"]
    original = service.store.one("SELECT data_json FROM analysis_runs WHERE id=$1", (analysis["id"],))
    original_reading = service.analysis.reading.get(analysis["reading_id"], owner)
    calls = copy.deepcopy(client.worker_reference.calls)
    first = client.post(endpoint + "/reviews", json=payload(analysis))
    assert first.status_code == 201, first.text
    first = first.json()
    assert first["labels"] == first["model_labels"] == payload(analysis)["labels"] and not any(first["axis_diff"].values())
    assert first["analysis_digest"] == digest(analysis) and first["source_digest"] == digest(analysis["sources"][0]) and first["reviewer_id"] == owner
    assert first["previous_review_id"] is None and first["contract_version"] == "analysis-review-v1"
    corrected_labels = {**first["labels"], "role": "unknown"}
    correction = payload(analysis, previous_review_id=first["id"], decision="corrected", labels=corrected_labels, note="保留自身需求，角色仍待人工确认", evidence=["找装修公司"])
    second = client.post(endpoint + "/reviews", json=correction)
    assert second.status_code == 201, second.text
    second = second.json()
    assert second["previous_review_id"] == first["id"] and second["axis_diff"] == {axis: axis == "role" for axis in AXES}
    for ref, decision in (("comment_1", "uncertain"), ("comment_2", "unreviewable")):
        response = client.post(endpoint + "/reviews", json=payload(analysis, ref, decision=decision, labels=None, note="缺少足够文字上下文，保持未判定"))
        assert response.status_code == 201, response.text
    summary = client.get(endpoint + "/review-summary").json()
    assert summary["eligible"] and summary["total"] == summary["reviewed"] == 3 and summary["pending"] == 0 and summary["judged"] == 1
    assert summary["decision_counts"] == {"confirmed": 0, "corrected": 1, "uncertain": 1, "unreviewable": 1}
    assert summary["agreement_count"] == 0 and summary["disagreement_count"] == 1 and summary["agreement_rate"] == 0
    assert summary["axis_diff_counts"] == {axis: int(axis == "role") for axis in AXES}
    assert len(summary["latest_reviews"]) == 3 and summary["latest_reviews"][0] == second and "非准确率" in summary["metric_label"]
    history = client.get(endpoint + "/reviews").json()
    assert history["total"] == 4 and history["items"][-1] == first
    assert client.get("/api/discovery/reviews/" + first["id"]).json() == first
    assert service.store.one("SELECT data_json FROM analysis_runs WHERE id=$1", (analysis["id"],)) == original
    assert service.analysis.reading.get(analysis["reading_id"], owner) == original_reading and client.worker_reference.calls == calls
    expected = StoreConflict if service.store.backend == "postgresql" else sqlite3.IntegrityError
    for operation in ("UPDATE analysis_reviews SET data_json='{}' WHERE id=$1", "DELETE FROM analysis_reviews WHERE id=$1"):
        with pytest.raises(expected):
            service.store.execute(operation, (second["id"],))
    if service.store.backend == "postgresql":
        with pytest.raises(StoreConflict):
            service.store.execute("TRUNCATE analysis_reviews CASCADE")
        restarted = create_app(service.store.home, FakeWorker(service.store.home), database_url=client.test_database_url, database_schema=client.test_schema)
        assert restarted.state.analysis_reviews.get(first["id"], owner) == first
    else:
        restarted = create_app(service.store.home, FakeWorker(service.store.home), database_url="")
        assert restarted.state.analysis_reviews.get(first["id"], owner) == first


def test_review_optimistic_source_chain_and_owner_checks(review_client):
    client = review_client
    service, owner, analysis = prepared(client)
    first = service.create(analysis["id"], payload(analysis), owner)
    cases = [payload(analysis), payload(analysis, previous_review_id="review_nonexistent"), payload(analysis, "comment_1", previous_review_id=first["id"])]
    for body in cases:
        with pytest.raises(HTTPException) as error:
            service.create(analysis["id"], body, owner)
        assert error.value.status_code == 409 and error.value.detail["code"] == "REVIEW_STALE"
    for operation in (lambda: service.create(analysis["id"], payload(analysis), "other-owner"), lambda: service.list(analysis["id"], "other-owner"),
                      lambda: service.get(first["id"], "other-owner"), lambda: service.summary(analysis["id"], "other-owner")):
        with pytest.raises(HTTPException) as error:
            operation()
        assert error.value.status_code == 404
    endpoint = "/api/discovery/analyses/" + analysis["id"]
    with TestClient(client.app_reference, base_url="http://127.0.0.1:8810") as anonymous:
        assert anonymous.get(endpoint + "/reviews").status_code == anonymous.get(endpoint + "/review-summary").status_code == 401
        assert anonymous.post(endpoint + "/reviews", json=payload(analysis)).status_code in {401, 403}
    bad = client.post(endpoint + "/reviews", json={**payload(analysis), "arbitrary": "not allowed"})
    assert bad.status_code == 422 and "not allowed" not in bad.text
    assert len(service.list(analysis["id"], owner)) == 1


def test_review_concurrent_first_and_successor_single_winner(review_client):
    service, owner, analysis = prepared(review_client)
    def compete(body):
        def create(_):
            try:
                return service.create(analysis["id"], body, owner)
            except HTTPException as error:
                return error.detail["code"]
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(pool.map(create, range(2)))
        assert len([result for result in results if isinstance(result, dict)]) == results.count("REVIEW_STALE") == 1
        return next(result for result in results if isinstance(result, dict))
    first = compete(payload(analysis))
    second = compete(payload(analysis, previous_review_id=first["id"], decision="uncertain", labels=None, note="现有文字不足以继续判定"))
    assert second["previous_review_id"] == first["id"] and len(service.list(analysis["id"], owner)) == 2
    summary = service.summary(analysis["id"], owner)
    assert summary["reviewed"] == 1 and summary["judged"] == 0 and summary["agreement_rate"] is None and summary["pending"] == 2


def test_review_contract_quotes_sensitive_notes_and_media_are_not_judged(review_client):
    service, owner, analysis = prepared(review_client)
    changed = {**payload(analysis)["labels"], "role": "unknown"}
    invalid = [
        (payload(analysis, decision="corrected", labels=changed, note="", evidence=["找装修公司"]), "REVIEW_DECISION_INVALID"),
        (payload(analysis, decision="corrected", labels=changed, note="角色待核", evidence=[]), "REVIEW_DECISION_INVALID"),
        (payload(analysis, decision="corrected", note="实际没有改变标签", evidence=["找装修公司"]), "REVIEW_DECISION_INVALID"),
        (payload(analysis, labels=changed), "REVIEW_DECISION_INVALID"),
        (payload(analysis, decision="uncertain", labels=None), "REVIEW_DECISION_INVALID"),
        (payload(analysis, decision="unreviewable", note="理由已填写但不应有标签"), "REVIEW_DECISION_INVALID"),
        (payload(analysis, evidence=["好看，分享一下已完工的效果"]), "REVIEW_EVIDENCE_INVALID"),
        (payload(analysis, note="手机号13812345678"), "REVIEW_SENSITIVE_INPUT"),
        (payload(analysis, note="api_key=synthetic-secret-value"), "REVIEW_SENSITIVE_INPUT"),
        (payload(analysis, note="请联系 a@example.test"), "REVIEW_SENSITIVE_INPUT"),
        (payload(analysis, "comment_2"), "REVIEW_MEDIA_UNREVIEWABLE"),
        (payload(analysis) | {"source_ref": "comment_missing"}, "REVIEW_SOURCE_NOT_FOUND"),
    ]
    for body, code in invalid:
        with pytest.raises(HTTPException) as error:
            service.create(analysis["id"], body, owner)
        assert error.value.detail["code"] == code
    assert service.list(analysis["id"], owner) == []
    media = service.create(analysis["id"], payload(analysis, "comment_2", decision="uncertain", labels=None, note="图片未解释，无法判断发言需求"), owner)
    assert media["labels"] is None and not any(media["axis_diff"].values())


def test_review_label_consistency_region_guard_and_set_comparison(review_client):
    service, owner, analysis = prepared(review_client, region_scope="unspecified")
    original = payload(analysis)["labels"]
    invalid_labels = [original | {"expressions": ["question"]}, original | {"demand": "discussion"}, original | {"role": "suspected_service_provider"},
                      original | {"expressions": ["self_need", "self_need"]}, original | {"expressions": ["self_need", "unknown"]}, original | {"match": "matched"}, original | {"match": "possible"}]
    for labels in invalid_labels:
        with pytest.raises(HTTPException) as error:
            service.create(analysis["id"], payload(analysis, decision="corrected", labels=labels, note="重新核对标签", evidence=["找装修公司"]), owner)
        assert error.value.detail["code"] == "REVIEW_LABELS_INVALID"
    unknown = service.create(analysis["id"], payload(analysis), owner)
    assert unknown["labels"]["match"] == "insufficient" and unknown["decision"] == "confirmed"
    # Expressions compare as a set regardless of maintainer order.
    labels = {**original, "expressions": ["self_need", "question"]}
    first = service.create(analysis["id"], payload(analysis, previous_review_id=unknown["id"], decision="corrected", labels=labels, note="同时表达自身需求和提问", evidence=["找装修公司"]), owner)
    repeated = service.create(analysis["id"], payload(analysis, previous_review_id=first["id"], decision="corrected", labels={**labels, "expressions": list(reversed(labels["expressions"]))}, note="顺序不影响表达集合", evidence=["找装修公司"]), owner)
    assert repeated["labels"] == first["labels"] and repeated["axis_diff"]["expressions"]


def test_review_legacy_failed_and_initial_summary_are_readonly(review_client):
    service, owner, analysis = prepared(review_client, version="demand-result-v1")
    initial = service.summary(analysis["id"], owner)
    assert not initial["eligible"] and initial["total"] == initial["pending"] == initial["judged"] == 0 and initial["agreement_rate"] is None
    assert service.list(analysis["id"], owner) == []
    with pytest.raises(HTTPException) as error:
        service.create(analysis["id"], payload(analysis), owner)
    assert error.value.detail["code"] == "REVIEW_ANALYSIS_INELIGIBLE"
    # Failed records of the new result contract are not eligible either.
    run_id = identifier("analysis")
    service.store.execute("INSERT INTO analysis_runs(id,owner_id,reading_run_id,data_json,created_at) VALUES($1,$2,$3,$4,$5)", (run_id, owner, analysis["reading_id"], encoded({**analysis, "status": "failed", "result_version": "demand-result-v2"}), now()))
    assert not service.summary(run_id, owner)["eligible"]
    with pytest.raises(HTTPException):
        service.create(run_id, payload(analysis), owner)