"""经费标准库 / 税号白名单 / 节假日 / 初审记录 的查询与维护。

预审内核只读本模块数据；维护操作（改标准、增删白名单、查验回填）在此实现。
"""

from __future__ import annotations

import json
import sqlite3
from typing import Any, Optional

from db.connection import get_conn, init_db
from util.logging_config import get_skill_logger
from util.timeutil import now_unix

VERIFY_PENDING = "待查验"
VERIFY_PASS = "已查验通过"
VERIFY_FAIL = "已查验不符"


def _log():
    try:
        return get_skill_logger()
    except RuntimeError:
        return None


def list_standards(category: Optional[str] = None) -> list[dict[str, Any]]:
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        if category:
            cur.execute(
                "SELECT id, category, city_tier, city_example, staff_title, max_amount, unit, basis_doc, clause, effective_date FROM expense_standards WHERE category = ? ORDER BY category, city_tier, staff_title",
                (category,),
            )
        else:
            cur.execute(
                "SELECT id, category, city_tier, city_example, staff_title, max_amount, unit, basis_doc, clause, effective_date FROM expense_standards ORDER BY category, city_tier, staff_title"
            )
        rows = cur.fetchall()
    finally:
        conn.close()
    return [
        {
            "id": r[0],
            "category": r[1],
            "city_tier": r[2],
            "city_example": r[3],
            "staff_title": r[4],
            "max_amount": r[5],
            "unit": r[6],
            "basis_doc": r[7],
            "clause": r[8],
            "effective_date": r[9],
        }
        for r in rows
    ]


def upsert_standard(
    *,
    category: str,
    city_tier: str,
    staff_title: str,
    max_amount: float,
    unit: str = "",
    basis_doc: str = "",
    clause: str = "",
    city_example: str = "",
) -> dict[str, Any]:
    """按 (类别, 城市等级, 职级) 幂等更新标准值。"""
    if not category or not city_tier or not staff_title:
        raise ValueError("类别 / 城市等级 / 职级 均不能为空")
    if max_amount <= 0:
        raise ValueError("标准值必须大于 0")
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT id FROM expense_standards WHERE category = ? AND city_tier = ? AND staff_title = ?",
            (category, city_tier, staff_title),
        )
        row = cur.fetchone()
        if row:
            cur.execute(
                """
                UPDATE expense_standards
                SET max_amount = ?, unit = ?, basis_doc = ?, clause = ?,
                    city_example = COALESCE(NULLIF(?, ''), city_example), updated_at = ?
                WHERE id = ?
                """,
                (max_amount, unit, basis_doc, clause, city_example, now_unix(), int(row[0])),
            )
            standard_id = int(row[0])
            action = "updated"
        else:
            cur.execute(
                """
                INSERT INTO expense_standards
                    (category, city_tier, city_example, staff_title, max_amount, unit, basis_doc, clause, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (category, city_tier, city_example, staff_title, max_amount, unit, basis_doc, clause, now_unix(), now_unix()),
            )
            standard_id = int(cur.lastrowid)
            action = "created"
        conn.commit()
    finally:
        conn.close()
    log = _log()
    if log:
        log.info(
            "standard_upserted standard_id=%s action=%s category=%s tier=%s title=%s stage=standard",
            standard_id,
            action,
            category,
            city_tier,
            staff_title,
        )
    return {"ok": True, "id": standard_id, "action": action, "category": category, "city_tier": city_tier, "staff_title": staff_title, "max_amount": max_amount}


def list_whitelist() -> list[dict[str, Any]]:
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT id, tax_no, org_name, note, created_at FROM taxid_whitelist ORDER BY id")
        rows = cur.fetchall()
    finally:
        conn.close()
    return [{"id": r[0], "tax_no": r[1], "org_name": r[2], "note": r[3], "created_at": r[4]} for r in rows]


def add_whitelist(tax_no: str, org_name: str, note: str = "") -> dict[str, Any]:
    tax_no = (tax_no or "").strip()
    if not tax_no:
        raise ValueError("税号不能为空")
    if not org_name:
        raise ValueError("单位名称不能为空")
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT id FROM taxid_whitelist WHERE tax_no = ?", (tax_no,))
        if cur.fetchone():
            raise ValueError(f"税号已存在：{tax_no}")
        cur.execute(
            "INSERT INTO taxid_whitelist (tax_no, org_name, note, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
            (tax_no, org_name, note, now_unix(), now_unix()),
        )
        whitelist_id = int(cur.lastrowid)
        conn.commit()
    finally:
        conn.close()
    return {"ok": True, "id": whitelist_id, "tax_no": tax_no, "org_name": org_name}


def remove_whitelist(tax_no: str) -> dict[str, Any]:
    tax_no = (tax_no or "").strip()
    if not tax_no:
        raise ValueError("税号不能为空")
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("DELETE FROM taxid_whitelist WHERE tax_no = ?", (tax_no,))
        deleted = cur.rowcount
        conn.commit()
    finally:
        conn.close()
    if not deleted:
        raise ValueError(f"白名单中不存在该税号：{tax_no}")
    return {"ok": True, "removed": tax_no}


def list_holidays(year: Optional[str] = None) -> list[dict[str, Any]]:
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        if year:
            cur.execute(
                "SELECT id, holiday_date, holiday_name FROM holiday_calendar WHERE holiday_date LIKE ? ORDER BY holiday_date",
                (f"{year}%",),
            )
        else:
            cur.execute("SELECT id, holiday_date, holiday_name FROM holiday_calendar ORDER BY holiday_date")
        rows = cur.fetchall()
    finally:
        conn.close()
    return [{"id": r[0], "holiday_date": r[1], "holiday_name": r[2]} for r in rows]


def list_reports(limit: int = 20, verdict: Optional[str] = None) -> list[dict[str, Any]]:
    init_db()
    limit = max(1, min(int(limit), 200))
    conn = get_conn()
    try:
        cur = conn.cursor()
        if verdict:
            cur.execute(
                """
                SELECT id, batch_no, invoice_number, seller_name, amount, verdict, verify_status, summary, report_file, created_at
                FROM prereview_reports WHERE verdict = ? ORDER BY id DESC LIMIT ?
                """,
                (verdict, limit),
            )
        else:
            cur.execute(
                """
                SELECT id, batch_no, invoice_number, seller_name, amount, verdict, verify_status, summary, report_file, created_at
                FROM prereview_reports ORDER BY id DESC LIMIT ?
                """,
                (limit,),
            )
        rows = cur.fetchall()
    finally:
        conn.close()
    return [
        {
            "id": r[0],
            "batch_no": r[1],
            "invoice_number": r[2],
            "seller_name": r[3],
            "amount": r[4],
            "verdict": r[5],
            "verify_status": r[6],
            "summary": r[7],
            "report_file": r[8],
            "created_at": r[9],
        }
        for r in rows
    ]


def get_report(report_id: int) -> dict[str, Any]:
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT * FROM prereview_reports WHERE id = ?", (report_id,))
        row = cur.fetchone()
        if not row:
            raise ValueError(f"初审报告不存在：{report_id}")
        columns = [d[0] for d in cur.description]
        report = dict(zip(columns, row))
        invoice: dict[str, Any] = {}
        if report.get("invoice_id"):
            cur.execute("SELECT * FROM invoice_records WHERE id = ?", (report["invoice_id"],))
            irow = cur.fetchone()
            if irow:
                icolumns = [d[0] for d in cur.description]
                invoice = dict(zip(icolumns, irow))
    finally:
        conn.close()
    for key in ("pass_json", "warn_json", "reject_json"):
        try:
            report[key] = json.loads(report.get(key) or "[]")
        except (TypeError, ValueError):
            report[key] = []
    report["invoice"] = invoice
    return report


def mark_verified(report_id: int, result: str, note: str = "") -> dict[str, Any]:
    """回填人工查验结果；查验“不符”则结论升级为驳回。"""
    result = (result or "").strip().lower()
    if result not in ("pass", "fail"):
        raise ValueError("查验结果只接受 pass / fail")
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT id, verdict, reject_json FROM prereview_reports WHERE id = ?", (report_id,))
        row = cur.fetchone()
        if not row:
            raise ValueError(f"初审报告不存在：{report_id}")
        _report_id, verdict, reject_json = int(row[0]), str(row[1]), row[2] or "[]"

        if result == "pass":
            verify_status = VERIFY_PASS
        else:
            verify_status = VERIFY_FAIL
            try:
                rejects = json.loads(reject_json)
            except (TypeError, ValueError):
                rejects = []
            rejects.append(
                {
                    "rule_id": "V1",
                    "name": "发票真伪",
                    "level": "reject",
                    "detail": "人工查验回填：票面信息与查验平台不符，判定为问题票据。",
                    "rule_ref": "全国增值税发票查验平台（inv-veri.chinatax.gov.cn）",
                }
            )
            reject_json = json.dumps(rejects, ensure_ascii=False)

        new_verdict = "驳回" if result == "fail" else verdict
        cur.execute(
            """
            UPDATE prereview_reports
            SET verify_status = ?, verify_note = ?, verify_updated_at = ?,
                verdict = ?, reject_json = ?
            WHERE id = ?
            """,
            (verify_status, note, now_unix(), new_verdict, reject_json, report_id),
        )
        conn.commit()
    finally:
        conn.close()
    log = _log()
    if log:
        log.info(
            "verify_marked report_id=%s result=%s verdict=%s stage=verify",
            report_id,
            result,
            new_verdict,
        )
    return {
        "ok": True,
        "report_id": report_id,
        "verify_status": verify_status,
        "verdict": new_verdict,
        "note": note,
    }


def stats() -> dict[str, Any]:
    init_db()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT verdict, COUNT(*) FROM prereview_reports GROUP BY verdict")
        by_verdict = {str(r[0]): int(r[1]) for r in cur.fetchall()}
        cur.execute("SELECT COUNT(*) FROM invoice_records")
        invoice_total = int(cur.fetchone()[0])
        cur.execute("SELECT COUNT(*) FROM taxid_whitelist")
        whitelist_total = int(cur.fetchone()[0])
        cur.execute("SELECT COUNT(*) FROM expense_standards")
        standards_total = int(cur.fetchone()[0])
        cur.execute(
            """
            SELECT seller_name, COUNT(*) AS cnt FROM invoice_records
            WHERE seller_name != ''
            GROUP BY seller_name ORDER BY cnt DESC LIMIT 5
            """
        )
        top_sellers = [{"seller_name": r[0], "count": int(r[1])} for r in cur.fetchall()]
    finally:
        conn.close()
    total_reports = sum(by_verdict.values())
    return {
        "ok": True,
        "invoice_total": invoice_total,
        "report_total": total_reports,
        "verdict_counts": by_verdict,
        "reject_rate": round(by_verdict.get("驳回", 0) / total_reports, 4) if total_reports else 0.0,
        "whitelist_total": whitelist_total,
        "standards_total": standards_total,
        "top_sellers": top_sellers,
    }