YANG JIANKUAN
110f710ff0
- 版面重排：主体左 Claude Usage（5h／7d／Fable 7d 三行），右 ChatGPT Usage（目前 mock，标「示例数据」）；每行为一张用量卡：标签＋F24 百分比＋pace／全宽进度条／「重置 HH:MM」＋「已过 NN%」 - 移除上一版的橘喵今日实时与 30 天趋势图及 trend totals 合计（业务数据留给 400×300 小屏），绿色暂不再使用 - data 用量结构新增 scoped（按模型限定额度）、reset（重置时刻文本）、updated（快照落盘时刻），新增 _usage_gpt mock 数据源，接入真实数据时保持结构替换即可 - usage_local 读取 statusline 新增的 seven_day_fable 与 updated_at；400×300 小屏 render 跳过 scoped 行 - 同步更新 CLAUDE.md 分支说明 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
297 lines
12 KiB
Python
297 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据层 —— 汇总仪表盘所需的全部数据。
|
||
|
||
橘喵今日经营 + 近30天趋势来自 jm_api（jm-devops 后端统计接口，免鉴权 GET）。
|
||
天气来自 weather_api（和风天气，定位南京·秦淮）。日期/时间用真实系统时间。
|
||
任何接口失败时相关数据以占位符 "--" 呈现，不回退 mock。
|
||
"""
|
||
import json
|
||
from datetime import datetime
|
||
|
||
import config
|
||
import jm_api
|
||
import weather_api
|
||
import usage_local
|
||
|
||
WEEK_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
||
|
||
TREND_DAYS = 30 # 近 N 天趋势
|
||
FIVE_HOUR, SEVEN_DAY = 5 * 3600, 7 * 86400 # 两个滚动窗口长度（秒），用于算「时间已流逝%」
|
||
|
||
|
||
def _read_weather_cache():
|
||
try:
|
||
with open(config.WEATHER_CACHE_PATH, encoding="utf-8") as f:
|
||
return json.load(f) # {"ts": float, "data": {...}}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _write_weather_cache(ts, data):
|
||
try:
|
||
with open(config.WEATHER_CACHE_PATH, "w", encoding="utf-8") as f:
|
||
json.dump({"ts": ts, "data": data}, f, ensure_ascii=False)
|
||
except Exception as e:
|
||
print(f"[警告] 天气缓存写入失败：{e}")
|
||
|
||
|
||
def _read_trend_cache():
|
||
try:
|
||
with open(config.TREND_CACHE_PATH, encoding="utf-8") as f:
|
||
return json.load(f) # {"date": "YYYY-MM-DD", "data": {...}}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _write_trend_cache(date_key, data):
|
||
try:
|
||
with open(config.TREND_CACHE_PATH, "w", encoding="utf-8") as f:
|
||
json.dump({"date": date_key, "data": data}, f, ensure_ascii=False)
|
||
except Exception as e:
|
||
print(f"[警告] 趋势缓存写入失败：{e}")
|
||
|
||
|
||
def _trend(now):
|
||
"""近30天趋势，按日期缓存：recent-days 每天只更新一次。
|
||
|
||
当天已有缓存（且有数据）→ 直接复用，不发请求；当天还没有 → 拉取并写缓存。
|
||
拉取失败时沿用任意旧缓存（哪怕非当天），仍无缓存才回退空。
|
||
API 降序 → 升序：左旧右新。
|
||
"""
|
||
date_key = now.strftime("%Y-%m-%d")
|
||
cached = _read_trend_cache()
|
||
if cached and cached.get("date") == date_key and cached.get("data", {}).get("series"):
|
||
return cached["data"]
|
||
try:
|
||
asc = list(reversed(jm_api.get_recent_days(TREND_DAYS)))
|
||
trend = {
|
||
"dates": [d["date"][5:].replace("-", "/") for d in asc], # yyyy-MM-dd → MM/DD
|
||
"series": [
|
||
{"k": "单量", "style": "solid", "data": [int(d["orderCount"]) for d in asc]},
|
||
{"k": "流水", "style": "dashed", "data": [float(d["payAmount"]) for d in asc]},
|
||
{"k": "毛利", "style": "dotted", "data": [float(d["grossProfit"]) for d in asc]},
|
||
],
|
||
}
|
||
_write_trend_cache(date_key, trend)
|
||
return trend
|
||
except Exception as e:
|
||
print(f"[警告] 近30天趋势拉取失败：{e}")
|
||
if cached and cached.get("data", {}).get("series"): # 失败沿用旧缓存，避免趋势图整片 "--"
|
||
print("[警告] 沿用上次趋势缓存")
|
||
return cached["data"]
|
||
return {"dates": [], "series": []}
|
||
|
||
|
||
def _weather(now):
|
||
"""和风天气（南京·秦淮），带 WEATHER_TTL 文件缓存。
|
||
|
||
缓存未过期 → 直接复用，不发请求（每分钟刷新时省掉 14/15 次天气调用）。
|
||
过期 → 重新拉取并写缓存；全部接口失败时沿用上次缓存，仍无缓存才回退占位 "--"。
|
||
"""
|
||
ts = now.timestamp()
|
||
cached = _read_weather_cache()
|
||
if cached and ts - cached.get("ts", 0) < config.WEATHER_TTL:
|
||
return cached["data"]
|
||
out = {"loc": config.WEATHER_LOC_NAME, "cond": "--", "temp": "--", "range": "--", "icon": ""}
|
||
ok = False
|
||
try:
|
||
w = weather_api.get_now()
|
||
out["cond"] = w.get("text") or "--"
|
||
out["temp"] = f"{w['temp']}°"
|
||
out["icon"] = w.get("icon") or "" # 和风图标代码，供 renderer 映射图标
|
||
ok = True
|
||
except Exception as e:
|
||
print(f"[警告] 实时天气拉取失败：{e}")
|
||
try:
|
||
today = weather_api.get_today()
|
||
out["range"] = f"{today['tempMin']}°~{today['tempMax']}°"
|
||
ok = True
|
||
except Exception as e:
|
||
print(f"[警告] 今日温区拉取失败：{e}")
|
||
if ok:
|
||
_write_weather_cache(ts, out)
|
||
return out
|
||
if cached: # 全部失败：沿用上次缓存，避免天气区整片 "--"
|
||
print("[警告] 天气全部拉取失败，沿用上次缓存")
|
||
return cached["data"]
|
||
return out
|
||
|
||
|
||
# ---------- 格式化 ----------
|
||
def _fmt_count(n):
|
||
return f"{int(n):,}" # 千分位，如 1,284
|
||
|
||
|
||
def _fmt_money(yuan):
|
||
"""元金额 → 紧凑展示：≥1万显示「X.XX万」(两位小数)，否则取整元。"""
|
||
v = float(yuan)
|
||
return f"{v / 10000:.2f}万" if abs(v) >= 10000 else f"{v:.0f}"
|
||
|
||
|
||
def _fmt_rate(rate):
|
||
"""增长率字符串 → (方向, 文本)。null/空 → (None, '--')，方向 None 表示不画三角。
|
||
数值统一两位小数；整数部分不足两位用前导 0 补齐(6.5→06.50)，便于环比/同比上下对齐；
|
||
三/四位数(≥100%)原样完整输出不截断。"""
|
||
if rate is None or rate == "":
|
||
return (None, "--")
|
||
v = float(rate)
|
||
return ("down" if v < 0 else "up", f"{abs(v):05.2f}%")
|
||
|
||
|
||
def _col(k, v, wow, yoy):
|
||
return {"k": k, "v": v, "hb": _fmt_rate(wow), "tb": _fmt_rate(yoy)}
|
||
|
||
|
||
def _placeholder_col(k):
|
||
return {"k": k, "v": "--", "hb": (None, "--"), "tb": (None, "--")}
|
||
|
||
|
||
# ---------- 橘喵业务数据 ----------
|
||
def _mao(now):
|
||
upd = now.strftime("%H:%M")
|
||
# 今日经营（realtime）
|
||
try:
|
||
rt = jm_api.get_realtime()
|
||
upd = (rt.get("dataTime") or "")[11:16] or upd
|
||
cols = [
|
||
_col("单量", _fmt_count(rt["todayOrderCount"]), rt["orderCountWowRate"], rt["orderCountYoyRate"]),
|
||
_col("流水", _fmt_money(rt["todayPayAmount"]), rt["payAmountWowRate"], rt["payAmountYoyRate"]),
|
||
_col("毛利", _fmt_money(rt["todayGrossProfit"]), rt["grossProfitWowRate"], rt["grossProfitYoyRate"]),
|
||
]
|
||
except Exception as e:
|
||
print(f"[警告] 今日经营拉取失败：{e}")
|
||
cols = [_placeholder_col(k) for k in ("单量", "流水", "毛利")]
|
||
# 近30天趋势（按日期缓存，每天只拉一次）；宽版另用其最后 7 天做「近7天」数据表
|
||
trend = _trend(now)
|
||
return {"upd": upd, "cols": cols, "trend": trend, "week": _recent_days(trend, 7)}
|
||
|
||
|
||
def _recent_days(trend, n):
|
||
"""趋势的最后 n 天 → 表格数据 {dates:[MM/DD…], rows:[(名, [文本…], 合计文本)]}，供宽版「橘喵·近7天」表使用。
|
||
recent-days 不含今日，故表头日期到昨天为止。单量按千分位、流水/毛利按紧凑金额格式；无数据返回空 dict。"""
|
||
dates, series = trend.get("dates") or [], trend.get("series") or []
|
||
if not dates or not series:
|
||
return {}
|
||
dates = dates[-n:]
|
||
rows = []
|
||
for s in series:
|
||
data = (s.get("data") or [])[-n:]
|
||
if len(data) != len(dates):
|
||
continue
|
||
fmt = _fmt_count if s["k"] == "单量" else _fmt_money
|
||
rows.append((s["k"], [fmt(v) for v in data], fmt(sum(data))))
|
||
return {"dates": dates, "rows": rows}
|
||
|
||
|
||
# ---------- Claude Code 用量 ----------
|
||
def _parse_reset_ts(s):
|
||
"""resets_at → epoch 秒。兼容三种：epoch 数字（Claude Code stdin 给的就是 epoch）、
|
||
ISO 时间串（含结尾 'Z'）、空。失败返回 None。"""
|
||
if s is None or s == "":
|
||
return None
|
||
s = str(s)
|
||
if s.replace(".", "", 1).isdigit(): # 纯数字 → 当作 epoch 秒
|
||
return float(s)
|
||
try:
|
||
return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _window_time_pct(resets_at, window, now):
|
||
"""该滚动窗口「已流逝时间百分比」= (window − 距重置剩余) / window ×100。失败返回 None。"""
|
||
reset = _parse_reset_ts(resets_at)
|
||
if reset is None:
|
||
return None
|
||
remaining = max(0, min(window, reset - now.timestamp()))
|
||
return int(round((window - remaining) / window * 100))
|
||
|
||
|
||
def _fmt_reset(resets_at, window):
|
||
"""重置时刻 → 文本：≤1 天的窗口只给 HH:MM，更长的窗口带 MM/DD。解析失败返回 None。"""
|
||
ts = _parse_reset_ts(resets_at)
|
||
if ts is None:
|
||
return None
|
||
dt = datetime.fromtimestamp(ts)
|
||
return dt.strftime("%H:%M") if window <= 86400 else dt.strftime("%m/%d %H:%M")
|
||
|
||
|
||
# Claude 用量窗口：（快照键，行标签，窗口长度秒，scoped）。scoped=True 为「按模型限定」的额度
|
||
# （Fable 7d，来自 /api/oauth/usage 的 limits[]，由 statusline 写入快照 seven_day_fable），
|
||
# 400x300 三色小屏版面放不下第三行，只在 800x480 宽版展示。
|
||
USAGE_WINDOWS = (
|
||
("five_hour", "5h", FIVE_HOUR, False),
|
||
("seven_day", "7d", SEVEN_DAY, False),
|
||
("seven_day_fable", "Fable 7d", SEVEN_DAY, True),
|
||
)
|
||
|
||
|
||
def _usage(now):
|
||
"""Claude Code 用量（5h / 7d / Fable 7d）——只读本地 statusline 生产的快照（见 usage_local.py），**不发任何请求**。
|
||
|
||
utilization 来自 statusline 从 Claude Code stdin 落盘的权威实时值（Fable 行来自其转存的 API limits）；
|
||
time_pct(时间已流逝%) 与 pace 每次按当前时间**现算**（resets_at 为绝对时间）。
|
||
文件缺失/损坏 → 回退占位 pct=None（renderer 显示 "--"）；快照里没有某窗口（如旧版 statusline 无 Fable）同样 "--"。
|
||
结构契约（renderer 依赖）：{title, warn, caution, updated?, bars:[{k, pct(0~100 或 None), time_pct?, reset?, scoped}]}；
|
||
pct ≥ warn 时该条进度条与百分比转红；pace = pct − time_pct（>0 超前↑ / <0 节余↓）；
|
||
reset 为重置时刻文本；updated 为快照落盘时刻 HH:MM（活跃使用时持续刷新，空闲时停在最后一次）。
|
||
"""
|
||
raw = {}
|
||
try:
|
||
raw = usage_local.get_usage()
|
||
except Exception as e:
|
||
print(f"[警告] Claude 用量读取失败：{e}")
|
||
|
||
bars = []
|
||
for key, label, win, scoped in USAGE_WINDOWS:
|
||
seg = raw.get(key) or {}
|
||
util = seg.get("utilization")
|
||
pct = int(round(float(util))) if util is not None else None
|
||
bar = {"k": label, "pct": pct, "scoped": scoped}
|
||
tp = _window_time_pct(seg.get("resets_at"), win, now)
|
||
if pct is not None and tp is not None:
|
||
bar["time_pct"] = tp
|
||
rs = _fmt_reset(seg.get("resets_at"), win)
|
||
if rs:
|
||
bar["reset"] = rs
|
||
bars.append(bar)
|
||
out = {"title": "Claude Usage", "warn": config.USAGE_WARN, "caution": config.USAGE_CAUTION, "bars": bars}
|
||
upd = _parse_reset_ts(raw.get("updated_at"))
|
||
if upd:
|
||
out["updated"] = datetime.fromtimestamp(upd).strftime("%H:%M")
|
||
return out
|
||
|
||
|
||
def _usage_gpt(now):
|
||
"""ChatGPT 用量——**目前为 mock 占位**（mock=True，renderer 在标题右侧标「示例数据」）。
|
||
结构与 _usage 完全一致，后续接入真实数据源时只需替换本函数、保持返回结构不变。
|
||
mock 取值刻意覆盖多种配色档：5h 42%（黑）、7d 71%（黄「注意」）、Codex 7d 55%（黑，scoped 与左侧 Fable 行对称）。"""
|
||
bars = []
|
||
for label, pct, win, remain, scoped in (("5h", 42, FIVE_HOUR, 2 * 3600, False),
|
||
("7d", 71, SEVEN_DAY, 3 * 86400, False),
|
||
("Codex 7d", 55, SEVEN_DAY, 3 * 86400, True)):
|
||
reset_ts = now.timestamp() + remain
|
||
bars.append({"k": label, "pct": pct, "scoped": scoped,
|
||
"time_pct": _window_time_pct(reset_ts, win, now),
|
||
"reset": _fmt_reset(reset_ts, win)})
|
||
return {"title": "ChatGPT Usage", "warn": config.USAGE_WARN, "caution": config.USAGE_CAUTION,
|
||
"bars": bars, "mock": True}
|
||
|
||
|
||
def get_dashboard_data(now=None):
|
||
"""返回渲染所需的完整数据字典。"""
|
||
now = now or datetime.now()
|
||
return {
|
||
"date": {
|
||
"greg": now.strftime("%Y/%m/%d"),
|
||
"week": WEEK_CN[now.weekday()],
|
||
"time": now.strftime("%H:%M"),
|
||
},
|
||
"weather": _weather(now),
|
||
"mao": _mao(now),
|
||
"usage": _usage(now),
|
||
"usage_gpt": _usage_gpt(now),
|
||
}
|