code/ai_finance_agent/backend/app/application/services.py

"""应用公共能力：权限、不可变快照与配置校验收敛，防止路由间业务口径漂移。"""
from collections import Counter
from contextvars import ContextVar
from datetime import datetime, timedelta
from decimal import Decimal
import math
import re
from zoneinfo import ZoneInfo
from sqlalchemy import inspect, select
from app.domain.models import AgentConfig, Alert, AuditRule, AuditRun, Department, Invoice, KnowledgeChunk, KnowledgeDocument, OperationLog, Reimbursement, Role, User
from app.domain.rules import CORE_RULES
from app.application.seed import latest_configs
from app.infrastructure.ai import ai, AIUnavailable
from app.shared.common import BusinessError, ROLES, cents, iso, money, utcnow
from app.shared.config import settings
from app.shared.schemas import ExtractedInvoice

request_context = ContextVar("request_context", default="system")


def serialize(model) -> dict:
    """通用序列化剔除敏感字段；金额和时间显式转换，避免 ORM 内部状态泄露。"""
    private = {"password_hash", "token_hash", "file_key"}
    aliases = {"config_json": "config", "params_json": "params", "scope_json": "scope", "snapshot_json": "snapshot",
               "result_json": "result", "details_json": "details", "nodes_json": "nodes", "input_json": "input", "output_json": "output"}
    amounts = {"total_cents": "total_amount", "net_cents": "amount", "tax_cents": "tax_amount", "amount_min_cents": "amount_min", "amount_max_cents": "amount_max"}
    data = {}
    for column in inspect(type(model)).columns:
        key, value = column.key, getattr(model, column.key)
        if key in private:
            continue
        if key in amounts:
            data[amounts[key]] = money(value)
        else:
            data[aliases.get(key, key)] = iso(value) if isinstance(value, datetime) else value
    return data


async def get(session, model, identifier):
    row = await session.get(model, identifier)
    if not row:
        raise BusinessError("资源不存在", 404, "NOT_FOUND")
    return row


def require(actor: dict, minimum: str):
    """菜单隐藏不是授权；所有敏感动作在后端重新核对角色层级。"""
    if ROLES.get(actor["role"], 0) < ROLES[minimum]:
        raise BusinessError("当前角色没有此操作权限", 403, "FORBIDDEN")


def readable(actor: dict, reimbursement: Reimbursement) -> bool:
    return actor["id"] == reimbursement.user_id or actor["role"] in {"auditor", "manager", "admin"} or actor["role"] == "supervisor" and actor["department_id"] == reimbursement.department_id


def guard(actor: dict, reimbursement: Reimbursement, edit: bool = False):
    if not readable(actor, reimbursement) or edit and actor["id"] != reimbursement.user_id:
        raise BusinessError("无权操作该报销单", 403, "OWNER_REQUIRED")
    if edit and reimbursement.status not in {"draft", "rejected"}:
        raise BusinessError("只有草稿或驳回单据可以修改", 409, "NOT_EDITABLE")


def revision_guard(row, revision: int):
    if row.revision != revision:
        raise BusinessError("单据已变更，请刷新后重试", 409, "REVISION_CONFLICT")


def log(session, actor: dict | None, action: str, resource: str, identifier: int | None = None,
        before: dict | None = None, after: dict | None = None, trace: str | None = None):
    """只传业务快照，禁止把完整登录请求或密钥交给日志函数。"""
    session.add(OperationLog(actor_id=actor["id"] if actor else None, action=action, resource_type=resource,
        resource_id=identifier, request_id=request_context.get(), trace_id=trace or request_context.get(),
        before_json=before or {}, after_json=after or {}))


async def alert(session, key: str, kind: str, message: str, severity: str = "warning", audit_id=None, reimbursement_id=None):
    """在同一个短写事务中查重；超时扫描不会重复创建相同记录。"""
    row = await session.scalar(select(Alert).where(Alert.dedup_key == key))
    if not row:
        session.add(Alert(dedup_key=key, type=kind, severity=severity, message=message,
                          audit_id=audit_id, reimbursement_id=reimbursement_id))


def normalized_fields(fields: dict) -> dict:
    """用同一发票契约验证用户补录与模型输出，但保留各自的来源标识。"""
    item = ExtractedInvoice.model_validate(fields).model_dump(mode="json")
    for name in ("amount", "tax_amount"):
        if item.get(name) is not None:
            item[name] = money(cents(item[name]))
    if item.get("tax_rate") is not None:
        rate = Decimal(str(item["tax_rate"]))
        if not rate.is_finite() or not 0 <= rate <= 1:
            raise BusinessError("税率必须为0到1，例如13%填写0.13")
    return item


def validate_agent(node: str, config: dict, enabled: bool):
    """固定主干不能被用户配置变成空流程；限制参数规模避免无限重试。"""
    allowed = {"extract": {"max_attempts", "timeout", "concurrency"}, "rule_check": set(),
               "risk_detect": {"timeout", "window_days", "amount_high", "amount_medium", "frequency_limit", "dept_ratio_limit"},
               "make_decision": set(), "feedback": set()}
    if node not in allowed or set(config) - allowed[node]:
        raise BusinessError("包含不支持的 Agent 参数", code="CONFIG_INVALID")
    if node in {"rule_check", "make_decision"} and not enabled:
        raise BusinessError("核心规则和确定性路由不能停用", code="CORE_REQUIRED")
    ranges = {"max_attempts": (1, 3), "concurrency": (1, 10), "timeout": (1, 120),
              "window_days": (1, 365), "frequency_limit": (1, 1000), "dept_ratio_limit": (1, 100)}
    for key, (low, high) in ranges.items():
        if key in config and (isinstance(config[key], bool) or not isinstance(config[key], (int, float)) or not math.isfinite(config[key]) or not low <= config[key] <= high):
            raise BusinessError(f"{key} 必须处于 {low} 到 {high}")
    for key in ("max_attempts", "concurrency", "window_days", "frequency_limit"):
        if key in config and not isinstance(config[key], int):
            raise BusinessError(f"{key} 必须为整数")
    if node == "risk_detect" and cents(config.get("amount_high", "50000")) <= cents(config.get("amount_medium", "20000")):
        raise BusinessError("高额阈值必须大于中额阈值")


def validate_rule(code: str, params: dict, scope: dict, severity: str, enabled: bool):
    """只允许参数化内置规则，核心范围和严重程度不能降低成自动放行。"""
    allowed = {"invoice_date_valid": {"max_age_days"}, "invoice_amount_valid": {"large_amount"},
        "invoice_header_valid": {"company_name"}, "invoice_no_duplicate": set(), "tax_calculation_valid": {"tolerance"},
        "tax_number_valid": {"lengths"}, "policy_compliance": {"required"}}
    if code not in allowed or set(params) - allowed[code] or set(scope) - {"types", "departments"}:
        raise BusinessError("规则参数或范围包含不支持字段")
    if code in CORE_RULES and (not enabled or scope or severity != ("critical" if code == "invoice_no_duplicate" else "error")):
        raise BusinessError("核心校验不能停用、降低严重程度或限定范围", code="CORE_REQUIRED")
    if "max_age_days" in params and (type(params["max_age_days"]) is not int or not 1 <= params["max_age_days"] <= 3650):
        raise BusinessError("有效期必须是1到3650天")
    for key in ("large_amount", "tolerance"):
        if key in params:
            value = cents(params[key], key == "tolerance")
            if key == "tolerance" and value > 100:
                raise BusinessError("税额容差不能超过1元")
    if "company_name" in params and (not isinstance(params["company_name"], str) or len(params["company_name"]) > 200):
        raise BusinessError("公司全称必须是最多200字符的文本")
    if "required" in params and type(params["required"]) is not bool:
        raise BusinessError("required 必须是布尔值")
    if "lengths" in params and (not isinstance(params["lengths"], list) or not params["lengths"] or any(type(x) is not int or not 1 <= x <= 30 for x in params["lengths"])):
        raise BusinessError("税号长度必须为非空整数数组，每个长度1到30")
    for key, values in scope.items():
        if not isinstance(values, list) or any(type(x) is not int or x < 1 or key == "types" and x not in {1, 2, 3} for x in values):
            raise BusinessError("适用范围必须为合法整数 ID 数组")


def chunks(text: str, size: int = 512, overlap: int = 64) -> list[str]:
    """先尊重标题/段落，长段才滑动切分，重叠只用在长段边界。"""
    sections = re.split(r"(?m)(?=^#{1,4}\s|^第[一二三四五六七八九十\d]+[章节条款])", text)
    output = []
    for section in sections:
        buffer = ""
        for paragraph in re.split(r"\n\s*\n", section):
            paragraph = paragraph.strip()
            if not paragraph:
                continue
            if buffer and len(buffer) + len(paragraph) + 1 > size:
                output.append(buffer)
                buffer = ""
            if len(paragraph) > size:
                for start in range(0, len(paragraph), size - overlap):
                    part = paragraph[start:start + size]
                    if part:
                        output.append(part)
                    if start + size >= len(paragraph):
                        break
            else:
                buffer = (buffer + "\n" + paragraph).strip()
        if buffer:
            output.append(buffer)
    return output


async def replace_chunks(session, document):
    """正文更新时先让旧向量失效；旧审核已冻结文本，不受该更新影响。"""
    from sqlalchemy import delete
    await session.execute(delete(KnowledgeChunk).where(KnowledgeChunk.document_id == document.id))
    for position, part in enumerate(chunks(document.content)):
        session.add(KnowledgeChunk(document_id=document.id, document_version=document.version, position=position, text=part))
    document.index_status, document.embedding_model = "not_indexed", None


async def policy_snapshot(session) -> list[dict]:
    documents = (await session.scalars(select(KnowledgeDocument).where(KnowledgeDocument.active.is_(True)))).all()
    result = []
    for document in documents:
        blocks = (await session.scalars(select(KnowledgeChunk).where(KnowledgeChunk.document_id == document.id, KnowledgeChunk.document_version == document.version))).all()
        for block in blocks:
            result.append({"chunk_id": block.id, "document_id": document.id, "title": document.title,
                "version": document.version, "category": document.category, "text": block.text,
                "vector": block.vector_json, "model": document.embedding_model, "dimension": block.dimension})
    return result


async def search_policies(query: str, blocks: list[dict], mode: str, top_k: int, category: str | None = None) -> list[dict]:
    """检索模式显式区分；没有真实向量时语义请求失败，不能偷偷换关键词装成功。"""
    candidates = [item for item in blocks if not category or item["category"] == category]
    ranked = []
    if mode == "semantic":
        candidates = [item for item in candidates if item["vector"] and item["model"] == settings.embedding_model and item["dimension"] == settings.embedding_dimension]
        if not candidates:
            raise AIUnavailable("没有与当前模型一致的有效政策向量")
        query_vector = (await ai.embeddings([query]))[0]
        query_norm = math.sqrt(sum(x * x for x in query_vector))
        for item in candidates:
            vector = item["vector"]
            score = sum(a * b for a, b in zip(vector, query_vector)) / (math.sqrt(sum(x * x for x in vector)) * query_norm)
            ranked.append({**item, "score": score})
    else:
        # 中文关键词以单字/相邻双字命中，明确是关键词相关度，不称作语义概率。
        terms = set(re.findall(r"[A-Za-z0-9]+|[\u4e00-\u9fff]", query.lower()))
        terms.update(query[i:i + 2] for i in range(len(query) - 1) if all("\u4e00" <= c <= "\u9fff" for c in query[i:i + 2]))
        for item in candidates:
            score = sum(term in item["text"].lower() for term in terms) / max(len(terms), 1)
            if score > 0:
                ranked.append({**item, "score": score})
    ranked.sort(key=lambda item: item["score"], reverse=True)
    ranked = ranked[:max(top_k, 10)]
    if mode == "semantic":
        ranked = await ai.rerank(query, ranked)
    return [{key: value for key, value in item.items() if key not in {"vector", "model", "dimension"}} for item in ranked[:top_k]]


async def audit_snapshot(session, reimbursement) -> dict:
    """事务内冻结输入与配置；等待模型时不持有写锁，回写前再比较 revision。"""
    invoices = (await session.scalars(select(Invoice).where(Invoice.reimbursement_id == reimbursement.id, Invoice.active.is_(True)))).all()
    rules = await latest_configs(session, AuditRule, "code")
    agents = await latest_configs(session, AgentConfig, "node_name")
    others = (await session.scalars(select(Reimbursement).where(Reimbursement.id != reimbursement.id,
        Reimbursement.status.in_(["pending_audit", "in_approval", "approved"])))).all()
    ids = [item.id for item in others]
    history = (await session.scalars(select(Invoice).where(Invoice.reimbursement_id.in_(ids), Invoice.active.is_(True)))).all() if ids else []
    config = agents["risk_detect"].config_json
    cutoff = utcnow() - timedelta(days=config.get("window_days", 30))
    submission_logs = (await session.scalars(select(OperationLog).where(OperationLog.action == "reimbursement.submitted", OperationLog.actor_id == reimbursement.user_id, OperationLog.created_at >= cutoff))).all()
    # 当前提交日志尚未写入，因此精确排除当前轮次，旧轮重提仍计入真实频率。
    frequency = len(submission_logs)
    department_amounts = [item.total_cents for item in others if item.department_id == reimbursement.department_id]
    average = str(Decimal(sum(department_amounts)) / len(department_amounts)) if department_amounts else None
    return {"reimbursement": serialize(reimbursement), "invoice_files": [{**serialize(item), "file_key": item.file_key} for item in invoices],
        "rules": [serialize(item) for item in rules.values()], "agents": {key: serialize(value) for key, value in agents.items()},
        "policies": await policy_snapshot(session), "context": {"existing_hashes": [item.file_hash for item in history],
            "same_hash_count": dict(Counter(item.file_hash for item in invoices)), "recent_submissions": frequency,
            "dept_average_cents": average, "statistic_at": iso(utcnow()), "today": datetime.now(ZoneInfo(settings.business_timezone)).date().isoformat(),
            "type": reimbursement.type, "department_id": reimbursement.department_id},
        "models": {"llm": settings.llm_model, "vlm": settings.vlm_model, "embedding": settings.embedding_model}}