#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import gzip import json import math import re import subprocess import sys from collections import defaultdict from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo import requests ROOT = Path(__file__).resolve().parents[2] GRAFANA_SCRIPTS = ROOT / "agentkit" / "skills" / "nutshell-skills" / "platform" / "op-clinic-grafana-access" / "scripts" if str(GRAFANA_SCRIPTS) not in sys.path: sys.path.insert(0, str(GRAFANA_SCRIPTS)) from common import grafana_query, grafana_session # noqa: E402 SHANGHAI = ZoneInfo("Asia/Shanghai") UTC = timezone.utc OUTPUT_ROOT = ROOT / "outputs" / "usage_reports" SPLIT_OUTPUT_ROOT = ROOT / "outputs" / "usage_daily_reports" DATASOURCE_ID = 80 GRAFANA_BASE_URL = "https://clinic.pingcap.com" GRAFANA_PATH_PREFIX = "/grafana" HELIX_BASE_URL = "https://helix.pingcap.net" HELIX_ASSIGNMENT_ID = "80b4a524-c232-400f-ab78-8995d9d1a774" KUBE_CONTEXT = "arn:aws:eks:us-west-2:557537366020:cluster/devops-prod" KUBE_NAMESPACE = "investibook" PROD_DB_PATH = "/data/db/investibook.db" PROM_LABELS = 'cluster="prod-devops",namespace="investibook",job="backend"' def parse_day(text: str) -> date: return date.fromisoformat(text) def iter_days(start: date, end: date): current = start while current <= end: yield current current += timedelta(days=1) def local_window(day: date) -> tuple[datetime, datetime]: start = datetime.combine(day, time.min, tzinfo=SHANGHAI) return start, start + timedelta(days=1) def iso(dt: datetime) -> str: return dt.isoformat() def utc_iso(dt: datetime) -> str: return dt.astimezone(UTC).isoformat() def as_float(value) -> float: try: out = float(value) except (TypeError, ValueError): return 0.0 if math.isnan(out) or math.isinf(out): return 0.0 return out def prom_scalar(payload: dict) -> float: result = (((payload or {}).get("data") or {}).get("result")) or [] if not result: return 0.0 return as_float((result[0].get("value") or [None, 0])[1]) def prom_vector(payload: dict) -> list[dict]: rows = [] for item in (((payload or {}).get("data") or {}).get("result")) or []: metric = item.get("metric") or {} value = as_float((item.get("value") or [None, 0])[1]) rows.append({"metric": metric, "value": value}) return rows def query_prometheus(session, expr: str, end: datetime) -> dict: return grafana_query(session, DATASOURCE_ID, expr, time_ts=int(end.astimezone(UTC).timestamp())) def load_team_mapping() -> dict[str, dict]: mapping: dict[str, dict] = {} by_email: dict[str, list[dict]] = defaultdict(list) with (ROOT / "pingcap_members_by_team.csv").open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): email = (row.get("email") or "").strip().lower() if email: by_email[email].append(row) for email, rows in by_email.items(): active = [row for row in rows if (row.get("status") or "").strip().lower() == "active"] or rows primary = [row for row in active if (row.get("is_primary_dept") or "").strip().lower() == "yes"] or active row = primary[0] mapping[email] = { "user_email": email, "name": row.get("name") or "", "en_name": row.get("en_name") or "", "team_path": row.get("team_path") or "", "team": row.get("team") or "", "mapping_source": "pingcap_members_by_team.csv", "ambiguous": len(primary) > 1, } override_path = ROOT / "docs" / "guides" / "investibook-usage-team-overrides.csv" with override_path.open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): email = (row.get("usage_email") or "").strip().lower() if not email: continue team_path = row.get("team_path") or "" mapping[email] = { "user_email": email, "name": row.get("name") or "", "en_name": row.get("en_name") or "", "team_path": team_path, "team": team_path.split("/")[-1].strip() if team_path else "", "mapping_source": "investibook-usage-team-overrides.csv", "ambiguous": False, } return mapping REMOTE_SQLITE_SUMMARY = r''' import gzip, json, sqlite3, sys from collections import defaultdict from datetime import datetime, timezone from zoneinfo import ZoneInfo db_path, start_utc, end_utc = sys.argv[1], sys.argv[2], sys.argv[3] output_path = sys.argv[4] if len(sys.argv) > 4 else "" shanghai = ZoneInfo("Asia/Shanghai") max_content_chars = 2000 def parse_dt(value): text = str(value or "").strip() if text.endswith("Z"): text = text[:-1] + "+00:00" dt = datetime.fromisoformat(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def day_key(value): return parse_dt(value).astimezone(shanghai).date().isoformat() def compact_content(value): text = value or "" if len(text) <= max_content_chars: return text return text[:max_content_chars] + "\n...[truncated]" conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row messages = defaultdict(lambda: defaultdict(lambda: { "message_count": 0, "user_message_count": 0, "assistant_message_count": 0, "tool_message_count": 0, "other_message_count": 0, })) user_messages = defaultdict(lambda: defaultdict(list)) all_user_messages = defaultdict(list) scenario_messages = defaultdict(lambda: defaultdict(list)) all_scenario_messages = defaultdict(list) for row in conn.execute( """ select id, workspace_id, role, content, created_at from messages_v2 where datetime(created_at) >= datetime(?) and datetime(created_at) < datetime(?) """, (start_utc, end_utc), ): day = day_key(row["created_at"]) bucket = messages[day][row["workspace_id"]] role = str(row["role"] or "") bucket["message_count"] += 1 if role in ("user", "system"): scenario_item = { "message_id": row["id"], "role": role, "content": compact_content(row["content"]), "content_chars": len(row["content"] or ""), "created_at": row["created_at"], } scenario_messages[day][row["workspace_id"]].append(scenario_item) all_scenario_messages[row["workspace_id"]].append(scenario_item) if role == "user": bucket["user_message_count"] += 1 user_messages[day][row["workspace_id"]].append({ "message_id": row["id"], "content": compact_content(row["content"]), "content_chars": len(row["content"] or ""), "created_at": row["created_at"], }) all_user_messages[row["workspace_id"]].append({ "message_id": row["id"], "content": compact_content(row["content"]), "content_chars": len(row["content"] or ""), "created_at": row["created_at"], }) elif role == "assistant": bucket["assistant_message_count"] += 1 elif role == "tool": bucket["tool_message_count"] += 1 else: bucket["other_message_count"] += 1 tokens = defaultdict(lambda: defaultdict(lambda: { "interaction_records": 0, "input_tokens": 0, "output_tokens": 0, "reasoning_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0, "cost_amount": 0.0, })) token_workspace_rows = defaultdict(lambda: defaultdict(lambda: { "workspace_id": "", "user_email": "", "channel": "unknown", "runtime": "unknown", "interaction_records": 0, "input_tokens": 0, "output_tokens": 0, "reasoning_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0, "cost_amount": 0.0, })) TOKEN_KEYS = ["input_tokens", "output_tokens", "reasoning_tokens", "cache_read_tokens", "cache_write_tokens"] def token_runtime(session_id): text = str(session_id or "") if text.startswith("ses_"): return "opencode" if text.startswith("image-parse:"): return "image-parse" return "pi" def usage_values(row): return { key: int(row[key] or 0) for key in TOKEN_KEYS } | {"cost_amount": float(row["cost_amount"] or 0)} def usage_diff(after, before): before = before or {key: 0 for key in TOKEN_KEYS} | {"cost_amount": 0.0} values = {} for key in TOKEN_KEYS: values[key] = max(0, int(after.get(key) or 0) - int(before.get(key) or 0)) values["cost_amount"] = max(0.0, float(after.get("cost_amount") or 0) - float(before.get("cost_amount") or 0)) return values def add_token_contribution(day, row, values, records): workspace_id = row["workspace_id"] runtime = token_runtime(row["session_id"]) bucket = tokens[day][workspace_id] bucket["interaction_records"] += records for key in TOKEN_KEYS: bucket[key] += int(values.get(key) or 0) bucket["cost_amount"] += float(values.get("cost_amount") or 0) row_key = (workspace_id, (row["user_email"] or "").strip().lower(), (row["channel"] or "unknown").strip() or "unknown", runtime) row_bucket = token_workspace_rows[day][row_key] row_bucket["workspace_id"] = workspace_id row_bucket["user_email"] = row_key[1] row_bucket["channel"] = row_key[2] row_bucket["runtime"] = runtime row_bucket["interaction_records"] += records for key in TOKEN_KEYS: row_bucket[key] += int(values.get(key) or 0) row_bucket["cost_amount"] += float(values.get("cost_amount") or 0) def iter_report_days(): start = parse_dt(start_utc).astimezone(shanghai).date() end = (parse_dt(end_utc).astimezone(shanghai) - __import__("datetime").timedelta(microseconds=1)).date() current = start delta = __import__("datetime").timedelta(days=1) while current <= end: yield current.isoformat(), datetime.combine(current, __import__("datetime").time.min, tzinfo=shanghai), datetime.combine(current + delta, __import__("datetime").time.min, tzinfo=shanghai) current += delta active_session_ids = [ row["session_id"] or "" for row in conn.execute( """ select distinct session_id from interaction_token_usage where datetime(created_at) >= datetime(?) and datetime(created_at) < datetime(?) """, (start_utc, end_utc), ) ] usage_rows = [] if active_session_ids: placeholders = ",".join("?" for _ in active_session_ids) usage_rows = list(conn.execute( f""" select id, workspace_id, session_id, channel, user_email, model_tier, input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, cost_amount, created_at from interaction_token_usage where session_id in ({placeholders}) and datetime(created_at) < datetime(?) order by session_id, datetime(created_at), id """, tuple(active_session_ids) + (end_utc,), )) usage_by_session = defaultdict(list) for row in usage_rows: usage_by_session[row["session_id"] or ""].append(row) for session_id, rows_for_session in usage_by_session.items(): runtime = token_runtime(session_id) if runtime == "opencode": for day, day_start, day_end in iter_report_days(): latest_before_start = None latest_before_end = None rows_in_day = 0 for row in rows_for_session: created = parse_dt(row["created_at"]) if created < day_start: latest_before_start = row if created < day_end: latest_before_end = row if day_start <= created < day_end: rows_in_day += 1 if not latest_before_end or rows_in_day <= 0: continue values = usage_diff( usage_values(latest_before_end), usage_values(latest_before_start) if latest_before_start else None, ) if values["cost_amount"] <= 0 and not any(values[key] > 0 for key in TOKEN_KEYS): continue add_token_contribution(day, latest_before_end, values, rows_in_day) else: for row in rows_for_session: created = parse_dt(row["created_at"]) if parse_dt(start_utc) <= created < parse_dt(end_utc): add_token_contribution(day_key(row["created_at"]), row, usage_values(row), 1) created_workspaces = defaultdict(list) for row in conn.execute( """ select id, created_at from workspaces where datetime(created_at) >= datetime(?) and datetime(created_at) < datetime(?) """, (start_utc, end_utc), ): created = row["created_at"] if not created: continue created_workspaces[day_key(created)].append(row["id"]) workspace_ids = set() for bucket in messages.values(): workspace_ids.update(bucket.keys()) for bucket in tokens.values(): workspace_ids.update(bucket.keys()) for ids in created_workspaces.values(): workspace_ids.update(ids) workspace_meta = {} if workspace_ids: placeholders = ",".join("?" for _ in workspace_ids) for row in conn.execute( f""" select w.id, w.name, w.description, w.user_id, u.email as owner_email, w.created_at, w.updated_at from workspaces w left join users u on u.id = w.user_id where w.id in ({placeholders}) """, tuple(sorted(workspace_ids)), ): workspace_meta[row["id"]] = dict(row) payload = json.dumps({ "workspace_meta": workspace_meta, "messages": {day: dict(items) for day, items in messages.items()}, "user_messages": {day: dict(items) for day, items in user_messages.items()}, "all_user_messages": dict(all_user_messages), "scenario_messages": {day: dict(items) for day, items in scenario_messages.items()}, "all_scenario_messages": dict(all_scenario_messages), "tokens": {day: dict(items) for day, items in tokens.items()}, "token_workspace_rows": { day: list(items.values()) for day, items in token_workspace_rows.items() }, "created_workspaces": dict(created_workspaces), }, ensure_ascii=False) if output_path: if output_path.endswith(".gz"): with gzip.open(output_path, "wt", encoding="utf-8") as f: f.write(payload) else: with open(output_path, "w", encoding="utf-8") as f: f.write(payload) else: print(payload) ''' def load_prod_sqlite_summary(start_day: date, end_day: date, pod: str) -> dict: start, _ = local_window(start_day) _, end = local_window(end_day) remote_output = f"/tmp/investibook_sqlite_summary_{start_day.isoformat()}_{end_day.isoformat()}.json.gz" cmd = [ "kubectl", "exec", "-n", KUBE_NAMESPACE, pod, "--context", KUBE_CONTEXT, "--", "python3", "-c", REMOTE_SQLITE_SUMMARY, PROD_DB_PATH, utc_iso(start), utc_iso(end), remote_output, ] subprocess.run(cmd, cwd=ROOT, text=True, capture_output=True, check=True) stat_proc = subprocess.run( [ "kubectl", "exec", "-n", KUBE_NAMESPACE, pod, "--context", KUBE_CONTEXT, "--", "python3", "-c", "import os,sys; print(os.path.getsize(sys.argv[1]))", remote_output, ], cwd=ROOT, text=True, capture_output=True, check=True, ) remote_size = int(stat_proc.stdout.strip()) local_output = Path("/tmp") / Path(remote_output).name cp_proc = subprocess.run( [ "kubectl", "cp", "--context", KUBE_CONTEXT, "-n", KUBE_NAMESPACE, f"{pod}:{remote_output}", str(local_output), ], cwd=ROOT, text=True, capture_output=True, check=False, ) if cp_proc.returncode == 0 and local_output.exists(): compressed = local_output.read_bytes() try: return json.loads(gzip.decompress(compressed).decode("utf-8")) except Exception as exc: raise RuntimeError( f"failed to parse copied remote SQLite summary gzip ({len(compressed)} bytes): {exc}" ) from exc chunk_size = 64 * 1024 compressed = bytearray() for offset in range(0, remote_size, chunk_size): limit = min(chunk_size, remote_size - offset) chunk_proc = None for _attempt in range(3): chunk_proc = subprocess.run( [ "kubectl", "exec", "-n", KUBE_NAMESPACE, pod, "--context", KUBE_CONTEXT, "--", "python3", "-c", ( "import sys; " "path=sys.argv[1]; offset=int(sys.argv[2]); size=int(sys.argv[3]); " "f=open(path,'rb'); f.seek(offset); sys.stdout.buffer.write(f.read(size))" ), remote_output, str(offset), str(limit), ], cwd=ROOT, capture_output=True, check=False, ) if len(chunk_proc.stdout) == limit: break assert chunk_proc is not None if len(chunk_proc.stdout) != limit: raise RuntimeError( f"short read from remote SQLite summary at offset {offset}: " f"expected {limit} bytes, got {len(chunk_proc.stdout)}" ) compressed.extend(chunk_proc.stdout) try: return json.loads(gzip.decompress(bytes(compressed)).decode("utf-8")) except Exception as exc: raise RuntimeError( f"failed to parse remote SQLite summary gzip ({len(compressed)} bytes): {exc}" ) from exc def find_backend_pod() -> str: proc = subprocess.run( [ "kubectl", "get", "pods", "-n", KUBE_NAMESPACE, "--context", KUBE_CONTEXT, "-l", "app=backend", "-o", "jsonpath={.items[0].metadata.name}", ], cwd=ROOT, text=True, capture_output=True, check=True, ) pod = proc.stdout.strip() if not pod: raise RuntimeError("backend pod not found") return pod def team_for(email: str, mapping: dict[str, dict]) -> dict: normalized = (email or "").strip().lower() return mapping.get( normalized, { "user_email": normalized, "name": "", "en_name": "", "team_path": "", "team": "", "mapping_source": "", "ambiguous": False, }, ) def round_money(value: float | None) -> float | None: if value is None: return None return round(value + 0.0000000001, 4) MESSAGE_CATEGORY_ORDER = [ "查询纠偏", "诊断纠偏", "产出物迭代", "Ad hoc 查询", "诊断启动", "解释追问", "流程控制/确认", "方法论/能力咨询", "背景补充", "未归类", ] MESSAGE_CATEGORY_MANUAL_OVERRIDES_PATH = ROOT / "tools" / "usage_reporting" / "message_category_manual_overrides.json" _MESSAGE_CATEGORY_MANUAL_OVERRIDES: list[dict] | None = None MESSAGE_CATEGORY_KEYWORDS = { "查询纠偏": [ "查错", "查的不对", "查得不对", "不对吧", "不对 table", "对不上", "数据是不对", "不太对", "不对，", "不对 我", "我没有看到", "是不是看的", "建表都没有", "不要构造", "見間違えて", "提供していないと思います", "とは思えない", "わかると思って", "不是 ，给你一个例子", "不是，给你一个例子", "没有 dmctl", "無い", "ないように", "なさそう", "not correct", "形式が違う", "url也不对", "也不对", "点进去确实", "不是tiproxy", "不是 ti proxy", "不是你", "查不到数据", "写错了", "你从那里看到", "你从哪里看到", "どこからの根拠", "適用されない", "都是要付费", "应该是", "没有到达瓶颈", "没有 tiflash", "没有 tilfash", "削除されていません", "間違えではない", "个数明显不对", "个数不对", "不一致", "認識は正しい", "并没有用tiup", "問題ないです", "不要用dash", "不是iops", "不是 iops", "不是release", "correction:", "less than the quota", "没找到", "没看到", "没有开启", "一堆 cpu", "看错", "查的话", "没有查到", "没有在", "不是今天", "ではなく", "时间有点怪", "正しいですか", "口径错", "数据源", "datasource", "应该查", "不是这个", "重新查", "过滤", "时间窗", "cluster 不对", "集群不对", ], "诊断纠偏": [ "结论不对", "根因不对", "逻辑不", "不成立", "不能这么", "不是根因", "推理", "质疑", "你这个结论", "你别乱给", "别乱给", "改口", "登错环境", "登錯環境", ], "产出物迭代": [ "总结", "报告", "飞书", "feishu", "docx", "markdown", "md 表格", "客户回复", "授权", "管理权限", "生成一个feishu", "评论一下", "draft", "翻译", "translate", "润色", "整理成", "整理", "写成", "同步到", "review", "レビュー", "まとめ", "できますか", "もらえますか", "記載して", "分けてもらえますか", "reply with", "/issue", "summary", "英文版", "日文版", "中文版", "日本語で", "简短", "簡潔", "完整的 正确写法", "完整的正确写法", "正确写法", "轉成", "转成", "横向的表格", "表格", "补齐", "コピーできるように", "マークダウン", "回答ドラフト", "回答案", "回答文", "回答全文", "回答全体", "回答", "書き直", "再作成", "全体再掲", "サマリー", "再掲", "md file", "context to a md", "evidence目录", "版本", "md 文件", "md文件", "doc reference", "ドラフトを作成", "ユーザに", "顧客の質問", "案内しています", "書かなくていい", "再缩短", "再縮短", "summarize", "英語にして", "英語で作成", "テキストでください", "urlをください", "URLをください", "パネルのURL", "列出每个", "补上", "完成日志", "输出", "作文", "顧客に質問", "書いてください", "主眼", "精査", "確かめてください", "作ってください", "作ってみて", "用中文回复", "中文回复", "用中文再回答", "用中文", "用英文", "英文", "重新写", "重新整理", "说清楚来龙去脉", "来龙去脉", "描述日志和现象", "请求发我", "发我", "分享这个文档", "改写下", "重新帮我写", "再修订", "修订一下", "原文发一下", "原样发", "不要做格式化", "upload to the doc", "file attachment", "workarounds", "github issue report", "日本語を直して", "直して", "缩句", "太长了", "只想了解过程", "top3 列举", "列举出来", "给我一段英语", "每一行的timeline", "加一个列", "生成一个 python 脚本", "python 脚本", "脚本实现", "用函数调用", "生产脚本", "给我拷贝", "脚本内容", "code snippet", "collect_asia_prod.py", "subcmd", "再简洁", "简洁啊", "提案してください", "统计窗口改为", "组件名称", "删掉", "重新画", "リンクをください", "linkをください", ], "Ad hoc 查询": [ "查一下", "查看", "看一下", "看下", "check一下", "拉一下", "拉取", "查询", "能找到", "你能查", "你能访问", "搜索一下", "找出", "检查下", "检查", "进一步追查", "通过日志排查", "完整的sql", "有没有", "是否有", "是不是", "是多少", "多大", "什么时候", "最新", "情况怎么样", "情况如何", "相关性如何", "情况呢", "状况", "状況", "如何定位", "怎么定位", "哪个", "哪三个", "哪些数据", "有哪些集群", "有关键 log", "获取到", "怎么下载", "怎么看log", "给我一个", "排序输出", "更新频率", "多吗", "利用率", "log commit", "提供 logs", "provide logs", "find evidence", "find evidences", "direct evidences", "current status", "did the", "are the", "重新看", "再看", "进一步查看", "抓一次", "抓1分钟", "抓一分钟", "cpu profiling", "cpu profilling", "官方文档", "documentation", "topsql", "slow query", "慢查询", "ddl", "qps", "rcu", "排查告警", "o11y-", "digestだせますか", "出せますか", "だせますか", "確認して", "確認できますか", "改善していますか", "調べて", "調べてください", "調査", "調査を", "教えて", "教えて下さい", "どうですか", "どうdすか", "どうでしょう", "状況は", "検索", "見えます", "情報はありますか", "cpu使用率", "cpu 呢", "tidb/tikv cpu", "ログ", "slow log", "查到", "查源码", "源码", "在哪看", "统计一下", "收集一下", "耗时统计", "io 吞吐", "验证", "请告诉我会话", "会话", "查找 deploy type", "deploy type", "会怎样", "會怎樣", "改用", "如果不改", "如何优化", "如何開啟", "如何开启", "情報が欲しい", "有 insert", "insert嗎", "根拠を提示", "提示してください", "cdc", "changefeed", "binding", "加速", "analyze 的速度", "调到多少", "調到多少", "多少节点", "如何通过", "如何一个 sql", "enabled by default", "エラーが出ていましたか", "パーマリンク", "メトリクス", "メトリクスをください", "見せてください", "確認しました", "看看日志", "单独查", "哪几个集群", "最近 1h", "最近1h", "cluster status", "クラスターステータス", "同一時間帯と比較", "比較", "io 帮我识别", "帮我识别", "火焰图", "哪里打开", "cmd_wait在哪里看", "按日统计", "用这个指标计算", "抽取出", "抓取每天", "你加载了哪些 tool", "提供 logs", "按分钟", "按日", "存储容量", "storage size", "所在节点", "scale cluster", "列出证据", "证据", "主要 ru", "tidb pod", "搜索", "最大配额", "效果如何", "带上了没有", "主要看", "tableScan", "table scan", "月均", "多少天", "使用情况", "p99", "p999", "列出依据", "依据", "用量", "iops使用", "raft iops", "sql digest", "metric", "latency", "干嘛的", ], "诊断启动": [ "分析", "诊断", "根因", "原因", "rca", "root cause", "为什么会", "告警原因", "判断", "影响", "处理建议", "故障", "异常", "有问题", "怎么回事", "导致的吗", "巡检", "帮我排查", "请排查", "排查一下", "排查该问题", "是否正常", "是否存在", "check集群", "資料消失", "数据消失", "上限", "oom", "内存", "メモリ", "扩容", "锁等待问题", "スケールダウン", "スケールアウト", "リバランス", "resource不足", "リソース不足", "worker busy", "server is busy", "クラッシュ", "エラーが発生", "grpc duration", "disk io", "coprocessor disk io", "任務建立失敗", "任务建立失败", "理由について", "出问题", "排查问题", "please investigate", "investigate whether", "lock contention", "热点", "cpu throttle", "503", "下一步排查", "下一步派差", "进一步深挖", "资源评估", "可行性", "是否已经恢复", "状态如何", "上游", "下游集群", "下游", "follower reader", "等待tso", "等待 tso", "排查下", "brプロセス", "imported length mismatch", "backfill", "total_suspend_time", "suspend_time", "CPU-bound", "bottleneck", "checksum", "rolling restart", "滚动重启", "卡住了", "still persists", ], "解释追问": [ "为什么", "为何", "理由", "起因", "説明して", "什么意思", "机制", "原理", "代码路径", "关系", "解释", "说明", "解读", "怎么理解", "是什么", "どうして", "なぜ", "不用", "没理解", "不理解", ], "流程控制/确认": [ "继续", "可以", "ok", "好的", "需要", "yes", "是的", "hi", "hello", "share", "更新下", "拉下这个 skill", "想想办法", "你能拿到吗", "不用覆盖", "只看", "我要看的是", "时间缩小", "そちらを", "でした", "ごめんなさい", "先这样", "开始", "再来", "再加", "追加", "force", "/new", "go ahead", "please continue", "continue where you stopped", "retry", "今天", "刚刚", "同期間", "先に", "同时拉", "确认", "確認をお願いします", "お願いします", "请重试", "重试", "先别动", "stop", "/reset", "作業を進めて", "はい", "如果查不到就算了", "处理一下", "/stats", "返信しました", "check it", ], "方法论/能力咨询": [ "你支持", "能不能支持", "能否支持", "能力边界", "subagent", "一般怎么", "通常怎么", "怎么判断", "如何判断", "怎么分析", "排查流程", "如何确定", "怎么操作", "如何删除", "如何识别", "如何从", "如何從", "有効にする手順", "手順", "步骤", "全流程", "操作全流程", "方法を知りたい", "実際に確認したい", "会占用很多资源吗", "必须有", "方法はありますか", "ほうがいいですか", "提供给我方法", "找业务团队", "找业务", "oncaller", "skills", "skill", "安装 ", "重新安装", "clinic を使う方法", "如何创建", "如何常看", "归档方案", "workaround", ], } BACKGROUND_CONTEXT_KEYWORDS = [ "uploaded image", "uploaded file", "补充一下", "背景", "并没有变化", "没有限制", "显示0", "ops.tidbcloud.com", "cluster id -->", "cluster id", "cluster_id", "集群id", "集群 id 为", "时间段", "nodediskverylowspace", "tidb_enable", "running sql", "fail to parse", "thread cpu", "cpuです", "网络连不通", "物理集群", "这是 staging", "img img_", "监控所观察到", "只是创建连接慢了", "cpu有一些上升", "アップロードされていません", "チャットで会話", "下面是执行计划", "这是执行计划", "スクリーンショット", "以前共有しました", "時間軸", "北京时间", "加算されますよね", "配置过了", "也重启了", "plan cache", "ops avg-miss", "main clusterのid", "idです", "api key", "gcpに問い合わせ", "gcpni", "上位に出ています", "resourcegroup", "duration值", "traceid", "show table regions", "memory dump", "mem", "压测时间", "describe pod", "没有event", "staging cluster", "cluster ", "tenant name", "任务显示还在 running", "默认都开启", "audit manager", "audit log", "aws us-", "gcp /", "dedicated 集群", "transaction id", "idも", "cluster-id", "record 的 byte", "byte 大小", "table:", "aws /", "op集群", "scale_cop_worker.svg", "snap-io-max-bytes-per-sec", "text file", "defaultの", "流量回落", "lag在下降", "最近1小时", "最近 1 小时", "最近7天", "最近 7 天", "最近30天", "最近 30 天", "最近12小时", "最近 12 小时", "过去7天", "过去 7 天", "get /pd/api", "save checkpoint", "dry-run", "log progress", "estimated remaining time", "最近 1h", "最近1h", "[info]", "报错", "error", "exception", "traceback", "stack trace", "stacktrace", "panic", "[error]", "[warn]", "warning", "store_not_match", "tikv-server", "tidb-server", "pd-server", ] EVIDENCE_QUERY_TERMS = [ "cluster", "集群", "log", "日志", "sql", "query", "metric", "grafana", "clinic", "tidb", "tikv", "pd", "tiflash", "cpu", "iops", "region", "table", "status", "状态", "配置", "config", "流量", "latency", "延迟", ] QUESTION_MARKERS = [ "?", "？", "吗", "嗎", "么", "呢", "吧", "会不会", "會不會", "会怎样", "會怎樣", "怎样", "怎樣", "哪里", "哪儿", "哪裡", "哪个", "哪個", "哪几", "那个", "那個", "啥", "是啥", "どこ", "对吧", "か？", "ですか", "ますか", ] URL_RE = re.compile(r"https?://|www\.", re.IGNORECASE) SQL_RE = re.compile( r"^\s*(select|show|explain|with|insert|update|delete|create|alter|drop|admin|analyze)\b", re.IGNORECASE, ) NUMERIC_ID_RE = re.compile(r"^\s*\d{12,}\s*$") LABELLED_ID_RE = re.compile( r"\b(cluster|org|tenant|workspace|table|index|session|process|connection|trace|transaction|digest)\s*id\b|" r"(集群|表|任务|工作区|会话|连接)\s*id", re.IGNORECASE, ) CASE_ID_RE = re.compile(r"\b(?:APID|GTOC|TCOC|TCID|O11Y|ST|ALT|JPID)-+\d+\b", re.IGNORECASE) FILE_REF_RE = re.compile(r"\b(?:file|img|image)_v\d+_[A-Za-z0-9_-]+|\bfile_[A-Za-z0-9_-]+\.(?:txt|log|json)\b") API_KEY_LIKE_RE = re.compile(r"^[A-Za-z0-9+/=]{40,}$") TIME_RANGE_RE = re.compile( r"\b\d{4}[-/]\d{1,2}[-/]\d{1,2}\b|\b\d{1,2}[-/]\d{1,2}\b|\b\d{1,2}:\d{2}\b|UTC[+-]\d+|JST\b|[+-]\d{4}\b", re.IGNORECASE, ) TECHNICAL_IDENTIFIER_RE = re.compile(r"^[A-Z][A-Za-z0-9_]*(?:Export|Backup|Import|Restore|Task|Job|Binding|Config)$") SNAKE_IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9]*(?:_[A-Za-z0-9]+){1,}$") MENTION_ONLY_RE = re.compile(r"^@_user_\d+\s*[↑]*$") DOTTED_IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_-]*(?:\.[A-Za-z0-9_-]+)+$") SCENARIO_RULES = [ { "group": "diagnosis", "category": "sql_plan_or_query", "keywords": [ "sql", "query", "digest", "topsql", "slow query", "slow sql", "慢查询", "执行计划", "plan", "optimizer", "优化器", "索引", "index", "scan", "coprocessor task", "process keys", "total keys", ], }, { "group": "diagnosis", "category": "ddl_or_schema_change", "keywords": [ "ddl", "show ddl", "drop database", "drop schema", "drop table", "alter table", "schema change", "add index", "drop index", "ddl job", "schemaid", ], }, { "group": "diagnosis", "category": "session_or_process_management", "keywords": [ "processlist", "process id", "connection id", "conn_id", "kill", "kill 不掉", "会话", "セッション", ], }, { "group": "diagnosis", "category": "transaction_or_lock_contention", "keywords": [ "transaction", "txn", "lock", "deadlock", "pessimistic", "prewrite", "mvcc", "commit conflict", "锁", "事务", "悲观锁", "锁等待", "冲突", ], }, { "group": "diagnosis", "category": "tikv_hotspot_or_coprocessor", "keywords": [ "tikv", "hotspot", "hot region", "热点", "coprocessor", "coprocess", "cop request", "cop请求", "region", "unified read pool", "readpool", "store:", "peer", "peer不足", "peer 不足", "no region", "no leader", "not_leader", ], }, { "group": "diagnosis", "category": "storage_or_io_bottleneck", "keywords": [ "rocksdb", "compaction", "pending compaction bytes", "raftstore", "raft", "commit-log-duration", "append-log-duration", "write queue", "iops", "disk", "io", "i/o", "磁盘", "存储", "写入放大", ], }, { "group": "diagnosis", "category": "backup_or_restore", "keywords": [ "backup", "restore", "pitr", "br", "snapshot", "dual region backup", "バックアップ", "リストア", "备份", "恢复", ], }, { "group": "diagnosis", "category": "data_import_or_load", "keywords": [ "import", "importinto", "import into", "show import jobs", "load data", "csv", "kv convert", "kvconvert", "source file", "data file", "导入", "数据导入", "インポート", "件数", "ファイル", ], }, { "group": "diagnosis", "category": "memory_or_oom_pressure", "keywords": ["oom", "memory", "heap", "内存", "メモリ"], }, { "group": "diagnosis", "category": "gc_or_allocator_pressure", "keywords": ["gc", "allocator", "malloc", "runtime", "go runtime"], }, { "group": "diagnosis", "category": "resource_control_or_capacity", "keywords": [ "ru", "resource control", "token limit", "quota", "throttle", "noisy neighbor", "限流", "容量", "capacity", "cpu", "qps", "p99", "p999", "latency", "duration", "高レイテンシー", ], }, { "group": "diagnosis", "category": "pd_tso_or_scheduling", "keywords": [ "pd", "tso", "timestamp", "get timestamp too slow", "scheduler", "scheduling", "operator", "evict", "leader", "调度", "スケジューリング", ], }, { "group": "diagnosis", "category": "cdc_or_replication_lag", "keywords": ["cdc", "ticdc", "changefeed", "replication", "同步", "延迟", "lag", "レプリケーション"], }, { "group": "diagnosis", "category": "network_or_connectivity", "keywords": ["network", "connect timeout", "connection timeout", "connectivity", "dns", "packet", "grpc", "连接", "ネットワーク"], }, { "group": "diagnosis", "category": "component_failure_or_bug", "keywords": [ "bug", "panic", "crash", "crashloop", "crash loop", "restart", "reboot", "down store", "store down", "discoverdownstore", "autoid", "auto-id", "起動しなかった", "不具合", "重启", "故障", "异常", "失敗", ], }, { "group": "diagnosis", "category": "cluster_topology_or_scaling", "keywords": [ "topology", "modifyclustersize", "scaling", "scale", "resize", "node count", "standby", "witness", "physical cluster", "逻辑集群", "物理集群", "トポロジ", "スケーリング", ], }, { "group": "non_diagnosis", "category": "inspection_or_report_generation", "keywords": ["巡检", "inspection", "health check", "daily inspection", "报告", "report", "weekly report", "周报"], }, { "group": "non_diagnosis", "category": "skill_or_workflow_setup", "keywords": ["skill", "skills", "安装", "重新安装", "加载", "opencode", "workflow", "runbook"], }, { "group": "non_diagnosis", "category": "eval_or_regression", "keywords": ["ws-eval", "eval", "regression", "system-unavailability", "incident-window"], }, { "group": "non_diagnosis", "category": "product_usage_or_admin", "keywords": ["feishu", "飞书", "权限", "admin", "配置页面", "usage report", "api key", "login"], }, { "group": "non_diagnosis", "category": "ticket_summary_or_customer_response", "keywords": ["回答", "回答案", "draft", "メール", "email", "escalate", "エスカレート", "客户回复", "总结", "review"], }, { "group": "non_diagnosis", "category": "fleet_inventory_or_capacity_analysis", "keywords": [ "所有 active", "active 的 essential", "cluster list", "basic info", "cluster basic", "storage size", "used storage", "tikv used storage", "row based storage", "表数量", "每日", "每天", "统计", "指标", "基本信息", "tenant", "region", "provider", "inventory", ], }, ] GENERIC_CASE_PATTERNS = ["tcid-", "tcoc-", "gtoc-", "o11y-", "st-", "tidb.atlassian.net/browse"] def load_message_category_manual_overrides() -> list[dict]: global _MESSAGE_CATEGORY_MANUAL_OVERRIDES if _MESSAGE_CATEGORY_MANUAL_OVERRIDES is not None: return _MESSAGE_CATEGORY_MANUAL_OVERRIDES if not MESSAGE_CATEGORY_MANUAL_OVERRIDES_PATH.exists(): _MESSAGE_CATEGORY_MANUAL_OVERRIDES = [] return _MESSAGE_CATEGORY_MANUAL_OVERRIDES payload = json.loads(MESSAGE_CATEGORY_MANUAL_OVERRIDES_PATH.read_text(encoding="utf-8")) overrides = [] for rule in payload.get("overrides") or []: if rule.get("disabled"): continue category = str(rule.get("category") or "").strip() pattern = str(rule.get("pattern") or "") match = str(rule.get("match") or "contains").strip() if not category or category not in MESSAGE_CATEGORY_ORDER: raise ValueError(f"invalid manual message category override category: {category!r}") if not pattern: raise ValueError("manual message category override pattern must not be empty") if match not in {"contains", "exact", "regex"}: raise ValueError(f"invalid manual message category override match type: {match!r}") overrides.append( { "match": match, "pattern": pattern, "category": category, "case_sensitive": bool(rule.get("case_sensitive")), } ) _MESSAGE_CATEGORY_MANUAL_OVERRIDES = overrides return _MESSAGE_CATEGORY_MANUAL_OVERRIDES def classify_manual_user_message(content: str) -> str | None: text = (content or "").strip() lowered = text.lower() for rule in load_message_category_manual_overrides(): pattern = rule["pattern"] target = text if rule["case_sensitive"] else lowered needle = pattern if rule["case_sensitive"] else pattern.lower() if rule["match"] == "exact" and target == needle: return rule["category"] if rule["match"] == "contains" and needle in target: return rule["category"] if rule["match"] == "regex": flags = 0 if rule["case_sensitive"] else re.IGNORECASE if re.search(pattern, text, flags): return rule["category"] return None def scenario_category_counts(items: list[dict]) -> dict[str, int]: counts: dict[str, int] = {} for item in items: category = item.get("scenario_category") if not category: continue counts[category] = counts.get(category, 0) + 1 return counts def classify_user_message(content: str) -> str: raw_text = (content or "").strip() text = raw_text.lower() if not text: return "未归类" manual_category = classify_manual_user_message(raw_text) if manual_category: return manual_category for category in MESSAGE_CATEGORY_ORDER: if category in {"背景补充", "未归类"}: continue if any(keyword.lower() in text for keyword in MESSAGE_CATEGORY_KEYWORDS.get(category, [])): return category if is_evidence_lookup_question(text): return "Ad hoc 查询" if is_question_like(text): return "解释追问" if is_background_context_message(raw_text): return "背景补充" return "未归类" def is_question_like(text: str) -> bool: if any(marker in text for marker in QUESTION_MARKERS): return True return bool(re.search(r"[ぁ-んァ-ン一-龯](か|でしょう|ですか|ますか)\s*[。.!！]*$", text)) def is_evidence_lookup_question(text: str) -> bool: return is_question_like(text) and any(term in text for term in EVIDENCE_QUERY_TERMS) def is_background_context_message(content: str) -> bool: text = (content or "").strip() lowered = text.lower() if not text: return False if URL_RE.search(text): return True if "`" in text: return True if NUMERIC_ID_RE.match(text): return True if LABELLED_ID_RE.search(text) and re.search(r"\d{6,}", text): return True if CASE_ID_RE.search(text): return True if FILE_REF_RE.search(text): return True if API_KEY_LIKE_RE.match(text): return True if TIME_RANGE_RE.search(text) and len(text) <= 120: return True if TECHNICAL_IDENTIFIER_RE.match(text): return True if SNAKE_IDENTIFIER_RE.match(text): return True if MENTION_ONLY_RE.match(text): return True if DOTTED_IDENTIFIER_RE.match(text): return True if text.lstrip().startswith(">"): return True if "```" in text or SQL_RE.search(text): return True if any(keyword in lowered for keyword in BACKGROUND_CONTEXT_KEYWORDS): return True if len(text.splitlines()) >= 3: return True if len(text) >= 80 and any(token in lowered for token in ["sql", "log", "metric", "grafana", "cluster", "workspace"]): return True return False def classify_workspace_scenario(row: dict, user_messages: list[dict] | None = None) -> dict: title = str(row.get("title") or "") user_text = "\n".join(str((message or {}).get("content") or "") for message in (user_messages or [])) message_category_text = " ".join((row.get("message_categories") or {}).keys()) title_text = title.lower() body_text = f"{user_text} {message_category_text}".lower() full_text = f"{title_text}\n{body_text}" scored = [] for rule in SCENARIO_RULES: matched_terms = [] score = 0 for keyword in rule["keywords"]: term = keyword.lower() if term in title_text: score += 3 matched_terms.append(keyword) elif term in body_text: score += 1 matched_terms.append(keyword) if score: scored.append( { "scenario_group": rule["group"], "scenario_category": rule["category"], "score": score, "matched_terms": sorted(set(matched_terms), key=str.lower), } ) scored.sort(key=lambda item: item["score"], reverse=True) diagnostic_hits = [item for item in scored if item["scenario_group"] == "diagnosis" and item["score"] >= 2] if len(diagnostic_hits) >= 3 and diagnostic_hits[1]["score"] >= max(2, diagnostic_hits[0]["score"] * 0.55): secondary_categories = [ { "scenario_group": item["scenario_group"], "scenario_category": item["scenario_category"], "score": item["score"], "matched_terms": item["matched_terms"][:8], } for item in diagnostic_hits[:5] ] return { "scenario_group": "diagnosis", "scenario_category": diagnostic_hits[0]["scenario_category"], "scenario_category_counts": scenario_category_counts(secondary_categories), "scenario_confidence": 0.62, "scenario_classification_method": "score_based_title_and_user_messages_v2", "scenario_secondary_categories": secondary_categories, "scenario_evidence": { "matched_terms": sorted({term for item in diagnostic_hits[:5] for term in item["matched_terms"]}, key=str.lower)[:16], "message_count_used": len(user_messages or []), "review_required": True, }, } if scored: top = scored[0] secondary = [ { "scenario_group": item["scenario_group"], "scenario_category": item["scenario_category"], "score": item["score"], "matched_terms": item["matched_terms"][:8], } for item in scored[1:6] if item["score"] >= 2 ] top_score = int(top["score"]) confidence = 0.9 if top_score >= 8 else 0.78 if top_score >= 4 else 0.62 if secondary and secondary[0]["score"] >= top_score * 0.8: confidence = min(confidence, 0.68) return { "scenario_group": top["scenario_group"], "scenario_category": top["scenario_category"], "scenario_category_counts": {top["scenario_category"]: 1}, "scenario_confidence": confidence, "scenario_classification_method": "score_based_title_and_user_messages_v2", "scenario_secondary_categories": secondary, "scenario_evidence": { "matched_terms": top["matched_terms"][:12], "message_count_used": len(user_messages or []), "review_required": confidence < 0.7, }, } if any(pattern in full_text for pattern in GENERIC_CASE_PATTERNS): return { "scenario_group": "non_diagnosis", "scenario_category": "support_case_triage_or_customer_response", "scenario_category_counts": {"support_case_triage_or_customer_response": 1}, "scenario_confidence": 0.55, "scenario_classification_method": "score_based_title_and_user_messages_v2", "scenario_secondary_categories": [], "scenario_evidence": { "matched_terms": [pattern for pattern in GENERIC_CASE_PATTERNS if pattern in full_text][:8], "message_count_used": len(user_messages or []), "review_required": True, }, } return { "scenario_group": "unknown", "scenario_category": None, "scenario_category_counts": {}, "scenario_confidence": 0.2, "scenario_classification_method": "score_based_title_and_user_messages_v2", "scenario_secondary_categories": [], "scenario_evidence": { "matched_terms": [], "message_count_used": len(user_messages or []), "review_required": True, }, } def percentile(values: list[float], pct: float) -> float: if not values: return 0.0 ordered = sorted(values) if len(ordered) == 1: return ordered[0] pos = (len(ordered) - 1) * pct low = math.floor(pos) high = math.ceil(pos) if low == high: return ordered[int(pos)] return ordered[low] * (high - pos) + ordered[high] * (pos - low) def coalesce_daily_workspace_rows(rows: list[dict]) -> list[dict]: """Count workspace-wide messages and usage once across billing channels.""" grouped = defaultdict(list) for row in rows: grouped[row["workspace_id"]].append(row) result = [] for contributions in grouped.values(): # Keep the largest contribution's user/channel as representative metadata. merged = dict(max(contributions, key=lambda row: float(row.get("cost") or 0))) by_channel = defaultdict(float) by_user = defaultdict(float) for row in contributions: amount = float(row.get("cost") or 0) by_channel[row.get("channel") or "unknown"] += amount by_user[row.get("user_email") or ""] += amount merged["cost"] = round_money(sum(by_channel.values())) merged["cost_by_channel"] = {key: round_money(value) for key, value in sorted(by_channel.items())} merged["cost_by_user"] = {key: round_money(value) for key, value in sorted(by_user.items())} result.append(merged) return sorted(result, key=lambda row: float(row.get("cost") or 0), reverse=True) def build_daily_report(day: date, prom: dict, sqlite_summary: dict, mapping: dict[str, dict]) -> dict: start, end = local_window(day) day_s = day.isoformat() workspace_meta = sqlite_summary.get("workspace_meta") or {} msg_by_ws = sqlite_summary.get("messages", {}).get(day_s, {}) or {} user_messages_by_ws = sqlite_summary.get("user_messages", {}).get(day_s, {}) or {} all_user_messages_by_ws = sqlite_summary.get("all_user_messages", {}) or {} scenario_messages_by_ws = sqlite_summary.get("scenario_messages", {}).get(day_s, {}) or {} all_scenario_messages_by_ws = sqlite_summary.get("all_scenario_messages", {}) or {} token_by_ws = sqlite_summary.get("tokens", {}).get(day_s, {}) or {} token_workspace_rows = sqlite_summary.get("token_workspace_rows", {}).get(day_s, []) or [] created_ws = set((sqlite_summary.get("created_workspaces", {}).get(day_s) or [])) cost_by_user = defaultdict(float) interactions_by_user = defaultdict(float) channels_by_user = defaultdict(lambda: defaultdict(float)) workspace_rows = {} for row in token_workspace_rows: workspace_id = (row.get("workspace_id") or "").strip() if not workspace_id: continue cost = float(row.get("cost_amount") or 0) if cost <= 0: continue meta = workspace_meta.get(workspace_id) or {} email = (row.get("user_email") or meta.get("owner_email") or "").strip().lower() channel = (row.get("channel") or "unknown").strip() or "unknown" if email: cost_by_user[email] += cost key = (workspace_id, email, channel) workspace_rows[key] = workspace_rows.get(key, 0.0) + cost for row in prom["user_interactions"]: email = (row["metric"].get("user_email") or "").strip().lower() if email and row["value"] > 0: interactions_by_user[email] += row["value"] for row in prom["user_channel_interactions"]: email = (row["metric"].get("user_email") or "").strip().lower() channel = (row["metric"].get("channel") or "unknown").strip() or "unknown" if email and row["value"] > 0: channels_by_user[email][channel] += row["value"] workspace_ids = {key[0] for key in workspace_rows} | set(msg_by_ws) | set(token_by_ws) | created_ws workspace_category_counts: dict[str, dict[str, int]] = {} for workspace_id, messages_for_ws in user_messages_by_ws.items(): counts = dict.fromkeys(MESSAGE_CATEGORY_ORDER, 0) for message in messages_for_ws or []: category = classify_user_message(str((message or {}).get("content") or "")) counts[category] = counts.get(category, 0) + 1 workspace_category_counts[workspace_id] = {key: value for key, value in counts.items() if value} users = [] for email in sorted(set(cost_by_user) | set(interactions_by_user) | set(channels_by_user)): mapped = team_for(email, mapping) user_workspace_ids = {workspace_id for (workspace_id, user_email, _channel) in workspace_rows if user_email == email} user_msg = sum((msg_by_ws.get(ws) or {}).get("message_count", 0) for ws in user_workspace_ids) user_user_msg = sum((msg_by_ws.get(ws) or {}).get("user_message_count", 0) for ws in user_workspace_ids) users.append( { "user_email": email, "name": mapped["name"], "en_name": mapped["en_name"], "team_path": mapped["team_path"], "team": mapped["team"], "mapping_source": mapped["mapping_source"], "cost": round_money(cost_by_user[email]), "acp_interactions": int(round(interactions_by_user[email])), "workspace_count": len(user_workspace_ids), "message_count": user_msg, "user_message_count": user_user_msg, "channel_breakdown": {k: int(round(v)) for k, v in sorted(channels_by_user[email].items())}, "frequency_bucket": None, "depth_bucket": None, "behavior_mode": None, } ) frequency_values = [float(user["workspace_count"] or 0) for user in users] depth_values = [ float(user["acp_interactions"] or 0) / float(user["workspace_count"] or 1) for user in users if int(user["workspace_count"] or 0) > 0 ] frequency_threshold = max(2.0, percentile(frequency_values, 0.6)) depth_threshold = max(4.0, percentile(depth_values, 0.6)) for user in users: workspace_count_for_user = int(user["workspace_count"] or 0) acp_for_user = int(user["acp_interactions"] or 0) avg_acp_per_ws = acp_for_user / workspace_count_for_user if workspace_count_for_user else 0.0 high_frequency = workspace_count_for_user >= frequency_threshold high_depth = avg_acp_per_ws >= depth_threshold user["avg_acp_per_ws"] = round(avg_acp_per_ws, 4) user["frequency_bucket"] = "高频" if high_frequency else "低频" user["depth_bucket"] = "高轮次" if high_depth else "低轮次" if high_frequency and high_depth: user["behavior_mode"] = "高频高轮次" elif high_frequency and not high_depth: user["behavior_mode"] = "高频低轮次" elif not high_frequency and high_depth: user["behavior_mode"] = "低频高轮次" else: user["behavior_mode"] = "低频低轮次" workspaces = [] for (workspace_id, email, channel), cost in sorted(workspace_rows.items(), key=lambda item: item[1], reverse=True): meta = workspace_meta.get(workspace_id) or {} msg = msg_by_ws.get(workspace_id) or {} token = token_by_ws.get(workspace_id) or {} categories = workspace_category_counts.get(workspace_id) or {} mapped = team_for(email, mapping) scenario = classify_workspace_scenario( {"title": meta.get("name") or "", "message_categories": categories}, all_scenario_messages_by_ws.get(workspace_id) or scenario_messages_by_ws.get(workspace_id) or all_user_messages_by_ws.get(workspace_id) or user_messages_by_ws.get(workspace_id) or [], ) workspaces.append( { "workspace_id": workspace_id, "title": meta.get("name") or "", "url": f"https://investibook.com/workspace/{workspace_id}", "created_at": meta.get("created_at") or "", "updated_at": meta.get("updated_at") or "", "workspace_lifecycle": "new" if workspace_id in created_ws else "updated_existing", "user_email": email, "team_path": mapped["team_path"], "team": mapped["team"], "channel": channel, "cost": round_money(cost), "acp_interactions": None, "message_count": msg.get("message_count", 0), "user_message_count": msg.get("user_message_count", 0), "assistant_message_count": msg.get("assistant_message_count", 0), "token_usage": token, "message_categories": categories, "scenario_group": scenario["scenario_group"], "scenario_category": scenario["scenario_category"], "scenario_category_counts": scenario.get("scenario_category_counts") or {}, "scenario_confidence": scenario["scenario_confidence"], "scenario_classification_method": scenario["scenario_classification_method"], "scenario_secondary_categories": scenario["scenario_secondary_categories"], "scenario_evidence": scenario["scenario_evidence"], "is_inspection": None, "quality_notes": "", } ) # Include message-only workspaces so message totals do not disappear when recorded token cost is absent. cost_workspace_ids = {row["workspace_id"] for row in workspaces} for workspace_id in sorted((set(msg_by_ws) | set(token_by_ws)) - cost_workspace_ids): meta = workspace_meta.get(workspace_id) or {} msg = msg_by_ws.get(workspace_id) or {} token = token_by_ws.get(workspace_id) or {} categories = workspace_category_counts.get(workspace_id) or {} owner_email = (meta.get("owner_email") or "").strip().lower() mapped = team_for(owner_email, mapping) scenario = classify_workspace_scenario( {"title": meta.get("name") or "", "message_categories": categories}, all_scenario_messages_by_ws.get(workspace_id) or scenario_messages_by_ws.get(workspace_id) or all_user_messages_by_ws.get(workspace_id) or user_messages_by_ws.get(workspace_id) or [], ) workspaces.append( { "workspace_id": workspace_id, "title": meta.get("name") or "", "url": f"https://investibook.com/workspace/{workspace_id}", "created_at": meta.get("created_at") or "", "updated_at": meta.get("updated_at") or "", "workspace_lifecycle": "new" if workspace_id in created_ws else "updated_existing", "user_email": owner_email, "team_path": mapped["team_path"], "team": mapped["team"], "channel": "unknown", "cost": 0.0, "acp_interactions": None, "message_count": msg.get("message_count", 0), "user_message_count": msg.get("user_message_count", 0), "assistant_message_count": msg.get("assistant_message_count", 0), "token_usage": token, "message_categories": categories, "scenario_group": scenario["scenario_group"], "scenario_category": scenario["scenario_category"], "scenario_category_counts": scenario.get("scenario_category_counts") or {}, "scenario_confidence": scenario["scenario_confidence"], "scenario_classification_method": scenario["scenario_classification_method"], "scenario_secondary_categories": scenario["scenario_secondary_categories"], "scenario_evidence": scenario["scenario_evidence"], "is_inspection": None, "quality_notes": "", } ) workspaces = coalesce_daily_workspace_rows(workspaces) team_acc = {} for user in users: team_path = user["team_path"] or "(unknown)" team = user["team"] or "(unknown)" bucket = team_acc.setdefault( team_path, { "team_path": team_path, "team": team, "cost": 0.0, "acp_interactions": 0, "workspace_ids": set(), "user_emails": set(), "message_count": 0, "user_message_count": 0, }, ) bucket["cost"] += float(user["cost"] or 0) bucket["acp_interactions"] += int(user["acp_interactions"] or 0) bucket["user_emails"].add(user["user_email"]) for row in workspaces: team_path = row["team_path"] or "(unknown)" team = row["team"] or "(unknown)" bucket = team_acc.setdefault( team_path, { "team_path": team_path, "team": team, "cost": 0.0, "acp_interactions": 0, "workspace_ids": set(), "user_emails": set(), "message_count": 0, "user_message_count": 0, }, ) bucket["workspace_ids"].add(row["workspace_id"]) bucket["message_count"] += int(row.get("message_count") or 0) bucket["user_message_count"] += int(row.get("user_message_count") or 0) if row.get("user_email"): bucket["user_emails"].add(row["user_email"]) teams = [] for bucket in team_acc.values(): workspace_count = len(bucket["workspace_ids"]) acp = bucket["acp_interactions"] cost = bucket["cost"] teams.append( { "team_path": bucket["team_path"], "team": bucket["team"], "cost": round_money(cost), "acp_interactions": acp, "workspace_count": workspace_count, "active_users": len(bucket["user_emails"]), "message_count": bucket["message_count"], "user_message_count": bucket["user_message_count"], "cost_per_ws": round_money(cost / workspace_count) if workspace_count else None, "cost_per_acp": round_money(cost / acp) if acp else None, "acp_per_ws": round(acp / workspace_count, 4) if workspace_count else None, "message_per_ws": round(bucket["message_count"] / workspace_count, 4) if workspace_count else None, "message_per_acp": round(bucket["message_count"] / acp, 4) if acp else None, "correction_message_ratio": None, } ) teams.sort(key=lambda row: (row["cost"] or 0), reverse=True) total_cost = sum(float(row.get("cost_amount") or 0) for row in token_workspace_rows) message_count = sum((item or {}).get("message_count", 0) for item in msg_by_ws.values()) user_message_count = sum((item or {}).get("user_message_count", 0) for item in msg_by_ws.values()) assistant_message_count = sum((item or {}).get("assistant_message_count", 0) for item in msg_by_ws.values()) message_category_totals = dict.fromkeys(MESSAGE_CATEGORY_ORDER, 0) category_chars = {category: {"user_chars": 0} for category in MESSAGE_CATEGORY_ORDER} for workspace_id, messages_for_ws in user_messages_by_ws.items(): for message in messages_for_ws or []: content = str((message or {}).get("content") or "") category = classify_user_message(content) message_category_totals[category] = message_category_totals.get(category, 0) + 1 category_chars.setdefault(category, {"user_chars": 0})["user_chars"] += int((message or {}).get("content_chars") or len(content)) message_categories = [] for category in MESSAGE_CATEGORY_ORDER: count = int(message_category_totals.get(category) or 0) if count <= 0: continue message_categories.append( { "category": category, "user_message_count": count, "share": round(count / user_message_count, 6) if user_message_count else 0, "estimated_cost_by_turn": round_money(float(total_cost or 0) * count / user_message_count) if user_message_count else None, "estimated_cost_by_output_weight": None, "avg_user_chars": round(category_chars[category]["user_chars"] / count, 2), "avg_next_assistant_chars": None, "classification_method": "rule_based_keywords_v2", } ) unknown_users = [user for user in users if not user["team_path"]] unknown_cost = sum(float(user.get("cost") or 0) for user in unknown_users) summary_active_users = len( { user["user_email"] for user in users if (user.get("cost") or 0) > 0 or (user.get("acp_interactions") or 0) > 0 } ) summary_workspace_count = len({workspace_id for (workspace_id, _email, _channel) in workspace_rows}) new_workspace_count = sum(1 for row in workspaces if row.get("workspace_lifecycle") == "new") updated_existing_workspace_count = sum(1 for row in workspaces if row.get("workspace_lifecycle") == "updated_existing") return { "date": day_s, "timezone": "Asia/Shanghai", "window_start": iso(start), "window_end": iso(end), "granularity": "daily", "data_status": "complete", "source": { "kind": "grafana_prometheus_and_prod_sqlite", "grafana_datasource": "DevOps-prometheus", "grafana_datasource_id": DATASOURCE_ID, "prometheus_labels": { "cluster": "prod-devops", "namespace": "investibook", "job": "backend", "prometheus": "monitoring/o11y-vmagent", }, "prod_backend_sqlite": PROD_DB_PATH, "notes": [ "Cost comes from prod SQLite interaction_token_usage with runtime-aware accounting: OpenCode ses_* rows use cumulative session deltas; PI UUID and image-parse rows use per-row increments.", "workspace_count in summary is cost-derived distinct positive-cost workspaces.", "Prometheus is used for ACP interaction counts only; investigatory PI sessions made investibook_acp_session_cost unsuitable as the cost source for mixed-runtime weekly reporting.", "message categories are rule-based user-message classifications.", "scenario labels are added to split daily workspace_changes rows.", ], }, "summary": { "total_cost": round_money(total_cost), "total_acp_interactions": int(round(prom["total_interactions"])), "active_users": summary_active_users, "workspace_count": summary_workspace_count, "created_workspace_count": len(created_ws), "new_workspace_count": new_workspace_count, "updated_existing_workspace_count": updated_existing_workspace_count, "changed_workspace_count": len(workspaces), "message_count": message_count, "user_message_count": user_message_count, "assistant_message_count": assistant_message_count, "unknown_user_count": len(unknown_users), "unknown_cost": round_money(unknown_cost), }, "teams": teams, "users": sorted(users, key=lambda row: (row["cost"] or 0), reverse=True), "workspaces": workspaces, "message_categories": message_categories, "unmatched_users": [ { "user_email": user["user_email"], "cost": user["cost"], "interactions": user["acp_interactions"], "workspace_count": user["workspace_count"], "top_workspace": next((row["workspace_id"] for row in workspaces if row["user_email"] == user["user_email"]), ""), } for user in unknown_users ], "charts": { "daily_cost_acp": [ { "date": day_s, "cost": round_money(total_cost), "acp_interactions": int(round(prom["total_interactions"])), "workspace_count": summary_workspace_count, "active_users": summary_active_users, } ], "team_cost_bar": [ { "team": row["team"], "team_path": row["team_path"], "cost": row["cost"], "acp_interactions": row["acp_interactions"], "workspace_count": row["workspace_count"], "active_users": row["active_users"], } for row in teams ], "team_ratio_scatter": [ { "team": row["team"], "team_path": row["team_path"], "x": row["acp_per_ws"], "y": row["cost_per_acp"], "size": row["workspace_count"], "cost": row["cost"], } for row in teams ], "message_category_stack": [ { "team": row["team"] or "(unknown)", "category": category, "count": count, } for row in teams for category, count in sorted( { category: sum( int((workspace.get("message_categories") or {}).get(category) or 0) for workspace in workspaces if (workspace.get("team_path") or "(unknown)") == row["team_path"] ) for category in MESSAGE_CATEGORY_ORDER }.items() ) if count > 0 ], "user_behavior_quadrant": [ { "user_email": user["user_email"], "x": user["workspace_count"], "y": user["avg_acp_per_ws"], "cost": user["cost"], "mode": user["behavior_mode"], } for user in users ], }, } def collect_prometheus_day(session, day: date) -> dict: _start, end = local_window(day) total_interactions_expr = f'sum(increase(investibook_acp_session_interactions_total{{{PROM_LABELS}}}[24h]))' user_interactions_expr = f'topk(1000, sum by (user_email) (increase(investibook_acp_session_interactions_total{{{PROM_LABELS},user_email!="unknown"}}[24h])))' user_channel_expr = f'topk(2000, sum by (user_email,channel) (increase(investibook_acp_session_interactions_total{{{PROM_LABELS},user_email!="unknown"}}[24h])))' return { "total_interactions": prom_scalar(query_prometheus(session, total_interactions_expr, end)), "user_interactions": prom_vector(query_prometheus(session, user_interactions_expr, end)), "user_channel_interactions": prom_vector(query_prometheus(session, user_channel_expr, end)), } def collect_existing_prometheus_day(day: date) -> dict: daily_report_path = SPLIT_OUTPUT_ROOT / day.isoformat() / "daily_report.json" if not daily_report_path.exists(): return { "total_interactions": 0, "user_interactions": [], "user_channel_interactions": [], } existing = json.loads(daily_report_path.read_text(encoding="utf-8")) users = existing.get("users") or [] user_interactions = [] user_channel_interactions = [] for user in users: email = (user.get("user_email") or "").strip().lower() if not email: continue interactions = float(user.get("acp_interactions") or 0) if interactions > 0: user_interactions.append({"metric": {"user_email": email}, "value": interactions}) for channel, value in (user.get("channel_breakdown") or {}).items(): channel_interactions = float(value or 0) if channel_interactions > 0: user_channel_interactions.append( { "metric": { "user_email": email, "channel": (channel or "unknown").strip() or "unknown", }, "value": channel_interactions, } ) return { "total_interactions": float(((existing.get("summary") or {}).get("total_acp_interactions")) or 0), "user_interactions": user_interactions, "user_channel_interactions": user_channel_interactions, } def write_json(path: Path, data: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def merge_token_usage(target: dict, source: dict) -> None: for key in [ "interaction_records", "input_tokens", "output_tokens", "reasoning_tokens", "cache_read_tokens", "cache_write_tokens", ]: target[key] = int(target.get(key) or 0) + int((source or {}).get(key) or 0) target["cost_amount"] = round_money(float(target.get("cost_amount") or 0) + float((source or {}).get("cost_amount") or 0)) def build_workspace_usage_window(reports: list[dict], start_day: date, end_day: date) -> dict: by_workspace: dict[str, dict] = {} for report in reports: day = report["date"] for row in report.get("workspaces") or []: workspace_id = row["workspace_id"] bucket = by_workspace.setdefault( workspace_id, { "workspace_id": workspace_id, "title": row.get("title") or "", "url": row.get("url") or "", "user_emails": set(), "team_paths": set(), "teams": set(), "channels": set(), "first_active_date": day, "last_active_date": day, "active_dates": set(), "total_cost": 0.0, "message_count": 0, "user_message_count": 0, "assistant_message_count": 0, "token_usage": { "interaction_records": 0, "input_tokens": 0, "output_tokens": 0, "reasoning_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0, "cost_amount": 0.0, }, "message_categories": dict.fromkeys(MESSAGE_CATEGORY_ORDER, 0), "daily": [], "scenario_category": None, "is_inspection": None, "quality_notes": "", }, ) if row.get("title") and not bucket["title"]: bucket["title"] = row["title"] if row.get("url") and not bucket["url"]: bucket["url"] = row["url"] if row.get("user_email"): bucket["user_emails"].add(row["user_email"]) if row.get("team_path"): bucket["team_paths"].add(row["team_path"]) if row.get("team"): bucket["teams"].add(row["team"]) if row.get("channel"): bucket["channels"].add(row["channel"]) bucket["first_active_date"] = min(bucket["first_active_date"], day) bucket["last_active_date"] = max(bucket["last_active_date"], day) bucket["active_dates"].add(day) bucket["total_cost"] += float(row.get("cost") or 0) bucket["message_count"] += int(row.get("message_count") or 0) bucket["user_message_count"] += int(row.get("user_message_count") or 0) bucket["assistant_message_count"] += int(row.get("assistant_message_count") or 0) merge_token_usage(bucket["token_usage"], row.get("token_usage") or {}) for category, count in (row.get("message_categories") or {}).items(): bucket["message_categories"][category] = int(bucket["message_categories"].get(category) or 0) + int(count or 0) bucket["daily"].append( { "date": day, "user_email": row.get("user_email") or "", "team_path": row.get("team_path") or "", "team": row.get("team") or "", "channel": row.get("channel") or "unknown", "cost": row.get("cost") or 0, "message_count": row.get("message_count") or 0, "user_message_count": row.get("user_message_count") or 0, "assistant_message_count": row.get("assistant_message_count") or 0, "token_usage": row.get("token_usage") or {}, "message_categories": row.get("message_categories") or {}, } ) workspaces = [] for bucket in by_workspace.values(): active_dates = sorted(bucket.pop("active_dates")) message_categories = {key: value for key, value in bucket.pop("message_categories").items() if value} user_emails = sorted(bucket.pop("user_emails")) team_paths = sorted(bucket.pop("team_paths")) teams = sorted(bucket.pop("teams")) channels = sorted(bucket.pop("channels")) total_cost = float(bucket.pop("total_cost") or 0) user_message_count = int(bucket.get("user_message_count") or 0) workspace = { **bucket, "user_emails": user_emails, "primary_user_email": user_emails[0] if len(user_emails) == 1 else "", "team_paths": team_paths, "teams": teams, "channels": channels, "active_dates": active_dates, "active_days": len(active_dates), "total_cost": round_money(total_cost), "message_categories": message_categories, "cost_per_user_message": round_money(total_cost / user_message_count) if user_message_count else None, } workspace["daily"].sort(key=lambda item: item["date"]) workspaces.append(workspace) workspaces.sort(key=lambda item: (item["total_cost"] or 0), reverse=True) return { "start": start_day.isoformat(), "end": end_day.isoformat(), "timezone": "Asia/Shanghai", "granularity": "workspace_window", "source": { "kind": "derived_from_daily_usage_reports", "daily_report_glob": "outputs/usage_reports/YYYY-MM-DD/derived_daily_usage.json", "notes": [ "Workspace cost comes from prod SQLite interaction_token_usage with runtime-aware accounting.", "Workspace message categories are rule-based counts over user messages.", "is_inspection and quality_notes are reserved for later review passes.", ], }, "summary": { "workspace_count": len(workspaces), "total_cost": round_money(sum(float(item.get("total_cost") or 0) for item in workspaces)), "message_count": sum(int(item.get("message_count") or 0) for item in workspaces), "user_message_count": sum(int(item.get("user_message_count") or 0) for item in workspaces), }, "workspaces": workspaces, "charts": { "top_workspace_cost": [ { "workspace_id": item["workspace_id"], "title": item["title"], "cost": item["total_cost"], "message_count": item["message_count"], "user_message_count": item["user_message_count"], } for item in workspaces[:50] ], "workspace_message_depth": [ { "workspace_id": item["workspace_id"], "x": item["user_message_count"], "y": item["message_count"], "size": item["total_cost"], "title": item["title"], } for item in workspaces ], }, } def workspace_change_row(row: dict) -> dict: scenario = ( { "scenario_group": row.get("scenario_group"), "scenario_category": row.get("scenario_category"), "scenario_category_counts": row.get("scenario_category_counts") or {}, "scenario_confidence": row.get("scenario_confidence"), "scenario_classification_method": row.get("scenario_classification_method"), "scenario_secondary_categories": row.get("scenario_secondary_categories") or [], "scenario_evidence": row.get("scenario_evidence") or {}, } if row.get("scenario_group") or row.get("scenario_classification_method") else classify_workspace_scenario(row) ) daily_delta = { "cost": row.get("cost") or 0, "message_count": row.get("message_count") or 0, "user_message_count": row.get("user_message_count") or 0, "assistant_message_count": row.get("assistant_message_count") or 0, "token_usage": row.get("token_usage") or {}, "message_categories": row.get("message_categories") or {}, } return { "workspace_id": row.get("workspace_id") or "", "title": row.get("title") or "", "url": row.get("url") or "", "workspace_lifecycle": row.get("workspace_lifecycle") or "updated_existing", "created_at": row.get("created_at") or "", "updated_at": row.get("updated_at") or "", "user_email": row.get("user_email") or "", "team_path": row.get("team_path") or "", "team": row.get("team") or "", "channel": row.get("channel") or "unknown", "cost_by_channel": row.get("cost_by_channel") or {}, "cost_by_user": row.get("cost_by_user") or {}, "scenario_group": scenario["scenario_group"], "scenario_category": scenario["scenario_category"], "scenario_category_counts": scenario.get("scenario_category_counts") or {}, "scenario_confidence": scenario["scenario_confidence"], "scenario_classification_method": scenario["scenario_classification_method"], "scenario_secondary_categories": scenario.get("scenario_secondary_categories") or [], "scenario_evidence": scenario.get("scenario_evidence") or {}, "is_inspection": row.get("is_inspection"), "quality_notes": row.get("quality_notes") or "", "daily_delta": daily_delta, } def build_split_daily_outputs(report: dict) -> tuple[dict, dict, dict, dict]: workspaces = [workspace_change_row(row) for row in report.get("workspaces") or []] new_workspaces = [row for row in workspaces if row["workspace_lifecycle"] == "new"] updated_existing_workspaces = [row for row in workspaces if row["workspace_lifecycle"] == "updated_existing"] all_changed_workspaces = sorted( workspaces, key=lambda row: (float(row["daily_delta"].get("cost") or 0), int(row["daily_delta"].get("user_message_count") or 0)), reverse=True, ) top_changed_workspaces = [ { "workspace_id": row["workspace_id"], "title": row["title"], "workspace_lifecycle": row["workspace_lifecycle"], "user_email": row["user_email"], "team_path": row["team_path"], "scenario_group": row["scenario_group"], "scenario_category": row["scenario_category"], "cost": row["daily_delta"]["cost"], "message_count": row["daily_delta"]["message_count"], "user_message_count": row["daily_delta"]["user_message_count"], } for row in all_changed_workspaces[:20] ] daily_summary = { "date": report["date"], "timezone": report["timezone"], "window_start": report["window_start"], "window_end": report["window_end"], "granularity": report["granularity"], "data_status": report["data_status"], "summary": report["summary"], "source": report["source"], } workspace_changes = { "date": report["date"], "timezone": report["timezone"], "window_start": report["window_start"], "window_end": report["window_end"], "granularity": "daily_workspace_changes", "summary": { "new_workspace_count": len(new_workspaces), "updated_existing_workspace_count": len(updated_existing_workspaces), "changed_workspace_count": len(all_changed_workspaces), }, "new_workspaces": new_workspaces, "updated_existing_workspaces": updated_existing_workspaces, "all_changed_workspaces": all_changed_workspaces, } daily_report = { "date": report["date"], "timezone": report["timezone"], "window_start": report["window_start"], "window_end": report["window_end"], "granularity": report["granularity"], "data_status": report["data_status"], "source": report["source"], "summary": report["summary"], "teams": report["teams"], "users": report["users"], "message_categories": report["message_categories"], "unmatched_users": report["unmatched_users"], "top_changed_workspaces": top_changed_workspaces, "charts": report["charts"], "workspace_changes_ref": "workspace_changes.json", } chart_ready = {"date": report["date"], "charts": report["charts"]} return daily_summary, daily_report, workspace_changes, chart_ready def rebuild_split_daily_series() -> dict: series = [] if SPLIT_OUTPUT_ROOT.exists(): for day_dir in sorted(path for path in SPLIT_OUTPUT_ROOT.iterdir() if path.is_dir()): summary_path = day_dir / "daily_summary.json" if not summary_path.exists(): continue summary = json.loads(summary_path.read_text(encoding="utf-8")) summary_values = summary.get("summary") or {} series.append( { "date": summary["date"], "cost": summary_values.get("total_cost") or 0, "acp_interactions": summary_values.get("total_acp_interactions") or 0, "workspace_count": summary_values.get("workspace_count") or 0, "active_users": summary_values.get("active_users") or 0, } ) dates = [item["date"] for item in series] return { "start": min(dates) if dates else None, "end": max(dates) if dates else None, "timezone": "Asia/Shanghai", "series": series, } def rebuild_split_workspace_index() -> dict: daily = [] if SPLIT_OUTPUT_ROOT.exists(): for day_dir in sorted(path for path in SPLIT_OUTPUT_ROOT.iterdir() if path.is_dir()): changes_path = day_dir / "workspace_changes.json" if not changes_path.exists(): continue changes = json.loads(changes_path.read_text(encoding="utf-8")) daily.append( { "date": changes["date"], "workspace_changes_ref": f"{changes['date']}/workspace_changes.json", "daily_report_ref": f"{changes['date']}/daily_report.json", "summary": changes.get("summary") or {}, } ) dates = [item["date"] for item in daily] return { "start": min(dates) if dates else None, "end": max(dates) if dates else None, "timezone": "Asia/Shanghai", "granularity": "daily_workspace_index", "daily": daily, "notes": [ "Use daily workspace_changes.json files as the appendable source of truth.", "Build workspace-window aggregates on demand from the daily refs in this index.", ], } def iter_split_workspace_change_files(): if not SPLIT_OUTPUT_ROOT.exists(): return for day_dir in sorted(path for path in SPLIT_OUTPUT_ROOT.iterdir() if path.is_dir()): changes_path = day_dir / "workspace_changes.json" if changes_path.exists(): yield changes_path def aggregate_split_workspaces() -> list[dict]: by_workspace: dict[str, dict] = {} for changes_path in iter_split_workspace_change_files() or []: changes = json.loads(changes_path.read_text(encoding="utf-8")) day = changes["date"] rows = changes.get("all_changed_workspaces") or [] for row in rows: workspace_id = row.get("workspace_id") or "" if not workspace_id: continue bucket = by_workspace.setdefault( workspace_id, { "workspace_id": workspace_id, "title": row.get("title") or "", "url": row.get("url") or "", "user_emails": set(), "team_paths": set(), "teams": set(), "channels": set(), "created_at": row.get("created_at") or "", "first_active_date": day, "last_active_date": day, "active_dates": set(), "new_dates": set(), "updated_existing_dates": set(), "total_cost": 0.0, "message_count": 0, "user_message_count": 0, "assistant_message_count": 0, "message_categories": dict.fromkeys(MESSAGE_CATEGORY_ORDER, 0), }, ) if row.get("title") and not bucket["title"]: bucket["title"] = row["title"] if row.get("url") and not bucket["url"]: bucket["url"] = row["url"] if row.get("created_at") and not bucket["created_at"]: bucket["created_at"] = row["created_at"] for key, target in [ ("user_email", "user_emails"), ("team_path", "team_paths"), ("team", "teams"), ("channel", "channels"), ]: if row.get(key): bucket[target].add(row[key]) bucket["first_active_date"] = min(bucket["first_active_date"], day) bucket["last_active_date"] = max(bucket["last_active_date"], day) bucket["active_dates"].add(day) if row.get("workspace_lifecycle") == "new": bucket["new_dates"].add(day) else: bucket["updated_existing_dates"].add(day) delta = row.get("daily_delta") or {} bucket["total_cost"] += float(delta.get("cost") or 0) bucket["message_count"] += int(delta.get("message_count") or 0) bucket["user_message_count"] += int(delta.get("user_message_count") or 0) bucket["assistant_message_count"] += int(delta.get("assistant_message_count") or 0) for category, count in (delta.get("message_categories") or {}).items(): bucket["message_categories"][category] = int(bucket["message_categories"].get(category) or 0) + int(count or 0) workspaces = [] for bucket in by_workspace.values(): active_dates = sorted(bucket.pop("active_dates")) new_dates = sorted(bucket.pop("new_dates")) updated_existing_dates = sorted(bucket.pop("updated_existing_dates")) message_categories = {key: value for key, value in bucket.pop("message_categories").items() if value} total_cost = float(bucket.pop("total_cost") or 0) workspace = { **bucket, "user_emails": sorted(bucket.pop("user_emails")), "team_paths": sorted(bucket.pop("team_paths")), "teams": sorted(bucket.pop("teams")), "channels": sorted(bucket.pop("channels")), "active_dates": active_dates, "new_dates": new_dates, "updated_existing_dates": updated_existing_dates, "active_days": len(active_dates), "total_cost": round_money(total_cost), "message_categories": message_categories, } workspaces.append(workspace) workspaces.sort(key=lambda item: (item.get("last_active_date") or "", item.get("total_cost") or 0), reverse=True) return workspaces def build_workspace_mode(row: dict) -> str: active_days = int(row.get("active_days") or 0) user_messages = int(row.get("user_message_count") or 0) total_cost = float(row.get("total_cost") or 0) has_updates = bool(row.get("updated_existing_dates")) if active_days >= 3 or has_updates: return "持续迭代" if user_messages >= 8 or total_cost >= 5: return "深度单次诊断" if user_messages <= 2 and total_cost < 1: return "轻量尝试" return "单次任务" def rebuild_workspace_mode_classification(workspaces: list[dict]) -> dict: rows = [] mode_counts = defaultdict(int) for workspace in workspaces: mode = build_workspace_mode(workspace) mode_counts[mode] += 1 rows.append( { "workspace_id": workspace["workspace_id"], "title": workspace.get("title") or "", "url": workspace.get("url") or "", "mode": mode, "classification_method": "rule_based_workspace_activity_v1", "signals": { "active_days": workspace.get("active_days") or 0, "active_dates": workspace.get("active_dates") or [], "updated_existing_dates": workspace.get("updated_existing_dates") or [], "total_cost": workspace.get("total_cost") or 0, "message_count": workspace.get("message_count") or 0, "user_message_count": workspace.get("user_message_count") or 0, "message_categories": workspace.get("message_categories") or {}, }, } ) return { "timezone": "Asia/Shanghai", "granularity": "workspace", "data_status": "derived", "summary": { "workspace_count": len(rows), "mode_counts": dict(sorted(mode_counts.items())), }, "workspaces": rows, } def build_profile_spend_by_usedby(end_day: date) -> dict: existing_path = SPLIT_OUTPUT_ROOT / "profile_spend_by_usedby.json" month = end_day.strftime("%Y-%m") try: from pycookiecheat import chrome_cookies session = requests.Session() for key, value in chrome_cookies(HELIX_BASE_URL, browser="chrome").items(): session.cookies.set(key, value, domain="helix.pingcap.net") refresh = session.post(f"{HELIX_BASE_URL}/api/v1/auth/refresh", timeout=15) refresh.raise_for_status() token = ((refresh.json() or {}).get("data") or {}).get("access_token") if not token: raise RuntimeError("Helix refresh did not return an access token") detail = session.get( f"{HELIX_BASE_URL}/api/v1/apikeys/assignments/{HELIX_ASSIGNMENT_ID}/spend/detail", params={"month": month}, headers={"Authorization": f"Bearer {token}"}, timeout=20, ) detail.raise_for_status() data = (detail.json() or {}).get("data") or {} profiles = [] for row in data.get("by_profile") or []: profiles.append( { "profile": row.get("profile") or "", "used_by": row.get("profile") or "", "spend": round_money(float(row.get("mtd_usd") or 0)), "share": round(float(row.get("percent_of_month") or 0), 4), "by_model": [ { "model": item.get("model") or "", "spend": round_money(float(item.get("mtd_usd") or 0)), "share_of_profile": round(float(item.get("percent_of_profile") or 0), 4), } for item in row.get("by_model") or [] ], } ) return { "timezone": "Asia/Shanghai", "granularity": "monthly_to_date", "data_status": "complete" if data.get("data_available") else "unavailable", "month": data.get("month") or month, "assignment_id": HELIX_ASSIGNMENT_ID, "assignment_url": f"{HELIX_BASE_URL}/me/api-keys/{HELIX_ASSIGNMENT_ID}", "source": { "kind": "helix_apikey_assignment_spend_detail", "endpoint": "/api/v1/apikeys/assignments/{assignment_id}/spend/detail", "last_fact_date": data.get("last_fact_date"), }, "summary": { "total_spend": round_money(float(data.get("mtd_usd") or 0)), "budget": round_money(float(data.get("budget_usd") or 0)), "budget_utilization_percent": round(float(data.get("percent") or 0), 4), "data_available": bool(data.get("data_available")), }, "profiles": profiles, "daily_series": [ { "date": item.get("date") or "", "spend": round_money(float(item.get("usd") or 0)), } for item in data.get("daily_series") or [] ], "charts": { "profile_spend_stack": [ { "profile": row["profile"], "used_by": row["used_by"], "spend": row["spend"], "share": row["share"], } for row in profiles ], }, } except Exception as exc: error_type = type(exc).__name__ if existing_path.exists(): try: existing = json.loads(existing_path.read_text(encoding="utf-8")) existing["data_status"] = existing.get("data_status") or "stale_preserved" existing.setdefault("source", {}).setdefault("notes", []).append( f"Helix refresh failed during collector run; preserved existing data. Error type: {error_type}." ) return existing except json.JSONDecodeError: pass return { "timezone": "Asia/Shanghai", "granularity": "monthly_to_date", "data_status": "not_collected", "month": month, "assignment_id": HELIX_ASSIGNMENT_ID, "assignment_url": f"{HELIX_BASE_URL}/me/api-keys/{HELIX_ASSIGNMENT_ID}", "source": { "kind": "helix_apikey_assignment_spend_detail", "notes": [ f"Helix fetch failed with {error_type}; open the assignment URL in Chrome and rerun.", ], }, "summary": { "total_spend": 0, "budget": 0, "budget_utilization_percent": 0, "data_available": False, }, "profiles": [], "daily_series": [], "charts": { "profile_spend_stack": [], }, } def build_feedback_demands() -> dict: return { "timezone": "Asia/Shanghai", "granularity": "item", "data_status": "not_required_by_current_report", "source": { "kind": "manual_lark_feedback_or_demand_list", "notes": [ "Demand/feedback items are not part of the prod usage JSON path.", "Keep this file empty until the report needs a curated feedback list.", ], }, "demands": [], "charts": { "demand_status": [], "demand_category": [], }, } def main() -> int: parser = argparse.ArgumentParser(description="Collect real Investibook daily usage reports.") parser.add_argument("--start", default="2026-06-25", help="Start date in Asia/Shanghai, inclusive") parser.add_argument("--end", default="2026-07-06", help="End date in Asia/Shanghai, inclusive") parser.add_argument("--profile", default="Default", help="Chrome profile for Clinic/Grafana cookies") parser.add_argument( "--reuse-existing-prometheus", action="store_true", help="Reuse ACP interaction counts from existing split daily JSON instead of querying Grafana.", ) args = parser.parse_args() start_day = parse_day(args.start) end_day = parse_day(args.end) mapping = load_team_mapping() pod = find_backend_pod() sqlite_summary = load_prod_sqlite_summary(start_day, end_day, pod) session = None if not args.reuse_existing_prometheus: session = grafana_session(GRAFANA_BASE_URL, profile_name=args.profile, path_prefix=GRAFANA_PATH_PREFIX) combined = [] reports = [] for day in iter_days(start_day, end_day): prom = ( collect_existing_prometheus_day(day) if args.reuse_existing_prometheus else collect_prometheus_day(session, day) ) report = build_daily_report(day, prom, sqlite_summary, mapping) reports.append(report) day_dir = OUTPUT_ROOT / day.isoformat() write_json(day_dir / "derived_daily_usage.json", report) write_json( day_dir / "daily_summary.json", { "date": report["date"], "timezone": report["timezone"], "window_start": report["window_start"], "window_end": report["window_end"], "granularity": report["granularity"], "data_status": report["data_status"], "summary": report["summary"], "source": report["source"], }, ) write_json(day_dir / "chart_ready_usage.json", {"date": report["date"], "charts": report["charts"]}) write_json(day_dir / "daily_workspace_detail.json", {"date": report["date"], "workspaces": report["workspaces"]}) split_day_dir = SPLIT_OUTPUT_ROOT / day.isoformat() daily_summary, daily_report, workspace_changes, chart_ready = build_split_daily_outputs(report) write_json(split_day_dir / "daily_summary.json", daily_summary) write_json(split_day_dir / "daily_report.json", daily_report) write_json(split_day_dir / "workspace_changes.json", workspace_changes) write_json(split_day_dir / "chart_ready_usage.json", chart_ready) combined.append(report["charts"]["daily_cost_acp"][0]) daily_series = { "start": start_day.isoformat(), "end": end_day.isoformat(), "timezone": "Asia/Shanghai", "series": combined, } workspace_usage = build_workspace_usage_window(reports, start_day, end_day) workspace_usage["source"]["daily_report_glob"] = "outputs/usage_daily_reports/YYYY-MM-DD/daily_report.json" write_json(OUTPUT_ROOT / "daily_cost_acp_series.json", daily_series) write_json(OUTPUT_ROOT / f"workspace_usage_{start_day.isoformat()}_to_{end_day.isoformat()}.json", workspace_usage) write_json(SPLIT_OUTPUT_ROOT / "daily_cost_acp_series.json", rebuild_split_daily_series()) write_json(SPLIT_OUTPUT_ROOT / "workspace_index.json", rebuild_split_workspace_index()) split_workspaces = aggregate_split_workspaces() write_json(SPLIT_OUTPUT_ROOT / "workspace_mode_classification.json", rebuild_workspace_mode_classification(split_workspaces)) write_json(SPLIT_OUTPUT_ROOT / "profile_spend_by_usedby.json", build_profile_spend_by_usedby(end_day)) write_json(SPLIT_OUTPUT_ROOT / "feedback_demands.json", build_feedback_demands()) return 0 if __name__ == "__main__": raise SystemExit(main())