"""C1 词库 CRUD + 预览。"""
from __future__ import annotations

import datetime as dt

from app.analytics.sensitive import scanner
from app.analytics.sensitive import words as words_svc
from app.db.models import Message, SensitiveHit, SensitiveScanState


async def test_create_and_duplicate(make_client):
    async def seed(session):
        w = await words_svc.create_word(
            session, word="飞单", category="私单", exclude_words=["飞单群"]
        )
        assert w.id is not None and w.enabled is True
        try:
            await words_svc.create_word(session, word="飞单")
            raise AssertionError("should raise")
        except ValueError:
            pass

    await make_client(seed)


async def test_preview_count_and_excludes(make_client):
    async def seed(session):
        ts = 1756900000000
        for i, text in enumerate(["今天聊飞单的事", "飞单群公告", "正常聊天"]):
            session.add(Message(msg_id=f"m{i}", channel="wechat",
                                owner_account_id="wx1", talker_id="t1",
                                timestamp=ts + i, is_send=False, type="text",
                                text=text, fetched_at=dt.datetime(2026, 9, 4)))
        await session.commit()
        n1 = await words_svc.preview_count(session, "飞单", None)
        assert n1 == 2  # SQLite 测试库走 LIKE 路径
        n2 = await words_svc.preview_count(session, "飞单", ["飞单群"])
        assert n2 == 1  # 排除词剔除「飞单群公告」

    await make_client(seed)


async def test_api_words_crud(make_client):
    async def seed(session):
        ...

    client = await make_client(seed)
    r = await client.post("/api/analytics/sensitive/words",
                          json={"word": "私单", "category": "飞单"})
    assert r.status_code == 200 and r.json()["word"] == "私单"
    r = await client.post("/api/analytics/sensitive/words",
                          json={"word": "私单"})
    assert r.status_code == 409
    r = await client.post("/api/analytics/sensitive/words/preview",
                          json={"word": "私单"})
    assert r.status_code == 200 and r.json()["count"] == 0
    wid = 1
    r = await client.patch(f"/api/analytics/sensitive/words/{wid}",
                           json={"enabled": False})
    assert r.status_code == 200 and r.json()["enabled"] is False
    r = await client.delete(f"/api/analytics/sensitive/words/{wid}")
    assert r.status_code == 200


# 注:单字词 MySQL 拒预览分支(words.py ValueError)在 SQLite 测试库不可达
# ——dialect 判定不满足,不做单测;由注释 + 路由 400 映射保障。


async def test_api_patch_unknown_404(make_client):
    async def seed(session):
        ...

    client = await make_client(seed)
    r = await client.patch("/api/analytics/sensitive/words/999",
                           json={"enabled": False})
    assert r.status_code == 404


async def test_api_delete_unknown_404(make_client):
    async def seed(session):
        ...

    client = await make_client(seed)
    r = await client.delete("/api/analytics/sensitive/words/999")
    assert r.status_code == 404


async def test_api_create_empty_word_409(make_client):
    async def seed(session):
        ...

    client = await make_client(seed)
    r = await client.post("/api/analytics/sensitive/words",
                          json={"word": "   "})
    assert r.status_code == 409


# ---------------------------------------------------------------- C1 scanner

def _msg(msg_id, text, fetched, ts=1756900000000, talker="t1"):
    return Message(msg_id=msg_id, channel="wechat", owner_account_id="wx1",
                   talker_id=talker, timestamp=ts, is_send=False, type="text",
                   text=text, fetched_at=fetched)


async def test_scan_incremental_and_idempotent(make_client):
    async def seed(session):
        f1 = dt.datetime(2026, 9, 4, 10, 0)
        session.add(_msg("m1", "聊聊私单", f1))
        session.add(_msg("m2", "正常", f1))
        await words_svc.create_word(session, word="私单")
        await session.commit()
        r1 = await scanner.run_incremental(session)
        await session.commit()
        assert r1["hits"] == 1
        assert r1["new_watermark"] == f1
        r2 = await scanner.run_incremental(session)  # 再跑:水位之后无新行
        await session.commit()
        assert r2["hits"] == 0
        from sqlalchemy import func, select

        n = (await session.execute(
            select(func.count()).select_from(SensitiveHit)
        )).scalar_one()
        assert n == 1  # 幂等不重复

    await make_client(seed)


async def test_late_arriving_history_scanned(make_client):
    """迟到数据:timestamp 很老但 fetched_at 新 → 必须被扫到(P2 水位设计)。"""
    async def seed(session):
        old = dt.datetime(2026, 9, 1, 8, 0)
        session.add(_msg("m_old", "历史私单消息", old))
        await words_svc.create_word(session, word="私单")
        await session.commit()
        r1 = await scanner.run_incremental(session)
        await session.commit()
        assert r1["hits"] == 1
        # 迟到行:timestamp 比 m_old 的还老(fetched_at 现在)——
        # 错误的 timestamp 水位实现会漏掉它,fetched_at 水位必须抓到
        session.add(_msg("m_late", "迟到的私单", dt.datetime(2026, 9, 4, 12, 0),
                         ts=1735000000000))
        await session.commit()
        r2 = await scanner.run_incremental(session)
        await session.commit()
        assert r2["hits"] == 1  # 按 fetched_at 水位,迟到历史不漏

    await make_client(seed)


async def test_single_char_like_fallback(make_client):
    async def seed(session):
        session.add(_msg("m1", "坑爹", dt.datetime(2026, 9, 4, 10, 0)))
        await words_svc.create_word(session, word="坑")  # 单字
        await session.commit()
        r = await scanner.run_incremental(session)
        await session.commit()
        assert r["hits"] == 1  # 单字走 LIKE 子串,增量行少语义精确

    await make_client(seed)


async def test_disposition_preserved(make_client):
    """重扫已处置行不覆盖 disposition。"""
    async def seed(session):
        session.add(_msg("m1", "私单", dt.datetime(2026, 9, 4, 10, 0)))
        await words_svc.create_word(session, word="私单")
        await session.commit()
        await scanner.run_incremental(session)
        await session.commit()
        from sqlalchemy import select

        hit = (await session.execute(
            select(SensitiveHit)
        )).scalars().one()
        hit.disposition = "false_positive"
        await session.commit()
        # 重扫同词同消息(水位回拨模拟)
        st = await session.get(SensitiveScanState, "message_text")
        st.last_fetched_at = None
        await session.commit()
        await scanner.run_incremental(session)
        await session.commit()
        hit2 = (await session.execute(
            select(SensitiveHit)
        )).scalars().one()
        assert hit2.disposition == "false_positive"

    await make_client(seed)


async def test_full_backfill_window_rows_scanned_by_incremental(
    make_client, monkeypatch
):
    """回填窗口内新入库的行不漏:水位必须在词循环前取。

    若 run_full 在词循环后才读 MAX(fetched_at),窗口内 commit 的行
    fetched_at ≤ 水位,>= 增量也扫不到——首部署夜间回填与 02:00 全量
    批写重叠时漏扫。m_max(无命中)垫高窗口内 MAX,锁死判别力。
    """
    async def seed(session):
        session.add(_msg("m1", "私单", dt.datetime(2026, 9, 4, 10, 0)))
        await words_svc.create_word(session, word="私单")
        await session.commit()

        orig = scanner._scan_batch
        fired = False

        async def _hook(sess, word, rows):
            nonlocal fired
            if not fired:  # 首个分片时模拟回填窗口内新消息入库
                fired = True
                sess.add(_msg("m_win", "窗口内私单", dt.datetime(2026, 9, 4, 11, 0)))
                sess.add(_msg("m_max", "正常", dt.datetime(2026, 9, 4, 12, 0)))
                await sess.commit()
            return await orig(sess, word, rows)

        monkeypatch.setattr(scanner, "_scan_batch", _hook)
        await scanner.run_full(session)  # 词行集已物化,窗口行不在其中
        await session.commit()
        st = await session.get(SensitiveScanState, "message_text")
        # 水位是扫描前取的(10:00),未被窗口行推后
        assert st.last_fetched_at == dt.datetime(2026, 9, 4, 10, 0)
        r = await scanner.run_incremental(session)
        await session.commit()
        assert r["hits"] == 1  # 窗口内命中行(m_win)由增量补上

    await make_client(seed)


# ------------------------------------------------------------- C1 命中列表/处置

async def test_hits_list_default_filter_and_dispose(make_client):
    async def seed(session):
        w = await words_svc.create_word(session, word="私单")
        session.add(SensitiveHit(word_id=w.id, msg_id="m1", channel="wechat",
                                 ts=1756900000000, snippet="…私单…",
                                 disposition="pending"))
        session.add(SensitiveHit(word_id=w.id, msg_id="m2", channel="wechat",
                                 ts=1756900001000, snippet="…私单…",
                                 disposition="ignored"))
        # 排序/word_id 过滤双断言用:同 disposition 两行、不同词、ts 拉开
        w2 = await words_svc.create_word(session, word="飞单")
        session.add(SensitiveHit(word_id=w.id, msg_id="m3", channel="wechat",
                                 ts=1756900005000, snippet="…",
                                 disposition="false_positive"))
        session.add(SensitiveHit(word_id=w2.id, msg_id="m4", channel="wechat",
                                 ts=1756900002000, snippet="…",
                                 disposition="false_positive"))
        await session.commit()

    client = await make_client(seed)
    r = await client.get("/api/analytics/sensitive/hits")
    body = r.json()
    assert body["total"] == 1  # 默认只 pending+confirmed
    ids = [h["id"] for h in body["items"]]
    r2 = await client.get("/api/analytics/sensitive/hits",
                          params={"disposition": "ignored"})
    assert r2.json()["total"] == 1
    r3 = await client.patch("/api/analytics/sensitive/hits",
                            json={"ids": ids, "disposition": "confirmed"})
    assert r3.json()["updated"] == 1
    r4 = await client.get("/api/analytics/sensitive/hits",
                          params={"disposition": "confirmed"})
    assert r4.json()["total"] == 1
    # ts 降序:m3(ts 更大)在前;m4 属他词被 word_id 过滤掉
    r5 = await client.get("/api/analytics/sensitive/hits",
                          params={"disposition": "false_positive"})
    fp = r5.json()
    assert fp["total"] == 2
    assert [h["msg_id"] for h in fp["items"]] == ["m3", "m4"]
    r6 = await client.get("/api/analytics/sensitive/hits",
                          params={"disposition": "false_positive",
                                  "word_id": fp["items"][0]["word_id"]})
    assert r6.json()["total"] == 1
    r7 = await client.get("/api/analytics/sensitive/hits",
                          params={"disposition": "bogus"})
    assert r7.status_code == 422  # 拼错值响亮 422,不静默空列表