refactor(storage): 计费读侧收口 core/storage/usage_report.py + 首批 DB 级测试

风险点(架构审查 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 <noreply@anthropic.com>

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}

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()

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

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,
     *,

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
 
 

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