"""敏感词扫描器——增量水位(fetched_at)+ 首次全量回填。

水位语义(spec §3.6 P2):last_fetched_at = MAX(message.fetched_at)。增量
补扫拉回的历史消息 timestamp 早于水位但 fetched_at 是新的,按 timestamp
做水位会永久漏扫。等值边界行可能重扫——UNIQUE(word_id,msg_id)+已有命中
跳过兜底,幂等。

方言:MySQL ngram MATCH(≥2字,快);单字词与其他方言走 LIKE。语义差
异(MATCH 分词 vs LIKE 子串)可接受——命中可处置,误报治理闭环在。

分批:增量直接拉水位后行集(4h 增量仅数千);全量回填按词扫——MATCH 在
服务端把 86 万行预筛成命中行(每词仅数千),整批取回后按 500 行分片插
入/flush 控挂起对象数。不流式消费游标:MySQL 驱动(asyncmy)下结果未
取尽时同连接发下一条语句会 out-of-sync,全量路径无测试覆盖,取保守形。
"""
from __future__ import annotations

import datetime as dt
import logging

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.analytics.sensitive.words import _hit_exclude
from app.db.models import Message, SensitiveHit, SensitiveScanState, SensitiveWord

logger = logging.getLogger(__name__)

ENTITY = "message_text"
_SNIPPET_RADIUS = 30  # 命中词前后各取 30 字


def _snippet(text: str, word: str) -> str:
    i = text.find(word)
    if i < 0:
        return text[: _SNIPPET_RADIUS * 2]
    lo, hi = max(0, i - _SNIPPET_RADIUS), min(len(text), i + len(word) + _SNIPPET_RADIUS)
    return text[lo:hi]


def _text_match_cond(word: str, dialect: str) -> list:
    """全量回填路径:MySQL ≥2字 → ngram MATCH;否则 LIKE(单字 ngram 不可用)。"""
    if dialect == "mysql" and len(word) >= 2:
        phrase = word.replace('"', "").replace("*", "")
        return [
            sa.text("MATCH(message.text) AGAINST (:w IN BOOLEAN MODE)")
            .bindparams(w=f'"{phrase}"')
        ]
    escaped = word.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return [Message.text.like(f"%{escaped}%", escape="\\")]


async def _get_state(session: AsyncSession) -> SensitiveScanState:
    st = await session.get(SensitiveScanState, ENTITY)
    if st is None:
        st = SensitiveScanState(entity=ENTITY)
        session.add(st)
        await session.flush()
    return st


async def _insert_hits(
    session: AsyncSession, word: SensitiveWord, rows: list, existing: set[str]
) -> int:
    n = 0
    for msg_id, channel, owner, talker, room, ts, text in rows:
        if msg_id in existing:
            continue
        if _hit_exclude(text, word.word, word.exclude_words or None):
            continue
        session.add(
            SensitiveHit(
                word_id=word.id, msg_id=msg_id, channel=channel,
                owner_account_id=owner, talker_id=talker, room_id=room,
                ts=ts, snippet=_snippet(text or "", word.word),
            )
        )
        existing.add(msg_id)
        n += 1
    return n


async def _scan_batch(
    session: AsyncSession, word: SensitiveWord, rows: list
) -> int:
    """对一个词扫一批 (msg_id..text) 行:查已有命中、插新命中。"""
    sel_rows = [tuple(r[:7]) for r in rows if word.word in (r[6] or "")]
    if not sel_rows:
        return 0
    existing = set(
        (
            await session.execute(
                sa.select(SensitiveHit.msg_id).where(
                    SensitiveHit.word_id == word.id,
                    SensitiveHit.msg_id.in_(
                        [r[0] for r in sel_rows] or ["-"]
                    ),
                )
            )
        ).scalars().all()
    )
    return await _insert_hits(session, word, sel_rows, existing)


async def run_incremental(session: AsyncSession) -> dict:
    """增量扫描:fetched_at > 水位 的行 × 全部 enabled 词。

    增量行少(4h 数千)——统一 Python 侧子串判定(=LIKE 语义,精确;MATCH
    只在全量回填的 MySQL 路径用,见 _text_match_cond)。
    """
    words = (
        await session.execute(
            sa.select(SensitiveWord).where(SensitiveWord.enabled.is_(True))
        )
    ).scalars().all()
    st = await _get_state(session)

    conds = [Message.text.is_not(None), Message.text != ""]
    if st.last_fetched_at is not None:
        conds.append(Message.fetched_at > st.last_fetched_at)
    rows = (
        await session.execute(
            sa.select(
                Message.msg_id, Message.channel, Message.owner_account_id,
                Message.talker_id, Message.room_id, Message.timestamp,
                Message.text, Message.fetched_at,
            ).where(*conds)
        )
    ).all()

    scanned = len(rows)
    hits = 0
    for w in words:
        hits += await _scan_batch(session, w, rows)

    new_wm = max((r[7] for r in rows), default=None) if rows else None
    if new_wm is not None:
        st.last_fetched_at = new_wm
    st.last_run_at = dt.datetime.now(dt.UTC)
    st.scanned += scanned
    st.hits += hits
    await session.flush()
    return {"scanned": scanned, "hits": hits, "new_watermark": new_wm}


async def run_full(session: AsyncSession) -> dict:
    """全量回填(CLI 一次性):每词单独扫全库;水位推进同增量。

    每词结果整批取回(MATCH/LIKE 已在服务端预筛成命中行)再分片复用
    _scan_batch——不在未取尽的结果上发新语句/flush(MySQL 驱动会
    out-of-sync)。日志只打数量,不打词与正文。
    """
    dialect = session.bind.dialect.name if session.bind else ""
    words = (
        await session.execute(
            sa.select(SensitiveWord).where(SensitiveWord.enabled.is_(True))
        )
    ).scalars().all()
    st = await _get_state(session)
    total_hits = 0
    for w in words:
        conds = [Message.text.is_not(None), Message.text != ""]
        conds += _text_match_cond(w.word, dialect)
        stmt = sa.select(
            Message.msg_id, Message.channel, Message.owner_account_id,
            Message.talker_id, Message.room_id, Message.timestamp, Message.text,
        ).where(*conds)
        rows = (await session.execute(stmt)).all()
        for i in range(0, len(rows), 500):
            total_hits += await _scan_batch(session, w, rows[i:i + 500])
        await session.flush()
    new_wm = (
        await session.execute(sa.select(sa.func.max(Message.fetched_at)))
    ).scalar_one_or_none()
    if new_wm is not None:
        st.last_fetched_at = new_wm
    st.last_run_at = dt.datetime.now(dt.UTC)
    await session.flush()
    logger.info("[sensitive-scan] full backfill done: words=%s hits=%s",
                len(words), total_hits)
    return {"words": len(words), "hits": total_hits}