pipeline-llm/pipeline_llm/accounting.py

"""pipeline_llm.accounting — 异步出账循环（owner 模型三方账）。

扫 llm_usage 中 accounting_status='created' 且 status='SUCCEEDED' 的用量流水，
逐条映射模型→产品，调产品层 product_accounting_generic 记账
（产品层内部按 is_self_use 分流：自用只记 PAY* 采购成本；跨机构另记 PAY 客户应付，
折扣精确到产品），成功置 'accounted'，失败置 'failed'（三态：created/accounted/failed）。
2026-09-06 用户定夺：status 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING；
只有 SUCCEEDED 的行才写 accounting_status（其余 NULL），出账循环自然只拾取 SUCCEEDED。

与账号/存储的同步记账分工明确（2026-09 用户定夺）：
  - 账号/存储：purchase_realtime 同步记账
  - 模型（本模块）：调用时只记成本侧 + created 流水，出账走本循环（异步）

单循环守卫（多进程部署时防重复出账）：Redis SETNX llm_acc:lock（db4，与限流同库）。
"""

import asyncio
import json
import logging
import time
from datetime import datetime

from ahserver.serverenv import ServerEnv

from .gateway import _get_db, _redis

logger = logging.getLogger("pipeline_llm.accounting")

_SCAN_INTERVAL = 60      # 扫描间隔（秒）
_BATCH = 200             # 每批处理条数
_LOCK_KEY = 'llm_acc:lock'
_LOCK_TTL = 90           # 锁 TTL（略大于单批预算耗时）

_trace_cleanup_day = ''  # trace 过期清理已跑日期（进程内每日一次）


async def _get_pending(sor):
    recs = await sor.sqlExe(
        "SELECT id, org_id, user_id, model_id, cost, ppid, usages, "
        "created_at "
        "FROM llm_usage WHERE accounting_status='created' AND status='SUCCEEDED' "
        "ORDER BY created_at LIMIT %d" % _BATCH, {})
    await sor.sqlExe("COMMIT", {})
    return recs or []


async def _find_product_id(sor, model_id):
    """llm_model.id → product.id（产品层导入时按 resource_ref_id 建立）。"""
    recs = await sor.sqlExe(
        "SELECT id FROM product WHERE resource_ref_id=${r}$ AND status='1' LIMIT 1",
        {"r": model_id})
    await sor.sqlExe("COMMIT", {})
    return (getattr(recs[0], 'id', '') or '') if recs else ''


async def _settle_one(sor, row):
    """单条出账。成功置 accounted；失败置 failed 并记录原因（不静默重试死循环）。"""
    env = ServerEnv()
    model_id = getattr(row, 'model_id', '') or ''
    product_id = await _find_product_id(sor, model_id)
    if not product_id:
        await sor.U('llm_usage', {
            'id': row.id, 'accounting_status': 'failed',
            'note': '出账失败：模型(%s)未映射到产品，请先在产品管理导入产线模型' % model_id,
        })
        await sor.sqlExe("COMMIT", {})
        logger.warning("pipeline_llm.accounting: model %s 无产品映射，流水 %s 置 failed",
                       model_id, row.id)
        return False
    # 计费唯一事实源 = usages JSON 全量原文（2026-09-10 用户定夺，req/resp_tokens
    # 冗余列已删 m0021）：token 计数（prompt_tokens/completion_tokens/
    # prompt_tokens_details）与按量因子（时长/分辨率/张数）都在其中。
    usage_data: dict = {}
    usages_raw = getattr(row, 'usages', '') or ''
    if usages_raw:
        try:
            extra = json.loads(usages_raw)
            if isinstance(extra, dict):
                usage_data.update(extra)
        except Exception:
            # usages 是计费事实源，解析失败必须显式记账失败（写明原因），
            # 禁止静默空 dict 兜底成 0 元账（历史上 [:2000] 截断即此病根）
            await sor.U('llm_usage', {
                'id': row.id, 'accounting_status': 'failed',
                'note': '出账失败：usages 不是合法 JSON（长度%d），无法计费——'
                        '请核查流水写入方是否截断' % len(usages_raw),
            })
            await sor.sqlExe("COMMIT", {})
            logger.warning("pipeline_llm.accounting: 流水 %s usages 非法 JSON，置 failed", row.id)
            return False
    if not usage_data:
        await sor.U('llm_usage', {
            'id': row.id, 'accounting_status': 'failed',
            'note': '出账失败：usages 为空，无计费事实源——请核查上游 usage 返回',
        })
        await sor.sqlExe("COMMIT", {})
        logger.warning("pipeline_llm.accounting: 流水 %s usages 为空，置 failed", row.id)
        return False
    # derived 依赖归一（2026-09-07 根治）：token 定价的 cached_tokens/uncache_tokens
    # 靠 YAML derived 变量引用 prompt_tokens_details.cached_tokens 计算——该键缺失时
    # DictObject 属性链抛 AttributeError，引擎兜底成 0，输入 token 永不计费
    # （实测 qwen3.8-max uncache_tokens=0 记账失败根因之一）。缺失补 0 结构，
    # derived 正常求值：无缓存时 uncache=prompt_tokens、cached=0。
    ptd = usage_data.get('prompt_tokens_details')
    if not isinstance(ptd, dict):
        usage_data['prompt_tokens_details'] = {'cached_tokens': 0}
    elif ptd.get('cached_tokens') is None:
        ptd['cached_tokens'] = 0
    # 忙闲时计价维度（2026-09-07）：价目表存在忙时(8点-22点)/闲时(22点-次日8点)
    # 两档定价（deepseek-v4-pro-0813），按调用时刻 llm_usage.created_at 的小时数注入
    # hour——YAML 用 hour 区间 filters 选档。多余键对不用它的定价方案无影响
    # （引擎只严格检查定价项里出现的维度键）。
    # 2026-09-10：注入浮点小时（时+分/60+秒/3600），支持 pricing 引擎 peak_times
    # 时段串（如 "9:00 =~= 12:00"）的精确时刻语义；对存量整数小时区间 YAML
    # （8 =~ 22 等）边界语义不变（整点值相同）。
    created_at = getattr(row, 'created_at', None)
    if created_at is not None:
        try:
            usage_data['hour'] = (int(created_at.hour)
                                  + int(created_at.minute) / 60.0
                                  + int(created_at.second) / 3600.0)
        except Exception:
            pass
    fn = getattr(env, 'product_accounting_generic', None)
    if fn is None:
        # 产品模块未加载 → 保留 created 下轮再试
        return False
    result = await fn(
        product_id=product_id,
        usage_data=usage_data,
        userorgid=getattr(row, 'org_id', '') or '',
        userid=getattr(row, 'user_id', '') or '',
    )
    if not result or not result.get('success', result.get('orderid')):
        msg = str((result or {}).get('message', result))[:180]
        await sor.U('llm_usage', {
            'id': row.id, 'accounting_status': 'failed',
            'note': '出账失败：%s' % msg,
        })
        await sor.sqlExe("COMMIT", {})
        logger.error("pipeline_llm.accounting: 流水 %s 出账失败: %s", row.id, msg)
        return False
    # 回填客户应付 + 置已出账
    await sor.sqlExe(
        "UPDATE llm_usage SET charge=${c}$, accounting_status='accounted', "
        "note=CONCAT(COALESCE(note,''),'|accounted:',${o}$) WHERE id=${i}$",
        {"c": float((result or {}).get('customer_amount', 0) or 0),
         "o": str((result or {}).get('orderid', ''))[:40],
         "i": row.id})
    await sor.sqlExe("COMMIT", {})
    return True


async def _run_one_pass():
    """一轮扫描-出账。返回处理条数。"""
    db, dbname = _get_db()
    async with db.sqlorContext(dbname) as sor:
        rows = await _get_pending(sor)
    if not rows:
        return 0
    ok_n = 0
    for row in rows:
        try:
            db, dbname = _get_db()
            async with db.sqlorContext(dbname) as sor:
                if await _settle_one(sor, row):
                    ok_n += 1
        except Exception as e:
            logger.error("pipeline_llm.accounting: 流水 %s 出账异常: %s",
                         getattr(row, 'id', ''), e)
            try:
                db, dbname = _get_db()
                async with db.sqlorContext(dbname) as sor:
                    await sor.U('llm_usage', {
                        'id': row.id, 'accounting_status': 'failed',
                        'note': '出账异常：%s' % str(e)[:180],
                    })
                    await sor.sqlExe("COMMIT", {})
            except Exception:
                pass
    return ok_n


async def _maybe_cleanup_traces():
    """trace 保留期清理：每日最多跑一次（按日期字符串判定，循环间隔 60s 足够）。

    trace.py 铁律 3 的落地点——cleanup_expired 此前无人调用（文档承诺
    "记账 worker 循环每日调一次"但循环里没接，2026-09-11 补齐）。
    永不抛出：清理失败不反噬出账主循环。
    """
    global _trace_cleanup_day
    today = datetime.now().strftime('%Y%m%d')
    if today == _trace_cleanup_day:
        return
    _trace_cleanup_day = today
    try:
        from .trace import cleanup_expired
        await cleanup_expired()
    except Exception as e:
        logger.warning("pipeline_llm.accounting: trace 过期清理失败（明日重试）: %s", e)


async def llm_usage_accounting_loop():
    """后台循环：抢分布式锁 → 出账一批 → 睡。永不抛出。

    只由独立记账进程（app/pipeline_accounting_worker.py）运行，
    不挂进 web 服务器的事件循环（同步记账在 web，异步记账在独立进程）。
    """
    logger.info("[pipeline_llm] 异步出账循环已启动（间隔 %ds）", _SCAN_INTERVAL)
    while True:
        try:
            # trace 过期清理（每日一次；放抢锁前——锁只护出账防重复记账，
            # 清理是幂等操作（目录整删+按 created_at 删行），多进程同日跑无害）
            await _maybe_cleanup_traces()
            r = _redis()
            if r is not None:
                # 抢锁（多进程只跑一个）；Redis 故障则直接跑（与限流 fail-open 一致）
                got = False
                try:
                    got = bool(r.set(_LOCK_KEY, str(time.time()), nx=True, ex=_LOCK_TTL))
                except Exception:
                    got = True
                if not got:
                    await asyncio.sleep(_SCAN_INTERVAL)
                    continue
            await _run_one_pass()
        except Exception as e:
            logger.error("pipeline_llm.accounting: 循环异常（继续）: %s", e)
        await asyncio.sleep(_SCAN_INTERVAL)