"""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, account_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 _deduct_wallet(sor, row, supplier_cost): """产品记账完成后扣减调用账号的钱包（2026-09-11 用户定夺）。 钱包 = llm_account.balance。1 个供应商财务账户(accounting.account) : n 个 llm_account 钱包——一个供应商可配多账号(钱包)，扣减主体按流水 account_id 定位「**调用模型的那个钱包**」，扣减金额 = accounting 对该供应商的记账额 （product_accounting_generic 返回的 supplier_cost，即 PAY* 分录的采购成本）。 幂等守卫：流水 cost 列在出账成功时回填 supplier_cost（govern_settle 创建时 cost=0）。cost≠0 ⟺ 本行已扣过钱包——accounted 行被手动重置回 created 重跑时 跳过扣减，防重复扣。（failed→created 追回流程安全：failed 行在扣减前就 return False，从未回填 cost，重跑正常扣。） 允许扣成负值：调用→出账有 ~60s 窗口且门禁阈值(缺省5元)是事前闸，并发超扣 是真实成本，宁可透支为负（待办 provider 派生待办催充值），不可丢失钱流或把 已记账流水打回 failed（账务不可逆）。扣减失败只 warning 不回滚客户侧记账。 """ try: prior_cost = float(getattr(row, 'cost', 0) or 0) except (TypeError, ValueError): prior_cost = 0.0 if prior_cost != 0: logger.info("pipeline_llm.accounting: 流水 %s 已回填 cost=%.6f（钱包已扣），" "幂等跳过重复扣减", getattr(row, 'id', ''), prior_cost) return account_id = getattr(row, 'account_id', '') or '' if not account_id: return try: amount = float(supplier_cost or 0) except (TypeError, ValueError): amount = 0.0 if amount <= 0: return try: # 只发 UPDATE 不单独 COMMIT——与调用方回填 cost/accounted 的 UPDATE 同一 # 事务提交（原子）：钱包扣减与 cost 回填要么都落要么都不落，杜绝 # "已扣钱但 cost 未回填→重跑重复扣" 的窗口。 await sor.sqlExe( "UPDATE llm_account SET balance=balance-${c}$ WHERE id=${i}$", {"c": round(amount, 6), "i": account_id}) except Exception as e: logger.warning("pipeline_llm.accounting: 账号钱包扣减失败（不回滚记账，对账修正）" "account=%s cost=%.6f: %s", account_id, amount, str(e)[:160]) 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 # 产品记账完成（2026-09-11 用户定夺）：记账成功之后扣减调用账号的钱包—— # supplier_cost 是定价引擎算出的真实进货成本，扣 llm_account.balance（流水 # account_id 定位"调用模型的钱包"）；同时把 cost 回填进流水（对账可查每笔 # 钱包扣了多少）。扣减失败只 warning，不回滚客户侧账务。 supplier_cost = float((result or {}).get('supplier_cost', 0) or 0) await _deduct_wallet(sor, row, supplier_cost) # 回填客户应付 + 成本侧金额 + 置已出账 await sor.sqlExe( "UPDATE llm_usage SET charge=${c}$, cost=${k}$, accounting_status='accounted', " "note=CONCAT(COALESCE(note,''),'|accounted:',${o}$) WHERE id=${i}$", {"c": float((result or {}).get('customer_amount', 0) or 0), "k": round(supplier_cost, 6), "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)