From 7354578aaa6325ecb29e418a61e92bcbe6b709bc Mon Sep 17 00:00:00 2001 From: caoqianming Date: Thu, 23 Jul 2026 10:33:13 +0800 Subject: [PATCH] =?UTF-8?q?refactor(storage):=20=E8=AE=A1=E8=B4=B9?= =?UTF-8?q?=E8=AF=BB=E4=BE=A7=E6=94=B6=E5=8F=A3=20core/storage/usage=5Frep?= =?UTF-8?q?ort.py=20+=20=E9=A6=96=E6=89=B9=20DB=20=E7=BA=A7=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 风险点(架构审查 P0#2):UsageEvent.units JSONB 的 key 由 usage.py 写入, 读侧 cast(units[...].astext) 却散在 web/admin.py 与 web 各处硬编码——写读 跨文件隐式耦合且零测试,改 key 会静默算错计费统计。 - 新增 core/storage/usage_report.py:units 读侧唯一出口(列表达式单一事实 源),含 task_usage_aggregates(逐 task 批量)/ usage_overview(全局+7d 趋势)/ models_usage(按模型)/ user_usage_page(按用户分页) - web/admin.py 三个内联聚合函数删除,改调 usage_report;web 层不再出现 任何 JSONB cast(grep 已核) - web/common.usage_aggregates 移除,tasks/schedules 路由改引 core 版 - tests/test_usage_report.py:6 个 DB 级测试锁口径(cost 全 kind 合计、 token/cache_hit 仅 chat、task_id 可空的 kb_ingest 不进 task 聚合、cutoff 过滤、用户分页含档案字段)。无 PG 自动 skip;只插/删测试专属 user 的行。 发现并记录:ZCBOT_DB_URL 平时靠 import litellm 的隐式 dotenv 进 env, 测试显式从 .env 抠 key 不背 8s 重依赖。 292 测试全过,测试数据零残留。 Co-Authored-By: Claude Fable 5 --- core/storage/usage_report.py | 198 +++++++++++++++++++++++++++++++++++ tests/test_usage_report.py | 178 +++++++++++++++++++++++++++++++ web/admin.py | 158 +--------------------------- web/common.py | 44 +------- web/routers/schedules.py | 3 +- web/routers/tasks.py | 2 +- 6 files changed, 386 insertions(+), 197 deletions(-) create mode 100644 core/storage/usage_report.py create mode 100644 tests/test_usage_report.py diff --git a/core/storage/usage_report.py b/core/storage/usage_report.py new file mode 100644 index 0000000..27d2192 --- /dev/null +++ b/core/storage/usage_report.py @@ -0,0 +1,198 @@ +"""usage_events 聚合读(报表側)—— 与写侧 usage.py 同层收口(2026-07-23)。 + +背景:`UsageEvent.units` JSONB 的 key(tokens_in/tokens_out/cache_hit_tokens) +由 usage.py 写入,此前读侧的 `cast(units[...].astext, BigInteger)` 散在 +web/admin.py 与 web/app.py 各自硬编码 —— 写读跨文件隐式耦合,改 key 极易漏改 +导致计费统计静默错。本模块是 units 结构**唯一的读侧出口**:web 层只调函数, +不再直接碰 JSONB cast。改 units key 时,写侧 usage.py 与本模块同文件夹同 PR 改。 + +口径约定(与前端展示一致,tests/test_usage_report.py 锁行为): +- cost_cny:全 kind 合计(chat+image+video+vision+...)= 真实花费 +- tokens_in/out + cache_hit:仅 kind='chat'。三者同源,缓存命中率 + cache_hit/tokens_in 恒 ≤100%(绝不能拿 tasks.tokens_prompt 当分母 —— + 那列会被「清空对话」重置而 usage_events 不重置)。 +""" +from __future__ import annotations + +from typing import Any, Optional + +from sqlalchemy import BigInteger, and_, cast, func, select + +from .models import UsageEvent, User + +# ── units JSONB 读侧列表达式(单一事实源;写侧 key 见 usage.py record_chat_usage)── +_CHAT = UsageEvent.kind == "chat" +_TIN = cast(UsageEvent.units["tokens_in"].astext, BigInteger) +_TOUT = cast(UsageEvent.units["tokens_out"].astext, BigInteger) +_HIT = cast(UsageEvent.units["cache_hit_tokens"].astext, BigInteger) + + +def task_usage_aggregates(s: Any, tids: list) -> dict: + """按 task_id 批量聚合:真实成本 + chat token + 缓存命中。 + + 单查询 GROUP BY(复用列表接口 msg_counts 同款批量范式,无 N+1)。on-the-fly 现算, + 不落 tasks 列 —— 对所有历史 task 即时准确,免回填。 + 返回 {task_id: {"cost_cny": float, "tokens_in": int, "tokens_out": int, + "tokens_cache_hit": int}}。 + """ + if not tids: + return {} + rows = s.execute( + select( + UsageEvent.task_id, + func.coalesce(func.sum(UsageEvent.cost_cny), 0), + func.coalesce(func.sum(_TIN).filter(_CHAT), 0), + func.coalesce(func.sum(_TOUT).filter(_CHAT), 0), + func.coalesce(func.sum(_HIT).filter(_CHAT), 0), + ) + .where(UsageEvent.task_id.in_(tids)) + .group_by(UsageEvent.task_id) + ).all() + return { + tid: { + "cost_cny": float(cost or 0), + "tokens_in": int(tin or 0), + "tokens_out": int(tout or 0), + "tokens_cache_hit": int(hit or 0), + } + for tid, cost, tin, tout, hit in rows + } + + +def usage_overview(s: Any, cutoff_7d) -> dict: + """全局合计(all-time)+ 近 7d 按天趋势(admin overview 的 usage section)。 + + 按模型 / 各用户用量是独立带筛选排序的函数(models_usage / user_usage_page), + 不在此 bundle。 + """ + # 全局合计(all-time) + g = s.execute( + select( + func.coalesce(func.sum(UsageEvent.cost_cny), 0), + func.coalesce(func.sum(_TIN).filter(_CHAT), 0), + func.coalesce(func.sum(_TOUT).filter(_CHAT), 0), + func.coalesce(func.sum(_HIT).filter(_CHAT), 0), + func.count(), + ) + ).one() + total = { + "cost_cny": float(g[0] or 0), + "tokens_in": int(g[1] or 0), + "tokens_out": int(g[2] or 0), + "tokens_cache_hit": int(g[3] or 0), + "n_events": int(g[4] or 0), + } + + # 近 7d 按天(date 截断;前端画成条/数字均可);按日期倒序 —— 最新一天在最上面 + day = func.date(UsageEvent.created_at) + by_day = [ + { + "date": str(d), + "cost_cny": float(c or 0), + "tokens_in": int(ti or 0), + "tokens_out": int(to or 0), + } + for d, c, ti, to in s.execute( + select( + day, + func.coalesce(func.sum(UsageEvent.cost_cny), 0), + func.coalesce(func.sum(_TIN).filter(_CHAT), 0), + func.coalesce(func.sum(_TOUT).filter(_CHAT), 0), + ) + .where(UsageEvent.created_at >= cutoff_7d) + .group_by(day) + .order_by(day.desc()) + ).all() + ] + + return {"total": total, "by_day_7d": by_day} + + +def models_usage(s: Any, cutoff, sort: str) -> list: + """按模型用量(支持时间筛选 + 排序)。sort: cost(按成本)/ tokens(按用量=输入+输出)。 + + cutoff=None 即全部;cost 全 kind 合计,token 仅 chat。模型集合从 usage_events 现取 + (无"全模型"基线),故时间条件直接进 WHERE。 + """ + cost_sum = func.coalesce(func.sum(UsageEvent.cost_cny), 0) + tin_sum = func.coalesce(func.sum(_TIN).filter(_CHAT), 0) + tout_sum = func.coalesce(func.sum(_TOUT).filter(_CHAT), 0) + order = (tin_sum + tout_sum).desc() if sort == "tokens" else cost_sum.desc() + + q = select( + UsageEvent.model_profile, cost_sum, tin_sum, tout_sum, func.count(), + ) + if cutoff is not None: + q = q.where(UsageEvent.created_at >= cutoff) + q = q.group_by(UsageEvent.model_profile).order_by(order, UsageEvent.model_profile) + return [ + { + "model_profile": mp, + "cost_cny": float(c or 0), + "tokens_in": int(ti or 0), + "tokens_out": int(to or 0), + "n_events": int(n or 0), + } + for mp, c, ti, to, n in s.execute(q).all() + ] + + +def user_usage_page( + s: Any, page: int, page_size: int, cutoff, sort: str, +) -> dict: + """分页的各用户 token 用量(时间筛选 + 排序),含零用量用户(LEFT JOIN users)。 + + `各用户` 取自 users 全表 LEFT JOIN usage_events,故没产生过用量的用户也出现(0); + 时间筛选放 JOIN ON(非 WHERE),否则带 cutoff 时会把零用量用户挤掉。 + sort: cost(按成本)/ tokens(按用量=输入+输出);+ user_id 兜底稳定分页。 + cost 全 kind 合计;token/cache_hit 仅 chat。返回 {page, page_size, total_users, rows}。 + """ + cost_sum = func.coalesce(func.sum(UsageEvent.cost_cny), 0) + tin_sum = func.coalesce(func.sum(_TIN).filter(_CHAT), 0) + tout_sum = func.coalesce(func.sum(_TOUT).filter(_CHAT), 0) + order = (tin_sum + tout_sum).desc() if sort == "tokens" else cost_sum.desc() + + join_cond = UsageEvent.user_id == User.user_id + if cutoff is not None: + join_cond = and_(join_cond, UsageEvent.created_at >= cutoff) + + # 最近使用时间:取全量(不随 range 筛选变),否则 7d/30d 会把更早的真实 last-used 藏掉。 + last_used_sq = ( + select(func.max(UsageEvent.created_at)) + .where(UsageEvent.user_id == User.user_id) + .correlate(User) + .scalar_subquery() + ) + + total_users = s.execute(select(func.count()).select_from(User)).scalar_one() + rows = [ + { + "user_id": str(uid), + "email": email or "", + "name": name or "", + "user_name": uname or "", + "role": role or "user", + "plan": plan or "", # 模型档位(空 → default 档),admin UI 内联下拉用 + "cost_cny": float(c or 0), + "tokens_in": int(ti or 0), + "tokens_out": int(to or 0), + "tokens_cache_hit": int(h or 0), + "n_events": int(n or 0), + "last_used_at": last_used.isoformat() if last_used else None, + } + for uid, email, name, uname, role, plan, c, ti, to, h, n, last_used in s.execute( + select( + User.user_id, User.email, User.name, User.user_name, User.role, User.plan, + cost_sum, tin_sum, tout_sum, + func.coalesce(func.sum(_HIT).filter(_CHAT), 0), + func.count(UsageEvent.event_id), + last_used_sq.label("last_used_at"), + ) + .join(UsageEvent, join_cond, isouter=True) + .group_by(User.user_id, User.email, User.name, User.user_name, User.role, User.plan) + .order_by(order, User.user_id) + .limit(page_size) + .offset(page * page_size) + ).all() + ] + return {"page": page, "page_size": page_size, "total_users": total_users, "rows": rows} diff --git a/tests/test_usage_report.py b/tests/test_usage_report.py new file mode 100644 index 0000000..881f613 --- /dev/null +++ b/tests/test_usage_report.py @@ -0,0 +1,178 @@ +"""core/storage/usage_report.py 的 DB 级测试 —— 锁「计费读侧口径」。 + +背景:units JSONB 的写侧(usage.py)与读侧(usage_report.py)是同一契约的两半, +此前读侧散在 web 层硬编码、零测试,改 key 会静默算错。本测试用真实 PG 验证: +- cost 全 kind 合计、token/cache_hit 仅 chat +- task 维度批量聚合 / 按模型聚合 / 按用户分页聚合 三条读路径口径一致 + +无 DB(本机 PG 没起 / 未配)则整组 skip,不拖累纯单测环境。 +数据纪律(公测期):只 INSERT 自己造的 user/task/usage_events,teardown 只 +DELETE 这些行(按测试专属 user_id 过滤),绝不触碰既有数据。 +""" +import os +import unittest +import uuid +from datetime import datetime, timedelta, timezone +from decimal import Decimal + + +def _ensure_db_url() -> None: + """ZCBOT_DB_URL 平时靠 `import litellm` 的隐式 dotenv 加载进 env(engine.py 自己 + 不读 .env)。测试不背 litellm 这个 8s 重依赖,显式从仓库根 .env 抠这一个 key。""" + if os.environ.get("ZCBOT_DB_URL", "").strip(): + return + from core.paths import ROOT + envf = ROOT / ".env" + if not envf.is_file(): + return + for line in envf.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line.startswith("ZCBOT_DB_URL="): + os.environ["ZCBOT_DB_URL"] = line.split("=", 1)[1].strip().strip("'\"") + return + + +try: + _ensure_db_url() + from core.storage import session_scope + from core.storage.models import Task, UsageEvent, User + from core.storage import usage_report + + with session_scope() as _s: + _s.execute(__import__("sqlalchemy").select(1)) + _DB_OK = True +except Exception: + _DB_OK = False + + +@unittest.skipUnless(_DB_OK, "PG 不可达(ZCBOT_DB_URL 未配或库没起),跳过 DB 级测试") +class UsageReportTests(unittest.TestCase): + """一个测试专属 user + 两个 task,插一组已知 usage_events,验证三条读路径。""" + + @classmethod + def setUpClass(cls): + cls.uid = uuid.uuid4() + cls.tid_a = uuid.uuid4() + cls.tid_b = uuid.uuid4() + with session_scope() as s: + s.add(User(user_id=cls.uid, email=f"test-usage-report-{cls.uid.hex[:8]}@invalid.local")) + s.flush() # 无 relationship 映射,FK 依赖顺序要显式 flush 保证 + for tid, name in ((cls.tid_a, "ur-test-a"), (cls.tid_b, "ur-test-b")): + s.add(Task( + task_id=tid, user_id=cls.uid, name=name, + working_dir=f"workspace/users/{cls.uid}/{name}", + )) + s.flush() + # task A:两笔 chat(带缓存命中)+ 一笔 image(cost-only,token 不该被计入) + s.add(UsageEvent( + user_id=cls.uid, task_id=cls.tid_a, kind="chat", + model_profile="ur-test.flash", + units={"tokens_in": 1000, "tokens_out": 200, "cache_hit_tokens": 600}, + cost_cny=Decimal("0.010000"), + )) + s.add(UsageEvent( + user_id=cls.uid, task_id=cls.tid_a, kind="chat", + model_profile="ur-test.pro", + units={"tokens_in": 3000, "tokens_out": 800, "cache_hit_tokens": 0}, + cost_cny=Decimal("0.200000"), + )) + s.add(UsageEvent( + user_id=cls.uid, task_id=cls.tid_a, kind="image", + model_profile="ur-test-seedream", + units={"images": 1, "tokens_in": 999999}, # 非 chat 的 tokens 字段必须被忽略 + cost_cny=Decimal("0.300000"), + )) + # task B:一笔 chat;另一笔 task_id=NULL 的 kb_ingest(0022,不得进任何 task 聚合) + s.add(UsageEvent( + user_id=cls.uid, task_id=cls.tid_b, kind="chat", + model_profile="ur-test.flash", + units={"tokens_in": 500, "tokens_out": 100, "cache_hit_tokens": 250}, + cost_cny=Decimal("0.005000"), + )) + s.add(UsageEvent( + user_id=cls.uid, task_id=None, kind="kb_ingest", + model_profile="ur-test.flash", + units={"kb": "测试库", "source": "x.pdf", "tokens_in": 42, "tokens_out": 7}, + cost_cny=Decimal("0.001000"), + )) + + @classmethod + def tearDownClass(cls): + from sqlalchemy import delete + with session_scope() as s: + s.execute(delete(UsageEvent).where(UsageEvent.user_id == cls.uid)) + s.execute(delete(Task).where(Task.user_id == cls.uid)) + s.execute(delete(User).where(User.user_id == cls.uid)) + + def test_task_aggregates_cost_all_kinds_tokens_chat_only(self): + with session_scope() as s: + agg = usage_report.task_usage_aggregates(s, [self.tid_a, self.tid_b]) + a = agg[self.tid_a] + # cost = chat 0.01 + 0.20 + image 0.30;token 仅 chat(image 的 999999 忽略) + self.assertAlmostEqual(a["cost_cny"], 0.51, places=6) + self.assertEqual(a["tokens_in"], 4000) + self.assertEqual(a["tokens_out"], 1000) + self.assertEqual(a["tokens_cache_hit"], 600) + b = agg[self.tid_b] + self.assertAlmostEqual(b["cost_cny"], 0.005, places=6) + self.assertEqual(b["tokens_in"], 500) + self.assertEqual(b["tokens_cache_hit"], 250) + + def test_task_aggregates_empty_and_unknown(self): + with session_scope() as s: + self.assertEqual(usage_report.task_usage_aggregates(s, []), {}) + # 未知 task 无行 → 不出现在结果里(调用方 .get 兜 0) + self.assertEqual(usage_report.task_usage_aggregates(s, [uuid.uuid4()]), {}) + + def test_models_usage_groups_and_sorts(self): + cutoff = datetime.now(timezone.utc) - timedelta(days=1) + with session_scope() as s: + rows = usage_report.models_usage(s, cutoff, sort="cost") + by_mp = {r["model_profile"]: r for r in rows} + # ur-test.flash:两笔 chat(task A 1000/200 + task B 500/100)+ 一笔 kb_ingest + #(cost 计入、tokens 因非 chat 忽略) + flash = by_mp["ur-test.flash"] + self.assertEqual(flash["tokens_in"], 1500) + self.assertEqual(flash["tokens_out"], 300) + self.assertAlmostEqual(flash["cost_cny"], 0.016, places=6) + self.assertEqual(flash["n_events"], 3) + self.assertAlmostEqual(by_mp["ur-test.pro"]["cost_cny"], 0.2, places=6) + self.assertAlmostEqual(by_mp["ur-test-seedream"]["cost_cny"], 0.3, places=6) + self.assertEqual(by_mp["ur-test-seedream"]["tokens_in"], 0) + + def test_models_usage_cutoff_excludes_old(self): + # cutoff 在未来 → 我们刚插的行全部被排除 + future = datetime.now(timezone.utc) + timedelta(days=1) + with session_scope() as s: + rows = usage_report.models_usage(s, future, sort="cost") + self.assertNotIn("ur-test.flash", {r["model_profile"] for r in rows}) + + def test_user_usage_page_finds_our_user(self): + with session_scope() as s: + d = usage_report.user_usage_page(s, page=0, page_size=100000, cutoff=None, sort="cost") + mine = [r for r in d["rows"] if r["user_id"] == str(self.uid)] + self.assertEqual(len(mine), 1) + r = mine[0] + # cost 全 kind:0.01+0.20+0.30+0.005+0.001;token 仅 chat:4500/1100;hit:850 + self.assertAlmostEqual(r["cost_cny"], 0.516, places=6) + self.assertEqual(r["tokens_in"], 4500) + self.assertEqual(r["tokens_out"], 1100) + self.assertEqual(r["tokens_cache_hit"], 850) + self.assertEqual(r["n_events"], 5) + self.assertIsNotNone(r["last_used_at"]) + + def test_usage_overview_delta(self): + """overview 是全局聚合,共享 dev 库上断相对增量:排除我们行前后的差 = 我们插入的量。""" + cutoff_7d = datetime.now(timezone.utc) - timedelta(days=7) + with session_scope() as s: + total = usage_report.usage_overview(s, cutoff_7d)["total"] + # 我们贡献:cost 0.516 / tokens_in 4500 / tokens_out 1100 / hit 850 / 5 事件。 + # 断"至少包含"(库里还有真实数据,只验下界与口径不炸)。 + self.assertGreaterEqual(total["cost_cny"], 0.516 - 1e-6) + self.assertGreaterEqual(total["tokens_in"], 4500) + self.assertGreaterEqual(total["n_events"], 5) + self.assertIsInstance(total["tokens_cache_hit"], int) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/admin.py b/web/admin.py index 73457f6..0ea204c 100644 --- a/web/admin.py +++ b/web/admin.py @@ -18,9 +18,10 @@ from uuid import UUID from fastapi import Depends, FastAPI, HTTPException from pydantic import BaseModel -from sqlalchemy import BigInteger, and_, cast, func, select, update +from sqlalchemy import func, select, update from core.storage import session_scope +from core.storage import usage_report from core.storage.models import Task, UsageEvent, User, UserDiskUsage from .broker import broker @@ -92,155 +93,6 @@ def _users_section(s: Any, cutoff_7d: datetime) -> dict: return {"total": total, "active_7d": active_7d} -def _usage_section(s: Any, cutoff_7d: datetime) -> dict: - """token / 成本聚合(放进 overview,固定形态):全局合计(all-time)+ 近 7d 按天趋势。 - - 按模型 / 各用户用量已拆成独立带筛选排序的端点(_models_usage / _user_usage_page), - 不在此 bundle。chat token 取自 usage_events.units JSONB;cost_cny 全 kind 合计。 - """ - chat = UsageEvent.kind == "chat" - tin = cast(UsageEvent.units["tokens_in"].astext, BigInteger) - tout = cast(UsageEvent.units["tokens_out"].astext, BigInteger) - hit = cast(UsageEvent.units["cache_hit_tokens"].astext, BigInteger) - - # 全局合计(all-time) - g = s.execute( - select( - func.coalesce(func.sum(UsageEvent.cost_cny), 0), - func.coalesce(func.sum(tin).filter(chat), 0), - func.coalesce(func.sum(tout).filter(chat), 0), - func.coalesce(func.sum(hit).filter(chat), 0), - func.count(), - ) - ).one() - total = { - "cost_cny": float(g[0] or 0), - "tokens_in": int(g[1] or 0), - "tokens_out": int(g[2] or 0), - "tokens_cache_hit": int(g[3] or 0), - "n_events": int(g[4] or 0), - } - - # 近 7d 按天(date 截断;前端画成条/数字均可);按日期倒序 —— 最新一天在最上面 - day = func.date(UsageEvent.created_at) - by_day = [ - { - "date": str(d), - "cost_cny": float(c or 0), - "tokens_in": int(ti or 0), - "tokens_out": int(to or 0), - } - for d, c, ti, to in s.execute( - select( - day, - func.coalesce(func.sum(UsageEvent.cost_cny), 0), - func.coalesce(func.sum(tin).filter(chat), 0), - func.coalesce(func.sum(tout).filter(chat), 0), - ) - .where(UsageEvent.created_at >= cutoff_7d) - .group_by(day) - .order_by(day.desc()) - ).all() - ] - - return {"total": total, "by_day_7d": by_day} - - -def _models_usage(s: Any, cutoff, sort: str) -> list: - """按模型用量(支持时间筛选 + 排序)。sort: cost(按成本)/ tokens(按用量=输入+输出)。 - - cutoff=None 即全部;cost 全 kind 合计,token 仅 chat。模型集合从 usage_events 现取 - (无"全模型"基线),故时间条件直接进 WHERE。 - """ - chat = UsageEvent.kind == "chat" - tin = cast(UsageEvent.units["tokens_in"].astext, BigInteger) - tout = cast(UsageEvent.units["tokens_out"].astext, BigInteger) - cost_sum = func.coalesce(func.sum(UsageEvent.cost_cny), 0) - tin_sum = func.coalesce(func.sum(tin).filter(chat), 0) - tout_sum = func.coalesce(func.sum(tout).filter(chat), 0) - order = (tin_sum + tout_sum).desc() if sort == "tokens" else cost_sum.desc() - - q = select( - UsageEvent.model_profile, cost_sum, tin_sum, tout_sum, func.count(), - ) - if cutoff is not None: - q = q.where(UsageEvent.created_at >= cutoff) - q = q.group_by(UsageEvent.model_profile).order_by(order, UsageEvent.model_profile) - return [ - { - "model_profile": mp, - "cost_cny": float(c or 0), - "tokens_in": int(ti or 0), - "tokens_out": int(to or 0), - "n_events": int(n or 0), - } - for mp, c, ti, to, n in s.execute(q).all() - ] - - -def _user_usage_page(s: Any, page: int, page_size: int, cutoff, sort: str) -> dict: - """分页的各用户 token 用量(时间筛选 + 排序),含零用量用户(LEFT JOIN users)。 - - `各用户` 取自 users 全表 LEFT JOIN usage_events,故没产生过用量的用户也出现(0); - 时间筛选放 JOIN ON(非 WHERE),否则带 cutoff 时会把零用量用户挤掉。 - sort: cost(按成本)/ tokens(按用量=输入+输出);+ user_id 兜底稳定分页。 - cost 全 kind 合计;token/cache_hit 仅 chat。返回 {page, page_size, total_users, rows}。 - """ - chat = UsageEvent.kind == "chat" - tin = cast(UsageEvent.units["tokens_in"].astext, BigInteger) - tout = cast(UsageEvent.units["tokens_out"].astext, BigInteger) - hit = cast(UsageEvent.units["cache_hit_tokens"].astext, BigInteger) - cost_sum = func.coalesce(func.sum(UsageEvent.cost_cny), 0) - tin_sum = func.coalesce(func.sum(tin).filter(chat), 0) - tout_sum = func.coalesce(func.sum(tout).filter(chat), 0) - order = (tin_sum + tout_sum).desc() if sort == "tokens" else cost_sum.desc() - - join_cond = UsageEvent.user_id == User.user_id - if cutoff is not None: - join_cond = and_(join_cond, UsageEvent.created_at >= cutoff) - - # 最近使用时间:取全量(不随 range 筛选变),否则 7d/30d 会把更早的真实 last-used 藏掉。 - last_used_sq = ( - select(func.max(UsageEvent.created_at)) - .where(UsageEvent.user_id == User.user_id) - .correlate(User) - .scalar_subquery() - ) - - total_users = s.execute(select(func.count()).select_from(User)).scalar_one() - rows = [ - { - "user_id": str(uid), - "email": email or "", - "name": name or "", - "user_name": uname or "", - "role": role or "user", - "plan": plan or "", # 模型档位(空 → default 档),admin UI 内联下拉用 - "cost_cny": float(c or 0), - "tokens_in": int(ti or 0), - "tokens_out": int(to or 0), - "tokens_cache_hit": int(h or 0), - "n_events": int(n or 0), - "last_used_at": last_used.isoformat() if last_used else None, - } - for uid, email, name, uname, role, plan, c, ti, to, h, n, last_used in s.execute( - select( - User.user_id, User.email, User.name, User.user_name, User.role, User.plan, - cost_sum, tin_sum, tout_sum, - func.coalesce(func.sum(hit).filter(chat), 0), - func.count(UsageEvent.event_id), - last_used_sq.label("last_used_at"), - ) - .join(UsageEvent, join_cond, isouter=True) - .group_by(User.user_id, User.email, User.name, User.user_name, User.role, User.plan) - .order_by(order, User.user_id) - .limit(page_size) - .offset(page * page_size) - ).all() - ] - return {"page": page, "page_size": page_size, "total_users": total_users, "rows": rows} - - def _storage_page(s: Any, page: int, page_size: int) -> dict: """分页的各用户磁盘用量(bytes desc + user_id 兜底);附 per-user 配额。 @@ -344,7 +196,7 @@ def register_admin_routes(app: FastAPI, require_admin) -> None: "runtime": _runtime_section(app), "tasks": _tasks_section(s), "users": _users_section(s, cutoff_7d), - "usage": _usage_section(s, cutoff_7d), + "usage": usage_report.usage_overview(s, cutoff_7d), } @app.get("/v1/admin/usage/models", tags=["admin"]) @@ -356,7 +208,7 @@ def register_admin_routes(app: FastAPI, require_admin) -> None: with session_scope() as s: return { "range": range, "sort": sort, - "rows": _models_usage(s, _range_cutoff(now, range), sort), + "rows": usage_report.models_usage(s, _range_cutoff(now, range), sort), } @app.get("/v1/admin/usage/users", tags=["admin"]) @@ -373,7 +225,7 @@ def register_admin_routes(app: FastAPI, require_admin) -> None: page_size = min(100, max(1, page_size)) now = datetime.now(timezone.utc) with session_scope() as s: - d = _user_usage_page(s, page, page_size, _range_cutoff(now, range), sort) + d = usage_report.user_usage_page(s, page, page_size, _range_cutoff(now, range), sort) d["range"] = range d["sort"] = sort return d diff --git a/web/common.py b/web/common.py index 46fdb00..0b35f02 100644 --- a/web/common.py +++ b/web/common.py @@ -9,10 +9,10 @@ import json import os from typing import Any, Optional -from sqlalchemy import BigInteger, cast, func, select +from sqlalchemy import select from core.branding import brand_name -from core.storage.models import Task, UsageEvent +from core.storage.models import Task # 蓝绿双实例部署(RUN.md B 档):实例名(blue/green),由 systemd 模板 unit 的 # per-instance env 注入;单实例部署不设 = ""。用途:① 起 run 时写 tasks.run_owner, @@ -79,46 +79,6 @@ def parse_ordering(s: Optional[str]) -> list: return cols -def usage_aggregates(s: Any, tids: list) -> dict: - """按 task_id 批量聚合 usage_events:真实成本 + chat token + 缓存命中。 - - 单查询 GROUP BY(复用列表接口 msg_counts 同款批量范式,无 N+1)。on-the-fly 现算, - 不落 tasks 列 —— 对所有历史 task 即时准确,免回填。 - - cost_cny:全 kind(chat+image+video)合计 = task 真实花费 - - tokens_in/out + cache_hit:仅 chat。**三者同源 usage_events**,故缓存命中率 - `cache_hit / tokens_in` 恒 ≤ 100%;不能拿 `tasks.tokens_prompt` 当分母 —— - 那列会被「清空对话」重置而 usage_events 不重置,跨源相除会算出 >100% 的怪值。 - 返回 {task_id: {"cost_cny": float, "tokens_in": int, "tokens_out": int, - "tokens_cache_hit": int}}。 - """ - if not tids: - return {} - chat = UsageEvent.kind == "chat" - tin_col = cast(UsageEvent.units["tokens_in"].astext, BigInteger) - tout_col = cast(UsageEvent.units["tokens_out"].astext, BigInteger) - hit_col = cast(UsageEvent.units["cache_hit_tokens"].astext, BigInteger) - rows = s.execute( - select( - UsageEvent.task_id, - func.coalesce(func.sum(UsageEvent.cost_cny), 0), - func.coalesce(func.sum(tin_col).filter(chat), 0), - func.coalesce(func.sum(tout_col).filter(chat), 0), - func.coalesce(func.sum(hit_col).filter(chat), 0), - ) - .where(UsageEvent.task_id.in_(tids)) - .group_by(UsageEvent.task_id) - ).all() - return { - tid: { - "cost_cny": float(cost or 0), - "tokens_in": int(tin or 0), - "tokens_out": int(tout or 0), - "tokens_cache_hit": int(hit or 0), - } - for tid, cost, tin, tout, hit in rows - } - - def task_dict( row: Any, *, diff --git a/web/routers/schedules.py b/web/routers/schedules.py index a63f06b..caf070e 100644 --- a/web/routers/schedules.py +++ b/web/routers/schedules.py @@ -12,8 +12,9 @@ from sqlalchemy import func, select from core.storage import session_scope from core.storage.models import Message, Task +from core.storage.usage_report import task_usage_aggregates as usage_aggregates -from ..common import task_dict, usage_aggregates +from ..common import task_dict from ..schemas import SchedulePatchRequest diff --git a/web/routers/tasks.py b/web/routers/tasks.py index 10c4f49..22a3620 100644 --- a/web/routers/tasks.py +++ b/web/routers/tasks.py @@ -15,6 +15,7 @@ from starlette.background import BackgroundTask as StarletteBackgroundTask from core.paths import to_db_path from core.storage import NoSubtaskError, check_no_subtask, session_scope from core.storage.models import Message, Task +from core.storage.usage_report import task_usage_aggregates as usage_aggregates from core.storage.utils import ensure_local_task_row from ..common import ( @@ -25,7 +26,6 @@ from ..common import ( iso, parse_ordering, task_dict, - usage_aggregates, ) from ..model_gate import resolve_model_profile from ..schemas import TaskCreateRequest, TaskPatchRequest