eink-push/data.py

#!/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天趋势（按日期缓存，每天只拉一次）；合计每次现算，不进缓存（缓存结构保持不变）
    trend = dict(_trend(now))
    trend["totals"] = _trend_totals(trend)
    return {"upd": upd, "cols": cols, "trend": trend}


def _trend_totals(trend):
    """近30天三指标合计 → [（名，文本）]，供宽版（E1002）趋势图标题行展示；无数据返回空列表。
    单量按千分位计数，流水/毛利按金额紧凑格式（≥1万 显示 X.XX万）。"""
    out = []
    for s in trend.get("series") or []:
        data = s.get("data") or []
        if not data:
            continue
        total = sum(data)
        out.append((s["k"], _fmt_count(total) if s["k"] == "单量" else _fmt_money(total)))
    return out


# ---------- 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 _usage(now):
    """Claude Code 用量（5h / 7d）——只读本地 statusline 生产的快照（见 usage_local.py），**不发任何请求**。

    utilization 来自 statusline 从 Claude Code stdin 落盘的权威实时值；
    time_pct(时间已流逝%) 与 pace 每次按当前时间**现算**（resets_at 为绝对时间）。
    文件缺失/损坏 → 回退占位 pct=None（renderer 显示 "--"）。
    结构契约（renderer 依赖）：{title, warn, bars:[{k, pct(0~100 或 None), time_pct(0~100)?}]}；
    pct ≥ warn 时该条进度条与百分比转红；pace = pct − time_pct（>0 超前↑ / <0 节余↓）。
    """
    wins = {}
    try:
        raw = usage_local.get_usage()
        for key, label, win in (("five_hour", "5h", FIVE_HOUR), ("seven_day", "7d", SEVEN_DAY)):
            seg = raw.get(key) or {}
            wins[label] = {"pct": int(round(float(seg.get("utilization", 0)))),
                           "resets_at": seg.get("resets_at"), "win": win}
    except Exception as e:
        print(f"[警告] Claude 用量读取失败：{e}")

    bars = []
    for label in ("5h", "7d"):
        w = wins.get(label) or {}
        pct = w.get("pct")
        bar = {"k": label, "pct": pct}
        tp = _window_time_pct(w.get("resets_at"), w.get("win"), now)
        if pct is not None and tp is not None:
            bar["time_pct"] = tp
        bars.append(bar)
    return {"title": "Claude Usage", "warn": config.USAGE_WARN, "caution": config.USAGE_CAUTION, "bars": bars}


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),
    }