host
c33bfd240e
- model_breakdown inferred params as a 1-tuple from the first branch, so params + (month,) was tuple[Any] vs tuple[Any, str]. Annotate params as bare tuple (tuple[Any, ...]) before branching. - Annotation only: the () initializer is never used (if/elif/else always overwrites it), runtime semantics unchanged. - mypy cloud/: 0 errors in 94 files (was 1 error); ruff clean; cost tests 39 pass. (cherry picked from commit a9fd0692361e8169554413b725c81135be2c5525)
630 lines
27 KiB
Python
630 lines
27 KiB
Python
"""Cost estimation and cost recording.
|
||
|
||
Zero third-party dependencies; standard library only.
|
||
|
||
Actual token counts come from the model/provider response when
|
||
available: host.py collects the app-server's thread/tokenUsage/updated
|
||
events during each turn (one per upstream completion, summed over the
|
||
turn window) and the worker records them with ``source="actual"``.
|
||
When no usage events arrive (old server, turn produced no completion),
|
||
the worker falls back to ``estimate_tokens`` with
|
||
``source="estimated"``; the two are stored separately so estimates are
|
||
never mixed into actual totals.
|
||
|
||
USD is tokens times a price table, not the relay's bill: the defaults
|
||
are OpenAI list prices (see ``_DEFAULT_MODEL_PRICES``), and the relay
|
||
may charge differently. Override per model with
|
||
``CLOUD_MODEL_PRICES_JSON`` or ``CLOUD_MODEL_PRICES_FILE``; models
|
||
with no price record ``usd_micros=None`` and reports count their
|
||
tokens in ``unpriced_actual_tokens`` instead of pricing them at zero.
|
||
|
||
cache_write_input_tokens: EXPLICITLY NOT BILLED HERE (P1 decision,
|
||
not an oversight). host.py collects ``cacheWriteInputTokens`` from
|
||
upstream tokenUsage events and sandbox.py carries it in usage dicts
|
||
for observability, but it is NOT added to ``total_tokens`` (upstream
|
||
total is input + output), NOT stored in ``cost_records`` (``record``
|
||
takes input/output/cached only), NOT priced by ``price_usd_micros``,
|
||
and NOT counted toward the unpriced-token cap (worker counts input +
|
||
output from ``source="actual"`` rows only). Rationale: no price-table
|
||
entry and no relay bill evidence treats cache creation as a
|
||
separately billed token stream for the OpenAI-style relay models in
|
||
use; cache_write here is prompt-cache bookkeeping detail, not a
|
||
second charge on top of input. If a future provider bills cache
|
||
creation separately, the price table must gain an explicit
|
||
cache_write rate FIRST -- only then may ``record``/pricing count it.
|
||
Until then, counting it would double-charge input tokens that are
|
||
already priced (or inflate the unpriced cap with non-spend).
|
||
|
||
Unpriced cap scope (P1): ``unpriced_actual_tokens`` and the worker's
|
||
cap query count ``source="actual"`` rows only. Estimated rows
|
||
(``source="estimated"``, character-count fallback when no usage
|
||
events arrive) are never real spend and must not trip the cap.
|
||
|
||
Dead end, documented so nobody retries it: the app-server's
|
||
account/usage/read method requires ChatGPT account authentication and
|
||
fails on a relay API key (2026-10-26 probe: ``codex account
|
||
authentication required to read token usage``). Account-level usage
|
||
is therefore not a source here; per-task actuals come from turn
|
||
events only.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
from cloud.db import connect as _db_connect
|
||
import time
|
||
from datetime import datetime, timezone
|
||
|
||
|
||
# OpenAI list prices, USD per 1M tokens, as (input, cached_input,
|
||
# output). gpt-6.1-sol $2.00/$0.10/$10.00 and gpt-6-sol
|
||
# $2.00/$0.20/$10.00, as reported at the 2026-09-30 launch.
|
||
# Again: list prices, not the relay's bill; override below.
|
||
_DEFAULT_MODEL_PRICES = {
|
||
"gpt-6.1-sol": (2.00, 0.10, 10.00),
|
||
"gpt-6-sol": (2.00, 0.20, 10.00),
|
||
}
|
||
|
||
|
||
def _coerce_rates(rates):
|
||
"""Accept [input, cached, output] or {input, cached_input, output}."""
|
||
try:
|
||
if isinstance(rates, dict):
|
||
return (float(rates["input"]), float(rates["cached_input"]),
|
||
float(rates["output"]))
|
||
return (float(rates[0]), float(rates[1]), float(rates[2]))
|
||
except (TypeError, ValueError, KeyError, IndexError):
|
||
return None
|
||
|
||
|
||
def model_prices():
|
||
"""Effective price table: defaults merged with env overrides.
|
||
|
||
``CLOUD_MODEL_PRICES_JSON`` (a JSON object) and
|
||
``CLOUD_MODEL_PRICES_FILE`` (path to a JSON file) both merge over
|
||
the defaults per model; unparsable content is ignored, never fatal.
|
||
"""
|
||
prices = dict(_DEFAULT_MODEL_PRICES)
|
||
raw = os.environ.get("CLOUD_MODEL_PRICES_JSON") or ""
|
||
if raw:
|
||
try:
|
||
override = json.loads(raw)
|
||
except ValueError:
|
||
override = None
|
||
if isinstance(override, dict):
|
||
for model, rates in override.items():
|
||
coerced = _coerce_rates(rates)
|
||
if coerced is not None:
|
||
prices[model] = coerced
|
||
path = os.environ.get("CLOUD_MODEL_PRICES_FILE") or ""
|
||
if path:
|
||
try:
|
||
with open(path) as f:
|
||
override = json.load(f)
|
||
except (OSError, ValueError):
|
||
override = None
|
||
if isinstance(override, dict):
|
||
for model, rates in override.items():
|
||
coerced = _coerce_rates(rates)
|
||
if coerced is not None:
|
||
prices[model] = coerced
|
||
return prices
|
||
|
||
|
||
def price_usd_micros(model, input_tokens, cached_input_tokens,
|
||
output_tokens):
|
||
"""USD micros for provider-measured tokens at the price table.
|
||
|
||
cached is a subset of input (prompt-cache reads): non-cached input
|
||
pays the input rate, cached pays the cached rate, output pays the
|
||
output rate. Reasoning tokens are already inside output_tokens
|
||
(provider total = input + output), never added twice. Returns None
|
||
when the model has no price entry (caller records usd as unknown).
|
||
cache_write_input_tokens is deliberately NOT a parameter here:
|
||
it is not billed separately (see module docstring); callers must
|
||
not add it to input_tokens to “count it anyway”.
|
||
"""
|
||
rates = model_prices().get(model)
|
||
if rates is None:
|
||
return None
|
||
input_rate, cached_rate, output_rate = rates
|
||
total_in = max(0, int(input_tokens or 0))
|
||
cached = max(0, min(int(cached_input_tokens or 0), total_in))
|
||
non_cached = total_in - cached
|
||
usd = (non_cached * input_rate + cached * cached_rate
|
||
+ max(0, int(output_tokens or 0)) * output_rate) / 1000000
|
||
return int(round(usd * 1000000))
|
||
|
||
|
||
# 双列文案：逐字复用 PR #212（feat-cost-quota，5b30c61）的
|
||
# BILLING_NOTE，本轨不拥有成本口径、不另造一套说法。
|
||
BILLING_NOTE = (
|
||
"价格表代理价（经 CLOUD_MODEL_PRICES_JSON / CLOUD_MODEL_PRICES_FILE"
|
||
" 覆盖层），非中转实际账单；未定价 token 不计入美元、不等于免费。"
|
||
)
|
||
|
||
|
||
def quota_cutoff_reason(quota_usd_micros, used_usd_micros,
|
||
unpriced_used_tokens, month, unpriced_cap=None):
|
||
"""配额硬切断的唯一判定函数（worker 执行侧与单测共用）。
|
||
|
||
返回 None 表示未超限、可继续；返回 quota_exceeded 文本表示
|
||
必须立即停：调用方不再开新回合、不再派生花钱路径，任务以
|
||
interrupted + 该原因收尾（原因写进 error_text 并记审计），
|
||
不是只告警。quota 为 None（管理员显式不限额）时不挡；USD
|
||
配额与无价 token 上限是两道独立的门，任一超限即停。
|
||
文本格式与 worker 旧分支逐字一致，旧单测按此前缀断言。
|
||
"""
|
||
if quota_usd_micros is None:
|
||
return None
|
||
if int(used_usd_micros or 0) >= int(quota_usd_micros):
|
||
return ("quota_exceeded: 本月配额已用完（已用 %d / "
|
||
"配额 %d 微美元，%s）"
|
||
% (int(used_usd_micros or 0), int(quota_usd_micros),
|
||
month))
|
||
cap = unpriced_token_cap() if unpriced_cap is None \
|
||
else int(unpriced_cap)
|
||
if int(unpriced_used_tokens or 0) >= cap:
|
||
return ("quota_exceeded: 本月无价模型 token 已用完"
|
||
"（已用 %d / 上限 %d，%s）"
|
||
% (int(unpriced_used_tokens or 0), cap, month))
|
||
return None
|
||
|
||
|
||
def dual_column_view(report):
|
||
"""双列视图：在已有 _report 结果上加定价列别名与账单说明。
|
||
|
||
不改 _report（旧单测按其键集断言）与旧键：usd_micros 仍是
|
||
有价行之和，unpriced_actual_tokens 仍是未定价 actual token；
|
||
本函数只加 priced_usd_micros（与 usd_micros 同值）与
|
||
billing_note。未定价模型分列（by_model / unpriced_models）
|
||
与按创建者口径是 PR #212 的范围，本轨不重复实现——#212
|
||
合并后本函数由其 dual_column_* 取代，端点改走 dual_column_*
|
||
（同名同值键，切换无语义差）；#212 未合并前，本视图是
|
||
任务/项目成本端点的双列最小实现。
|
||
"""
|
||
view = dict(report)
|
||
view["priced_usd_micros"] = view["usd_micros"]
|
||
view["billing_note"] = BILLING_NOTE
|
||
return view
|
||
|
||
|
||
def unpriced_token_cap():
|
||
"""Monthly unpriced-token ceiling for users who have a quota row.
|
||
|
||
Models with no price entry (today: the default gpt-5.6-terra and
|
||
the other relay models) record usd as unknown, so a USD quota
|
||
alone never binds them -- without this cap a quota user could
|
||
burn unlimited unpriced tokens. The cap is a volume guardrail,
|
||
NOT a price: cost reports still show those tokens as unpriced,
|
||
never converted to fake dollars. Only users with a quota row
|
||
are capped (no row still means unlimited, as before). Env
|
||
``CLOUD_QUOTA_UNPRICED_TOKEN_CAP`` overrides the default
|
||
20_000_000 tokens/month; unparsable values fall back to it.
|
||
"""
|
||
try:
|
||
return max(0, int(os.environ.get(
|
||
"CLOUD_QUOTA_UNPRICED_TOKEN_CAP", "20000000")))
|
||
except (TypeError, ValueError):
|
||
return 20000000
|
||
|
||
|
||
_USAGE_SUM_KEYS = ("input_tokens", "cached_input_tokens",
|
||
"cache_write_input_tokens", "output_tokens",
|
||
"reasoning_output_tokens", "total_tokens", "completions")
|
||
|
||
|
||
def sum_usage(a, b):
|
||
"""Add two usage dicts (host.py shape); either side may be None."""
|
||
if not a:
|
||
return dict(b) if b else None
|
||
if not b:
|
||
return dict(a)
|
||
out = dict(a)
|
||
for k in _USAGE_SUM_KEYS:
|
||
if isinstance(b.get(k), int):
|
||
out[k] = out.get(k, 0) + b[k]
|
||
return out
|
||
|
||
|
||
def estimate_tokens(prompt_text, output_text) -> int:
|
||
"""Estimate total tokens as (chars of prompt + output) // 4."""
|
||
prompt_text = prompt_text or ""
|
||
output_text = output_text or ""
|
||
return (len(prompt_text) + len(output_text)) // 4
|
||
|
||
|
||
def fetch_account_usage(usage_client) -> dict:
|
||
"""Fetch account usage through an injected ``usage_client`` callable.
|
||
|
||
This is a call wrapper around the app-server account/usage/read
|
||
concept: the real ``usage_client`` is injected by the control plane
|
||
once the billing route is confirmed available. This function itself
|
||
never runs a model turn and never touches the network.
|
||
|
||
On success it returns ``{"available": True, "data": <raw data>}``.
|
||
If ``usage_client`` raises, or returns None / an empty dict,
|
||
it returns ``{"available": False, "data": None}`` and never raises
|
||
to the caller. On a relay, usage being unreadable is a normal
|
||
outcome (available=False), not a bug.
|
||
"""
|
||
try:
|
||
data = usage_client()
|
||
except Exception:
|
||
return {"available": False, "data": None}
|
||
if data is None:
|
||
return {"available": False, "data": None}
|
||
if isinstance(data, dict) and not data:
|
||
return {"available": False, "data": None}
|
||
return {"available": True, "data": data}
|
||
|
||
|
||
class CostStore:
|
||
"""SQLite store for per-task / per-project / per-day cost records."""
|
||
|
||
_SOURCES = ("actual", "estimated")
|
||
|
||
_COST_DDL = """CREATE TABLE IF NOT EXISTS cost_records (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
task_id TEXT,
|
||
project_id TEXT,
|
||
model TEXT,
|
||
input_tokens INTEGER,
|
||
output_tokens INTEGER,
|
||
cached_input_tokens INTEGER DEFAULT 0,
|
||
total_tokens INTEGER,
|
||
usd_micros INTEGER,
|
||
source TEXT,
|
||
fallback_count INTEGER DEFAULT 0,
|
||
completions INTEGER DEFAULT 0,
|
||
day TEXT,
|
||
created_at REAL
|
||
)"""
|
||
|
||
def _ensure_cached_column(self, conn):
|
||
"""老库补 cached_input_tokens 列（SQLite 与 Postgres 两支：
|
||
|
||
PRAGMA table_info 经 db.py 适配层翻译，缺列时 ALTER 补列；
|
||
并发下两个进程同时补列时第二个 ALTER 报错，直接吞掉）。
|
||
"""
|
||
try:
|
||
cols = {row[1] for row in conn.execute(
|
||
"PRAGMA table_info(cost_records)").fetchall()}
|
||
except Exception:
|
||
return
|
||
if "cached_input_tokens" not in cols:
|
||
try:
|
||
conn.execute(
|
||
"ALTER TABLE cost_records"
|
||
" ADD COLUMN cached_input_tokens INTEGER DEFAULT 0")
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
if "completions" not in cols:
|
||
try:
|
||
conn.execute(
|
||
"ALTER TABLE cost_records"
|
||
" ADD COLUMN completions INTEGER DEFAULT 0")
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
def __init__(self, db_path):
|
||
self.db_path = db_path
|
||
if db_path != ":memory:":
|
||
parent = os.path.dirname(os.path.abspath(db_path))
|
||
os.makedirs(parent, exist_ok=True)
|
||
self._conn = _db_connect(db_path)
|
||
self._conn.execute(self._COST_DDL)
|
||
self._conn.commit()
|
||
self._ensure_cached_column(self._conn)
|
||
|
||
def _new_conn(self):
|
||
"""每次写/读开一个新连接，用完即关。
|
||
|
||
旧实现在 __init__ 里建一条连接并终身共用，单线程没问题；
|
||
worker 并发后多个线程共用同一条 sqlite 连接会直接报
|
||
ProgrammingError（连接是线程绑定的）。其它 store（Worker
|
||
的 _set_status、_audit 等）本来就是每次开新连接，这里
|
||
向它们看齐。_conn 仍保留（建表 + 旧测试直接用它读），
|
||
但 record/_report 不再走它。
|
||
例外：db_path 为 :memory: 时新连接是另一个空库，必须
|
||
复用 _conn（:memory: 只用于单线程单测，并发 worker
|
||
从不用它）。
|
||
"""
|
||
if self.db_path == ":memory:":
|
||
return self._conn
|
||
conn = _db_connect(self.db_path)
|
||
conn.execute(self._COST_DDL)
|
||
conn.commit()
|
||
self._ensure_cached_column(conn)
|
||
return conn
|
||
|
||
def record(self, task_id, project_id, model, input_tokens=0,
|
||
output_tokens=0, cached_input_tokens=0, usd_micros=None,
|
||
source="actual", fallback_count=0, completions=0):
|
||
"""Insert one cost record; returns its row id.
|
||
|
||
``total_tokens`` is computed as input + output, and ``day`` is
|
||
the UTC date (YYYY-MM-DD) of ``created_at``. cached is a
|
||
subset of input (prompt-cache reads), stored separately so
|
||
reports can show why a task was cheap; it is never added to
|
||
total on top of input.
|
||
"""
|
||
if source not in self._SOURCES:
|
||
raise ValueError(
|
||
"source must be one of %r, got %r" % (self._SOURCES, source)
|
||
)
|
||
input_tokens = int(input_tokens or 0)
|
||
output_tokens = int(output_tokens or 0)
|
||
cached_input_tokens = int(cached_input_tokens or 0)
|
||
total_tokens = input_tokens + output_tokens
|
||
created_at = time.time()
|
||
day = datetime.fromtimestamp(created_at, tz=timezone.utc).strftime("%Y-%m-%d")
|
||
conn = self._new_conn()
|
||
try:
|
||
cursor = conn.execute(
|
||
"""INSERT INTO cost_records
|
||
(task_id, project_id, model, input_tokens, output_tokens,
|
||
cached_input_tokens, total_tokens, usd_micros, source,
|
||
fallback_count, completions, day, created_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(task_id, project_id, model, input_tokens, output_tokens,
|
||
cached_input_tokens, total_tokens, usd_micros, source,
|
||
int(fallback_count or 0), int(completions or 0), day,
|
||
created_at),
|
||
)
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
finally:
|
||
# :memory: 时 conn 就是共享的 _conn，绝不能关掉它。
|
||
if conn is not self._conn:
|
||
conn.close()
|
||
|
||
def record_estimated(self, task_id, project_id, model, prompt_text,
|
||
output_text):
|
||
"""Record an estimated cost row; returns its row id.
|
||
|
||
The token split cannot be recovered from a character estimate,
|
||
so ``input_tokens`` holds the whole estimate and
|
||
``output_tokens`` is 0; ``source`` is ``"estimated"`` and
|
||
``usd_micros`` is None.
|
||
"""
|
||
total = estimate_tokens(prompt_text, output_text)
|
||
return self.record(
|
||
task_id, project_id, model,
|
||
input_tokens=total, output_tokens=0,
|
||
usd_micros=None, source="estimated", fallback_count=0,
|
||
)
|
||
|
||
def _report(self, where, params):
|
||
conn = self._new_conn()
|
||
try:
|
||
row = conn.execute(
|
||
"""SELECT
|
||
COALESCE(SUM(total_tokens), 0),
|
||
COALESCE(SUM(usd_micros), 0),
|
||
COUNT(*),
|
||
COALESCE(SUM(fallback_count), 0),
|
||
COALESCE(SUM(CASE WHEN source = 'actual'
|
||
THEN total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(CASE WHEN source = 'estimated'
|
||
THEN total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(cached_input_tokens), 0),
|
||
COALESCE(SUM(CASE WHEN source = 'actual'
|
||
AND usd_micros IS NULL
|
||
THEN total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(completions), 0)
|
||
FROM cost_records WHERE %s""" % where,
|
||
params,
|
||
).fetchone()
|
||
finally:
|
||
if conn is not self._conn:
|
||
conn.close()
|
||
(total_tokens, usd_micros, turns, fallback_count, actual,
|
||
estimated, cached, unpriced, completions) = row
|
||
return {
|
||
"total_tokens": int(total_tokens),
|
||
"usd_micros": int(usd_micros),
|
||
"turns": int(turns),
|
||
"fallback_count": int(fallback_count),
|
||
"actual_tokens": int(actual),
|
||
"estimated_tokens": int(estimated),
|
||
"cached_input_tokens": int(cached),
|
||
# usd 求和只覆盖有价的行；无价 actual 行的 token 在此，
|
||
# 不会被静默算成 0 美元。
|
||
"unpriced_actual_tokens": int(unpriced),
|
||
# LLM 请求数：host.py 每回合汇总的 completions 之和
|
||
# （每个 thread/tokenUsage/updated 算一次上游补全）；
|
||
# turns 只是记账行数，两者不是一回事。
|
||
"llm_requests": int(completions),
|
||
}
|
||
|
||
def report_by_task(self, task_id) -> dict:
|
||
"""Aggregate cost records for one task."""
|
||
return self._report("task_id = ?", (task_id,))
|
||
|
||
def report_by_project(self, project_id) -> dict:
|
||
"""Aggregate cost records for one project."""
|
||
return self._report("project_id = ?", (project_id,))
|
||
|
||
def report_by_day(self, day) -> dict:
|
||
"""Aggregate cost records for one UTC day (YYYY-MM-DD)."""
|
||
return self._report("day = ?", (day,))
|
||
|
||
def _report_joined(self, where, params):
|
||
"""_report 的 JOIN 版：按 tasks.created_by 归属汇总。
|
||
|
||
cost_records 本身没有创建者列，归属只能经 tasks JOIN；
|
||
协调员子任务行无 created_by 时跟父任务走的口径在
|
||
worker 侧，库层这里只按行上的 created_by 计——无属主
|
||
任务的成本不归任何人，不会被静默算进某个创建者。
|
||
"""
|
||
conn = self._new_conn()
|
||
try:
|
||
row = conn.execute(
|
||
"""SELECT
|
||
COALESCE(SUM(c.total_tokens), 0),
|
||
COALESCE(SUM(c.usd_micros), 0),
|
||
COUNT(*),
|
||
COALESCE(SUM(c.fallback_count), 0),
|
||
COALESCE(SUM(CASE WHEN c.source = 'actual'
|
||
THEN c.total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(CASE WHEN c.source = 'estimated'
|
||
THEN c.total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(c.cached_input_tokens), 0),
|
||
COALESCE(SUM(CASE WHEN c.source = 'actual'
|
||
AND c.usd_micros IS NULL
|
||
THEN c.total_tokens ELSE 0 END), 0),
|
||
COALESCE(SUM(c.completions), 0)
|
||
FROM cost_records c JOIN tasks t
|
||
ON c.task_id = t.id WHERE %s""" % where,
|
||
params,
|
||
).fetchone()
|
||
finally:
|
||
if conn is not self._conn:
|
||
conn.close()
|
||
(total_tokens, usd_micros, turns, fallback_count, actual,
|
||
estimated, cached, unpriced, completions) = row
|
||
return {
|
||
"total_tokens": int(total_tokens),
|
||
"usd_micros": int(usd_micros),
|
||
"turns": int(turns),
|
||
"fallback_count": int(fallback_count),
|
||
"actual_tokens": int(actual),
|
||
"estimated_tokens": int(estimated),
|
||
"cached_input_tokens": int(cached),
|
||
"unpriced_actual_tokens": int(unpriced),
|
||
"llm_requests": int(completions),
|
||
}
|
||
|
||
def report_by_creator(self, creator_id, month=None) -> dict:
|
||
"""按任务创建者汇总：当月（UTC，YYYY-MM）或全部。
|
||
|
||
month 为 None 时不过滤月份；month 格式须为 YYYY-MM，
|
||
否则 ValueError（调用方转 400）。返回 _report 同款键
|
||
外加 creator_id 与 month，方便分账报表直接按人对账。
|
||
"""
|
||
if month is not None:
|
||
if not isinstance(month, str) or len(month) != 7 \
|
||
or month[4] != "-":
|
||
raise ValueError("month must be YYYY-MM")
|
||
report = self._report_joined(
|
||
"t.created_by = ? AND c.day LIKE ?",
|
||
(creator_id, month + "%"))
|
||
else:
|
||
report = self._report_joined(
|
||
"t.created_by = ?", (creator_id,))
|
||
report["creator_id"] = creator_id
|
||
report["month"] = month
|
||
return report
|
||
|
||
def model_breakdown(self, task_id=None, project_id=None,
|
||
creator_id=None, month=None) -> list:
|
||
"""按模型分列：恰好指定 task / project / creator 其中一种。
|
||
|
||
每行：model、total_tokens、turns、llm_requests、
|
||
priced_usd_micros（该模型无任何有价行时为 None，不写 0）、
|
||
unpriced_tokens（source='actual' 且 usd IS NULL 的行，
|
||
估算行不计）、priced（是否有任一有价行）。未定价模型
|
||
由 unpriced_tokens > 0 识别，调用方用它单独列出，绝不
|
||
把 priced_usd_micros=None 的模型按 0 美元合并进合计。
|
||
creator 口径经 tasks JOIN，month 同 report_by_creator。
|
||
"""
|
||
scopes = [s for s in (task_id, project_id, creator_id)
|
||
if s is not None]
|
||
if len(scopes) != 1:
|
||
raise ValueError(
|
||
"exactly one of task_id, project_id, creator_id required")
|
||
join = ""
|
||
# 类型标注而已（A6）：首分支会把 params 推断成单元素元组，
|
||
# 导致下面 params + (month,) 报 tuple[Any] vs tuple[Any, str]；
|
||
# 标注成裸 tuple（即 tuple[Any, ...]）后三分支与拼接均合法。
|
||
# 初值 () 永不被使用——if/elif/else 三分支必居其一重写它，
|
||
# 运行时语义不变。
|
||
params: tuple = ()
|
||
if task_id is not None:
|
||
where, params = "c.task_id = ?", (task_id,)
|
||
elif project_id is not None:
|
||
where, params = "c.project_id = ?", (project_id,)
|
||
else:
|
||
join = " JOIN tasks t ON c.task_id = t.id"
|
||
where, params = "t.created_by = ?", (creator_id,)
|
||
if month is not None:
|
||
if not isinstance(month, str) or len(month) != 7 \
|
||
or month[4] != "-":
|
||
raise ValueError("month must be YYYY-MM")
|
||
where += " AND c.day LIKE ?"
|
||
params = params + (month + "%",)
|
||
conn = self._new_conn()
|
||
try:
|
||
rows = conn.execute(
|
||
"""SELECT c.model,
|
||
COALESCE(SUM(c.total_tokens), 0),
|
||
COUNT(*),
|
||
COALESCE(SUM(c.completions), 0),
|
||
SUM(c.usd_micros),
|
||
SUM(CASE WHEN c.usd_micros IS NOT NULL
|
||
THEN 1 ELSE 0 END),
|
||
COALESCE(SUM(CASE WHEN c.source = 'actual'
|
||
AND c.usd_micros IS NULL
|
||
THEN c.total_tokens ELSE 0 END), 0)
|
||
FROM cost_records c%s WHERE %s
|
||
GROUP BY c.model
|
||
ORDER BY SUM(c.total_tokens) DESC, c.model"""
|
||
% (join, where),
|
||
params,
|
||
).fetchall()
|
||
finally:
|
||
if conn is not self._conn:
|
||
conn.close()
|
||
out = []
|
||
for row in rows:
|
||
(model, total, turns, completions, usd_sum, priced_rows,
|
||
unpriced) = row
|
||
out.append({
|
||
"model": model or "unknown",
|
||
"total_tokens": int(total),
|
||
"turns": int(turns),
|
||
"llm_requests": int(completions),
|
||
"priced_usd_micros": (int(usd_sum) if priced_rows else None),
|
||
"unpriced_tokens": int(unpriced),
|
||
"priced": bool(priced_rows),
|
||
})
|
||
return out
|
||
|
||
def _dual_column(self, base, breakdown):
|
||
"""双列报告：在 _report 结果上加第二列与分列明细。
|
||
|
||
_report 本身不动（旧单测按其精确键集断言）；双列是
|
||
新方法返回的扩展：priced_usd_micros 与 usd_micros 同值
|
||
（只计有价行之和），unpriced_models 单独列出未定价
|
||
模型，billing_note 明确标注代理价、非中转实际账单。
|
||
"""
|
||
report = dict(base)
|
||
report["priced_usd_micros"] = report["usd_micros"]
|
||
report["unpriced_models"] = sorted(
|
||
row["model"] for row in breakdown
|
||
if row["unpriced_tokens"] > 0)
|
||
report["by_model"] = breakdown
|
||
report["billing_note"] = BILLING_NOTE
|
||
return report
|
||
|
||
def dual_column_report_by_task(self, task_id) -> dict:
|
||
"""单任务双列报告（见 _dual_column）。"""
|
||
return self._dual_column(
|
||
self.report_by_task(task_id),
|
||
self.model_breakdown(task_id=task_id))
|
||
|
||
def dual_column_report_by_project(self, project_id) -> dict:
|
||
"""单项目双列报告（见 _dual_column）。"""
|
||
return self._dual_column(
|
||
self.report_by_project(project_id),
|
||
self.model_breakdown(project_id=project_id))
|
||
|
||
def dual_column_report_by_creator(self, creator_id,
|
||
month=None) -> dict:
|
||
"""按创建者双列报告（见 _dual_column）。"""
|
||
return self._dual_column(
|
||
self.report_by_creator(creator_id, month=month),
|
||
self.model_breakdown(creator_id=creator_id, month=month))
|