# -*- coding: utf-8 -*-
"""providers 模块测试：判定 JSON 解析、分类归一化、图片编码、usage 规整、output_schema、重试。"""
import asyncio
import base64

import pytest

from violation_detector.config import AppConfig
from violation_detector.providers import (
    CANON, OUTPUT_SCHEMA, SCHEMA_CATEGORIES, SCHEMA_NAME, _chat_detect,
    _extract_output_text, _norm_usage, _result_responses,
    _schema_explicitly_unsupported, build_chat_payload,
    build_responses_payload, encode_image, norm_category, parse_judgment,
)

GOOD = ('{"文字/图形属性": "黑色T恤印字母", "侵权/违规逻辑": "未授权使用商标", '
        '"违规分类": "12. 侵权 - 除人物外的其他侵权"}')


def test_parse_plain_json():
    r = parse_judgment(GOOD)
    assert r is not None
    assert r[0] == "黑色T恤印字母"
    assert r[1] == "未授权使用商标"
    assert r[2] == "12. 侵权 - 除人物外的其他侵权"


def test_parse_fenced_json():
    r = parse_judgment("```json\n" + GOOD + "\n```")
    assert r is not None and r[2].startswith("12.")


def test_parse_json_with_surrounding_text():
    r = parse_judgment("分析如下：\n" + GOOD + "\n以上。")
    assert r is not None


def test_parse_invalid():
    assert parse_judgment("") is None
    assert parse_judgment("这不是JSON") is None
    assert parse_judgment('{"文字/图形属性": "x"}') is None  # 缺字段
    assert parse_judgment('{"a": [1,2]') is None  # 截断


def test_norm_category_variants():
    # 不同写法归一到标准 17 类名称
    assert norm_category("12. 侵权 - 人物外") == CANON[12]
    assert norm_category("14脏话、侮辱性") == CANON[14]
    assert norm_category("5. 负向敏感") == CANON[5]
    # 非编号类原样保留
    assert norm_category("无违规") == "无违规"
    assert norm_category("违规不明") == "违规不明"
    # 未知编号保留原文
    assert norm_category("99. 未来分类") == "99. 未来分类"


def test_encode_image_roundtrip(tmp_path):
    img = tmp_path / "a.jpg"
    img.write_bytes(b"\xff\xd8\xff\xe0fake")
    url = encode_image(str(img))
    assert url.startswith("data:image/jpeg;base64,")
    assert base64.b64decode(url.split(",", 1)[1]) == b"\xff\xd8\xff\xe0fake"


def test_encode_image_unsupported(tmp_path):
    p = tmp_path / "a.txt"
    p.write_text("x", encoding="utf-8")
    with pytest.raises(ValueError):
        encode_image(str(p))


def test_norm_usage_handles_missing_details():
    assert _norm_usage({"prompt_tokens": 100, "completion_tokens": 50}) == {
        "prompt": 100, "completion": 50, "cached": 0}
    assert _norm_usage({}) == {"prompt": None, "completion": None, "cached": 0}
    assert _norm_usage({"prompt_tokens": 100, "prompt_tokens_details": {"cached_tokens": 80}})["cached"] == 80


# ---------- output_schema ----------

def test_output_schema_shape():
    assert OUTPUT_SCHEMA["type"] == "object"
    assert set(OUTPUT_SCHEMA["required"]) == {"文字/图形属性", "侵权/违规逻辑", "违规分类"}
    assert OUTPUT_SCHEMA["additionalProperties"] is False
    assert set(OUTPUT_SCHEMA["properties"]) == set(OUTPUT_SCHEMA["required"])


def test_schema_categories_covers_canon_and_specials():
    expect = [CANON[n] for n in sorted(CANON)] + ["无违规", "违规不明"]
    assert SCHEMA_CATEGORIES == expect


def test_build_chat_payload_schema_vs_json():
    # 豆包：结构化输出 = response_format.json_schema(strict)；降级 = json_object
    messages = [{"role": "user", "content": []}]
    s = build_chat_payload("m", messages, {"temperature": 0}, True)
    assert s["model"] == "m" and s["temperature"] == 0
    rf = s["response_format"]
    assert rf["type"] == "json_schema"
    assert rf["json_schema"]["name"] == SCHEMA_NAME
    assert rf["json_schema"]["strict"] is True
    assert rf["json_schema"]["schema"] is OUTPUT_SCHEMA
    j = build_chat_payload("m", messages, {}, False)
    assert j["response_format"] == {"type": "json_object"}


def test_build_responses_payload_schema():
    # DeepSeek：结构化输出 = text.format.json_schema（Responses API）
    items = [{"role": "user", "content": [{"type": "input_text", "text": "p"}]}]
    p = build_responses_payload("m", items, {"max_output_tokens": 5000}, True)
    assert p["model"] == "m" and p["max_output_tokens"] == 5000
    fmt = p["text"]["format"]
    assert fmt["type"] == "json_schema"
    assert fmt["name"] == SCHEMA_NAME
    assert fmt["schema"] is OUTPUT_SCHEMA


def test_extract_output_text_concats_output_text():
    data = {"output": [
        {"type": "reasoning", "role": "assistant",
         "content": [{"type": "reasoning_text", "text": "思考"}],
         "status": "completed"},
        {"type": "message", "role": "assistant",
         "content": [{"type": "output_text", "text": '{"a": 1}'}],
         "status": "completed"},
    ]}
    assert _extract_output_text(data) == '{"a": 1}'


def test_result_responses_ok():
    data = {
        "status": "completed",
        "usage": {"input_tokens": 10, "output_tokens": 20,
                  "input_tokens_details": {"cached_tokens": 4}},
        "output": [{"type": "message", "role": "assistant", "content": [
            {"type": "output_text", "text": '{"文字/图形属性": "T恤", '
                                            '"侵权/违规逻辑": "无", "违规分类": "无违规"}'}]}],
    }
    r = _result_responses("a.jpg", data)
    assert r["status"] == "ok" and r["category"] == "无违规"
    assert r["usage"] == {"prompt": 10, "completion": 20, "cached": 4}
    assert r["finish_reason"] == "completed"


def test_schema_explicitly_unsupported():
    assert _schema_explicitly_unsupported(
        "Unrecognized request argument supplied: response_format.json_schema")
    assert _schema_explicitly_unsupported("model not support json_schema")
    assert _schema_explicitly_unsupported("不支持的参数 json_schema")
    assert not _schema_explicitly_unsupported("HTTP 400 some other problem")


# ---------- 重试机制（错误码 / 空正文 / 解析失败） ----------

RETRY_GOOD = '{"文字/图形属性": "T恤", "侵权/违规逻辑": "无", "违规分类": "无违规"}'
RETRY_BAD = '{"文字/图形属性": "T恤"}'  # 缺字段，解析失败


def _chat_ok(content):
    return {"choices": [{"message": {"content": content}, "finish_reason": "stop"}],
            "usage": {}}


class _FakeResp:
    def __init__(self, status, data=None, body=""):
        self.status = status
        self._data = data
        self._body = body

    async def json(self):
        return self._data

    async def text(self):
        return self._body


class _FakeCM:
    def __init__(self, resp):
        self._resp = resp

    async def __aenter__(self):
        return self._resp

    async def __aexit__(self, *exc):
        return False


class _FakeSession:
    def __init__(self, responses):
        self.responses = list(responses)
        self.posts = 0

    def post(self, url, json, headers):
        resp = self.responses[min(self.posts, len(self.responses) - 1)]
        self.posts += 1
        return _FakeCM(resp)


def _run_chat(session, retries=3):
    cfg = AppConfig(output_schema_mode="off", retries=retries)
    return asyncio.run(_chat_detect(
        session, asyncio.Semaphore(1), cfg, "a.jpg", stop=lambda: False,
        provider="deepseek", label="DeepSeek",
        base_url="https://fake.deepseek.com", api_key="k", model="m",
        messages=[{"role": "user", "content": "p"}],
        extras={"temperature": 0, "max_tokens": 50}))


async def _no_sleep(_):
    return None


def test_parse_fail_retries_then_recovers(monkeypatch):
    # 第 1 次返回不可解析 → 触发重试；第 2 次成功
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD)),
                         _FakeResp(200, _chat_ok(RETRY_GOOD))])
    res = _run_chat(sess)
    assert res["status"] == "ok" and res["category"] == "无违规"
    assert sess.posts == 2


def test_parse_fail_exhausts_to_parse_fail(monkeypatch):
    # 一直解析失败 → 第 1 次进入解析段，独立再重试 3 次后落 parse_fail（违规不明）
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD))])
    res = _run_chat(sess, retries=3)
    assert res["status"] == "parse_fail" and res["category"] == "违规不明"
    assert sess.posts == 1 + 3  # 首判 + 3 次独立解析重试


def test_parse_fail_dedicated_budget_ignores_http_errors(monkeypatch):
    # 先 1 次 HTTP 500，再解析失败：HTTP 错误不消耗解析段次数，
    # 解析仍独立获得 cfg.retries 次专用重试
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(500, body="boom"),
                         _FakeResp(200, _chat_ok(RETRY_BAD))])
    res = _run_chat(sess, retries=3)
    assert res["status"] == "parse_fail" and res["category"] == "违规不明"
    assert sess.posts == 2 + 3  # 1×HTTP500 + 首判解析失败 + 3×解析专用重试


def test_error_code_retries_then_recovers(monkeypatch):
    # HTTP 500 → 退避重试；随后 200 成功
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(500, body="boom"),
                         _FakeResp(200, _chat_ok(RETRY_GOOD))])
    res = _run_chat(sess)
    assert res["status"] == "ok" and res["category"] == "无违规"
    assert sess.posts == 2


def test_error_code_exhausts_to_error(monkeypatch):
    # 一直 HTTP 500 → 重试耗尽落 error（检测异常）
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(500, body="boom")])
    res = _run_chat(sess, retries=3)
    assert res["status"] == "error" and res["category"] == "检测异常"
    assert sess.posts == 3


def test_empty_content_retries(monkeypatch):
    # 空正文 → 重试；随后成功
    monkeypatch.setattr(asyncio, "sleep", _no_sleep)
    sess = _FakeSession([_FakeResp(200, _chat_ok("")),
                         _FakeResp(200, _chat_ok(RETRY_GOOD))])
    res = _run_chat(sess)
    assert res["status"] == "ok" and res["category"] == "无违规"
    assert sess.posts == 2