YANG JIANKUAN
3fa6bb5de6
- 环比/同比增长率恰为 0 时方向记 flat，三角位画等号（与用量行 pace 持平一致） - 金额拆为「¥ / 主数字 / 万」三段，¥ 与 万 用 F12 小号伪粗体、与 F24 主数字基线对齐；单量列不变 - 环比/同比数值、BTC 涨跌幅、顶部电量/温度/湿度数字改 F12 伪粗体（字号不变），占位 -- 不加粗 - 等号绘制抽为基类 _eq 供 pace 与环比/同比共用；网页版 flat 同样输出等号 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
464 lines
20 KiB
Python
464 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据层 —— 汇总仪表盘所需的全部数据。
|
||
|
||
橘喵今日经营 + 近30天趋势来自 jm_api（jm-devops 后端统计接口，免鉴权 GET）。
|
||
天气来自 weather_api（和风天气，定位南京·江宁）。日期/时间用真实系统时间（统一按 config.TZ_NAME 时区）。
|
||
任何接口失败时相关数据以占位符 "--" 呈现，不回退 mock。
|
||
|
||
两种调用形态：
|
||
- 本机脚本（main.py）：get_dashboard_data() —— 带文件缓存（天气/趋势/BTC），用量读本地快照；
|
||
- 图片服务（server.py）：直接用各 `*_fetch()` 纯拉取函数 + `_usage(now, raw=<推送数据>)`，不落盘。
|
||
"""
|
||
import json
|
||
from datetime import datetime
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import config
|
||
import jm_api
|
||
import weather_api
|
||
import claude_usage
|
||
import codex_usage
|
||
import btc_api
|
||
import sensecraft_api
|
||
|
||
WEEK_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
||
TZ = ZoneInfo(config.TZ_NAME)
|
||
|
||
|
||
def now_tz():
|
||
"""当前时刻（带 config.TZ_NAME 时区）。"""
|
||
return datetime.now(TZ)
|
||
|
||
|
||
def _local(ts):
|
||
"""epoch 秒 → 该时区的 datetime。"""
|
||
return datetime.fromtimestamp(ts, TZ)
|
||
|
||
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, ok = weather_fetch()
|
||
if ok:
|
||
_write_weather_cache(ts, out)
|
||
return out
|
||
if cached: # 全部失败：沿用上次缓存，避免天气区整片 "--"
|
||
print("[警告] 天气全部拉取失败，沿用上次缓存")
|
||
return cached["data"]
|
||
return out
|
||
|
||
|
||
def weather_fetch():
|
||
"""实时拉取天气（不读不写缓存）→ （weather dict，是否至少一个接口成功）。服务端每次请求直接调用。"""
|
||
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}")
|
||
return out, ok
|
||
|
||
|
||
# ---------- 格式化 ----------
|
||
# 数值统一拆成三段 (前缀, 主数字, 后缀)：渲染器把主数字画大、前后缀画小（如「¥」「万」，2026-09-14 用户要求），
|
||
# 拼接后的整串仍放在 col["v"]，供网页版等只认字符串的消费方使用。
|
||
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 → 'flat'，渲染器在三角位画等号「=」（与用量行 pace 持平一致，2026-09-14 用户要求，原来 0 也画上三角）。
|
||
数值统一两位小数；整数部分不足两位用前导 0 补齐(6.5→06.50)，便于环比/同比上下对齐；
|
||
三/四位数(≥100%)原样完整输出不截断。"""
|
||
if rate is None or rate == "":
|
||
return (None, "--")
|
||
v = float(rate)
|
||
direction = "flat" if v == 0 else "down" if v < 0 else "up"
|
||
return (direction, f"{abs(v):05.2f}%")
|
||
|
||
|
||
def _col(k, parts, wow, yoy, wide=False):
|
||
"""一列指标。parts 为 (前缀, 主数字, 后缀)；wide=True 表示仅宽版（800×480）展示，400×300 小屏三列放不下、渲染时跳过。"""
|
||
return {"k": k, "v": "".join(parts), "vp": tuple(parts), "hb": _fmt_rate(wow), "tb": _fmt_rate(yoy), "wide": wide}
|
||
|
||
|
||
def _placeholder_col(k, wide=False):
|
||
return {"k": k, "v": "--", "vp": ("", "--", ""), "hb": (None, "--"), "tb": (None, "--"), "wide": wide}
|
||
|
||
|
||
# 今日实时五列：（标签、接口字段前缀、格式化、是否仅宽版）。字段前缀 base 对应 today{Base} / {base}WowRate / {base}YoyRate，
|
||
# 见 docs/首页看板接口文档-20260911.md。团购两列为新增字段，旧后端无此键 → 占位 "--"（不报错、不影响其余列）。
|
||
REALTIME_COLS = (
|
||
("探店单量", "orderCount", _fmt_count, False),
|
||
("探店流水", "payAmount", _fmt_money, False),
|
||
("团购单量", "groupbuyOrderCount", _fmt_count, True),
|
||
("团购流水", "groupbuyPayAmount", _fmt_money, True),
|
||
("毛利", "grossProfit", _fmt_money, False), # 口径已含团购毛利（接口合计，未拆分）
|
||
)
|
||
|
||
|
||
def _col_from(rt, k, base, fmt, wide):
|
||
cap = base[0].upper() + base[1:]
|
||
v = rt.get(f"today{cap}")
|
||
if v is None or v == "":
|
||
return _placeholder_col(k, wide)
|
||
return _col(k, fmt(v), rt.get(f"{base}WowRate"), rt.get(f"{base}YoyRate"), wide)
|
||
|
||
|
||
# ---------- 橘喵业务数据 ----------
|
||
def mao_fetch(now, with_trend=True):
|
||
"""橘喵今日经营（实时接口）；with_trend=False 时不拉近 30 天趋势（服务端与宽版都不用它，且趋势有文件缓存）。
|
||
返回（mao dict，实时接口是否成功）。"""
|
||
upd = now.strftime("%H:%M")
|
||
ok = True
|
||
# 今日经营（realtime）
|
||
try:
|
||
rt = jm_api.get_realtime()
|
||
upd = (rt.get("dataTime") or "")[11:16] or upd
|
||
cols = [_col_from(rt, k, base, fmt, wide) for k, base, fmt, wide in REALTIME_COLS]
|
||
except Exception as e:
|
||
print(f"[警告] 今日经营拉取失败：{e}")
|
||
cols = [_placeholder_col(k, wide) for k, _, _, wide in REALTIME_COLS]
|
||
ok = False
|
||
# 近30天趋势（按日期缓存，每天只拉一次）
|
||
trend = _trend(now) if with_trend else {"dates": [], "series": []}
|
||
return {"upd": upd, "cols": cols, "trend": trend}, ok
|
||
|
||
|
||
def _mao(now):
|
||
return mao_fetch(now)[0]
|
||
|
||
|
||
# ---------- 比特币 K 线 ----------
|
||
def _read_btc_cache():
|
||
try:
|
||
with open(config.BTC_CACHE_PATH, encoding="utf-8") as f:
|
||
return json.load(f) # {"ts": float, "data": {...}}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _write_btc_cache(ts, data):
|
||
try:
|
||
with open(config.BTC_CACHE_PATH, "w", encoding="utf-8") as f:
|
||
json.dump({"ts": ts, "data": data}, f, ensure_ascii=False)
|
||
except Exception as e:
|
||
print(f"[警告] BTC 缓存写入失败：{e}")
|
||
|
||
|
||
def _btc(now):
|
||
"""比特币 K 线（宽版底部面板；周期 BTC_BAR、根数 BTC_LIMIT），带 BTC_TTL 文件缓存；
|
||
拉取失败沿用旧缓存，仍无缓存回退空 candles。
|
||
结构契约（renderer 依赖）：{symbol, bar, src, candles:[{t, o, h, l, c}], last, chg, chg_label}，其中 t 为轴标签文本；
|
||
last 为最新一根（当前未收盘）的收盘价；chg 为涨跌幅（%）——时 K 取相对 24 根前（24h）收盘、不足则相对首根，
|
||
日 K 取相对上一根收盘，chg_label 说明口径（"24h"/"1d"）；涨=绿 / 跌=红 由 renderer 按 c≥o 判定。"""
|
||
ts = now.timestamp()
|
||
bar = config.BTC_BAR
|
||
cached = _read_btc_cache()
|
||
cd = (cached or {}).get("data", {})
|
||
# 周期或根数（BTC_BAR/BTC_LIMIT）变更即视为失效，避免改配置后 TTL 内仍显示旧范围
|
||
if cached and ts - cached.get("ts", 0) < config.BTC_TTL and cd.get("bar") == bar and len(cd.get("candles", [])) == config.BTC_LIMIT:
|
||
return cached["data"]
|
||
try:
|
||
out = btc_fetch()
|
||
_write_btc_cache(ts, out)
|
||
return out
|
||
except Exception as e:
|
||
print(f"[警告] BTC K 线拉取失败：{e}")
|
||
if cached:
|
||
print("[警告] 沿用上次 BTC 缓存")
|
||
return cached["data"]
|
||
return btc_empty()
|
||
|
||
|
||
def btc_empty():
|
||
return {"symbol": "BTC/USDT", "bar": config.BTC_BAR, "src": "", "candles": [], "last": None, "chg": None, "chg_label": ""}
|
||
|
||
|
||
def btc_fetch():
|
||
"""实时拉取 BTC K 线并整理成渲染结构（不读不写缓存）；失败抛异常。服务端每次请求直接调用。"""
|
||
bar = config.BTC_BAR
|
||
src, rows = btc_api.get_candles(bar, config.BTC_LIMIT)
|
||
fmt = "%m/%d %H:%M" if bar == "1H" else "%m/%d"
|
||
candles = [{"t": _local(r["ts"]).strftime(fmt), "o": r["o"], "h": r["h"], "l": r["l"], "c": r["c"]} for r in rows]
|
||
last = rows[-1]["c"]
|
||
if bar == "1H":
|
||
base = rows[-25]["c"] if len(rows) >= 25 else rows[0]["c"]
|
||
chg_label = "24h" if len(rows) >= 25 else f"{len(rows) - 1}h"
|
||
else:
|
||
base, chg_label = rows[-2]["c"], "1d"
|
||
return {"symbol": "BTC/USDT", "bar": bar, "src": src, "candles": candles,
|
||
"last": last, "chg": (last - base) / base * 100 if base else 0.0, "chg_label": chg_label}
|
||
|
||
|
||
# ---------- 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 = _local(ts)
|
||
return dt.strftime("%H:%M") if window <= 86400 else dt.strftime("%m/%d %H:%M")
|
||
|
||
|
||
# Claude 用量窗口：（快照键，短标签，长标签，窗口长度秒，scoped）。
|
||
# 短标签 k 给 400x300 小屏（标签列只有 26px）；长标签 label 给 800x480 宽版（2026-09-14 去掉 all 前缀，就叫 5h / 7d）。
|
||
# scoped=True 为「按模型限定」的额度（Fable 7d，来自 /api/oauth/usage 的 limits[] weekly_scoped，由 claude_usage.normalize
|
||
# 写入快照 seven_day_fable）：小屏 render() 跳过；宽版**并入紧邻其前的非 scoped 行**（7d 行拆成上下两条细条：上 = 总 7d，
|
||
# 下 = fable 7d），使左右两块都是两行、视觉对称（此前三行 vs 两行不对称）。
|
||
USAGE_WINDOWS = (
|
||
("five_hour", "5h", "5h", FIVE_HOUR, False),
|
||
("seven_day", "7d", "7d", SEVEN_DAY, False),
|
||
("seven_day_fable", "Fable 7d", "fable 7d", SEVEN_DAY, True),
|
||
)
|
||
# Codex 用量窗口（ChatGPT 订阅的 Codex 额度池，由 codex_usage.py 经 app-server RPC 取得）：只有 5h / 7d 两个窗口。
|
||
# ChatGPT 普通聊天额度没有账户级读取途径（2026-09 调研结论），故右侧块监控的就是 Codex 池并如实命名「Codex Usage」。
|
||
# 快照里各窗口可带 window（真实窗口秒数），pace 优先按它算；rateLimitsByLimitId 多桶原样存于 buckets，看到真实数据后再决定是否加第三行。
|
||
GPT_WINDOWS = (
|
||
("five_hour", "5h", "5h", FIVE_HOUR, False),
|
||
("seven_day", "7d", "7d", SEVEN_DAY, False),
|
||
)
|
||
|
||
|
||
def usage_from_raw(now, raw, windows, title):
|
||
"""原始快照 → 渲染结构（Claude / ChatGPT 通用）。
|
||
raw 形如 {five_hour:{utilization, resets_at, window?}, seven_day:{...}, <scoped 键>:{...}|null, updated_at?}；
|
||
utilization 为已用百分比，resets_at 为 epoch 秒或 ISO，window 为该窗口真实长度秒（缺省用 windows 表的常量）；缺哪个窗口哪个显示 "--"（宽版按 0 画）；utilization 为 0 且 resets_at 为 null 的空闲窗口记 time_pct=0，保证 pace 列不缺席。
|
||
time_pct（时间已流逝%）与 pace 按传入的 now **现算**——所以推送端只需推原始快照，服务端渲染时 pace 仍随时钟准确推进。
|
||
结构契约（renderer 依赖）：{title, warn, ok, pace_tol, updated?, bars:[{k, label, pct(0~100 或 None), time_pct?, reset?, scoped}]}；
|
||
pct ≥ warn 时该条进度条与百分比转红（宽版另按 ok/warn 分 绿/黑/红 三档，百分比随条色）；
|
||
pace = pct − time_pct（>0 超前↑ / <0 节余↓，宽版按 ±pace_tol 分 绿/黑/红 三档）；
|
||
reset 为重置时刻文本；updated 为快照落盘时刻 HH:MM（活跃使用时持续刷新，空闲时停在最后一次）。"""
|
||
raw = raw or {}
|
||
bars = []
|
||
for key, k, label, win, scoped in windows:
|
||
seg = raw.get(key) or {}
|
||
util = seg.get("utilization")
|
||
pct = int(round(float(util))) if util is not None else None
|
||
bar = {"k": k, "label": label, "pct": pct, "scoped": scoped}
|
||
win = int(seg.get("window") or win)
|
||
tp = _window_time_pct(seg.get("resets_at"), win, now)
|
||
if pct is not None and tp is None and pct == 0:
|
||
tp = 0 # 窗口空闲（utilization 0 且 resets_at null，5h 窗口重置后常见）：窗口尚未开始，已流逝 0%、pace 持平
|
||
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": title, "warn": config.USAGE_WARN, "ok": config.USAGE_OK, "pace_tol": config.PACE_TOL, "bars": bars}
|
||
upd = _parse_reset_ts(raw.get("updated_at"))
|
||
if upd:
|
||
out["updated"] = _local(upd).strftime("%H:%M")
|
||
return out
|
||
|
||
|
||
def _usage(now, raw=None):
|
||
"""Claude 账户级用量（5h / 7d / Fable 7d）。raw 为 None 时本机经 claude_usage 采集（复用 Claude Code 钥匙串登录态调
|
||
/api/oauth/usage，带 ≥300s 最小间隔缓存与退避）；服务端传入推送来的原始快照。"""
|
||
if raw is None:
|
||
raw = {}
|
||
try:
|
||
raw = claude_usage.get_usage()
|
||
except Exception as e:
|
||
print(f"[警告] Claude 用量读取失败：{e}")
|
||
return usage_from_raw(now, raw, USAGE_WINDOWS, "Claude Usage")
|
||
|
||
|
||
def gpt_mock_raw(now):
|
||
"""Codex 原始快照的 mock（服务端尚未收到推送时用）：5h 88%（≥warn 红）、7d 22%（≤ok 绿）覆盖两端配色。"""
|
||
t = now.timestamp()
|
||
return {"five_hour": {"utilization": 88, "resets_at": t + 2 * 3600},
|
||
"seven_day": {"utilization": 22, "resets_at": t + 3 * 86400}}
|
||
|
||
|
||
def _usage_gpt(now, raw=None, allow_local=True):
|
||
"""Codex 用量（ChatGPT 订阅的 Codex 额度池）。raw 为推送来的原始快照（codex_usage 产出，结构同 gpt_mock_raw）；
|
||
服务端未收到推送时（raw=None）用 mock 并标 mock=True（renderer 在标题右侧标「示例数据」）。本机形态直接经 codex_usage 采集。"""
|
||
mock = raw is None
|
||
if mock and allow_local:
|
||
try:
|
||
raw, mock = codex_usage.get_usage(), False
|
||
except Exception as e:
|
||
print(f"[警告] Codex 用量读取失败：{e}")
|
||
out = usage_from_raw(now, gpt_mock_raw(now) if mock else raw, GPT_WINDOWS, "Codex Usage")
|
||
if mock:
|
||
out["mock"] = True
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- 设备遥测（E1002 电量/温度/湿度）
|
||
def device_empty():
|
||
return {"battery": None, "charging": False, "temp": None, "humidity": None}
|
||
|
||
|
||
def device_fetch():
|
||
"""纯拉取（服务端用）：返回 (device dict, True)；失败抛出。结构 {battery, charging, temp, humidity}。"""
|
||
return sensecraft_api.get_iot_data(), True
|
||
|
||
|
||
def _device(now):
|
||
"""本机形态：带 DEVICE_TTL 文件缓存；失败沿用旧缓存，仍无则占位（渲染为 "--"）。"""
|
||
ts = now.timestamp()
|
||
cached = _read_json_cache(config.DEVICE_CACHE_PATH)
|
||
if cached and ts - cached.get("ts", 0) < config.DEVICE_TTL:
|
||
return cached["data"]
|
||
try:
|
||
out, _ = device_fetch()
|
||
_write_json_cache(config.DEVICE_CACHE_PATH, {"ts": ts, "data": out})
|
||
return out
|
||
except Exception as e:
|
||
print(f"[警告] 设备遥测拉取失败：{e}")
|
||
if cached:
|
||
return cached["data"]
|
||
return device_empty()
|
||
|
||
|
||
def _read_json_cache(path):
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _write_json_cache(path, obj):
|
||
try:
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(obj, f, ensure_ascii=False)
|
||
except Exception as e:
|
||
print(f"[警告] 缓存写入失败 {path}：{e}")
|
||
|
||
|
||
def get_dashboard_data(now=None):
|
||
"""返回渲染所需的完整数据字典（本机脚本形态：带文件缓存，用量读本地快照）。"""
|
||
now = now or now_tz()
|
||
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),
|
||
"btc": _btc(now),
|
||
"device": _device(now),
|
||
}
|