YANG JIANKUAN
3e0e6c73a1
- config.py 顶部新增零依赖 _load_dotenv()，凭证/参数统一从同目录 .env 读取， 真实环境变量（cron/命令行注入）优先；移除所有明文 fallback - 新增 .env.example 模板，.env 加入 .gitignore - pusher.py 删除 get_devices()，resolve_device_id() 直接取 ZECTRIX_DEVICE_ID， 不再调 GET /devices 列表 - jm_api.py 改为设备密钥授权（client_secret，免验证码），token 进程内缓存 + 401 重试 - 同步 CLAUDE.md / main.py 文档 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
橘喵 jm-devops 后端客户端。
|
||
|
||
鉴权：设备密钥授权（grant_type=client_secret，免验证码）。
|
||
先用 CLIENT_ID + CLIENT_SECRET 调 POST /auth/login 换 token（缓存于进程内，按有效期提前刷新），
|
||
再带 Authorization: Bearer <token> + clientid 头调统计接口；遇 401 清 token 重登一次重试。
|
||
|
||
接口：
|
||
- GET /system/statistics/realtime 今日经营实时数据
|
||
- GET /system/statistics/recent-days?days 近 N 天每日数据（按日期降序、不含今天）
|
||
"""
|
||
import json
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
import config
|
||
|
||
# 进程内 token 缓存（cron 每次独立进程，不落盘；单次运行内复用）
|
||
_token = None
|
||
_token_expire_at = 0.0
|
||
# token 到期前提前刷新的余量（秒）
|
||
_TOKEN_REFRESH_SKEW = 60
|
||
|
||
|
||
def _login():
|
||
"""用 client_secret 授权换取 token，写入模块级缓存。"""
|
||
global _token, _token_expire_at
|
||
# 设备专用明文登录端点（不走 @ApiEncrypt，仅放行 client_secret 授权）
|
||
url = config.BASE_API + "/auth/deviceLogin"
|
||
payload = json.dumps({
|
||
"clientId": config.CLIENT_ID,
|
||
"grantType": config.GRANT_TYPE,
|
||
"clientSecret": config.CLIENT_SECRET,
|
||
"tenantId": config.TENANT_ID,
|
||
}).encode("utf-8")
|
||
headers = {"Content-Type": "application/json;charset=utf-8", "clientid": config.CLIENT_ID}
|
||
req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e:
|
||
raise RuntimeError(f"登录 HTTP {e.code}: {e.read().decode('utf-8', 'ignore')[:200]}") from e
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"/auth/login 返回 code={data.get('code')}: {data.get('msg')}")
|
||
body = data["data"]
|
||
_token = body["access_token"]
|
||
# expire_in 为秒；提前 skew 刷新，缺省给个保守值
|
||
expire_in = body.get("expire_in") or 1800
|
||
_token_expire_at = time.time() + max(0, expire_in - _TOKEN_REFRESH_SKEW)
|
||
return _token
|
||
|
||
|
||
def _ensure_token():
|
||
if _token and time.time() < _token_expire_at:
|
||
return _token
|
||
return _login()
|
||
|
||
|
||
def _get(path, _retry=True):
|
||
global _token, _token_expire_at
|
||
token = _ensure_token()
|
||
url = config.BASE_API + path
|
||
headers = {
|
||
"Content-Type": "application/json;charset=utf-8",
|
||
"clientid": config.CLIENT_ID,
|
||
"Authorization": f"Bearer {token}",
|
||
}
|
||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e:
|
||
# token 失效（401）则清缓存重登一次重试
|
||
if e.code == 401 and _retry:
|
||
_token, _token_expire_at = None, 0.0
|
||
return _get(path, _retry=False)
|
||
raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'ignore')[:200]}") from e
|
||
# 业务层也可能用 401 表示未登录
|
||
if data.get("code") == 401 and _retry:
|
||
_token, _token_expire_at = None, 0.0
|
||
return _get(path, _retry=False)
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"{path} 返回 code={data.get('code')}: {data.get('msg')}")
|
||
return data["data"]
|
||
|
||
|
||
def get_realtime() -> dict:
|
||
"""今日经营实时数据（单量/流水/毛利 + 环比/同比）。"""
|
||
return _get("/system/statistics/realtime")
|
||
|
||
|
||
def get_recent_days(days=30) -> list:
|
||
"""近 N 天每日经营数据（API 按日期降序、不含今天）。"""
|
||
return _get(f"/system/statistics/recent-days?days={days}")["days"]
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("realtime:", json.dumps(get_realtime(), ensure_ascii=False)[:300])
|
||
print("recent[0]:", json.dumps(get_recent_days(30)[0], ensure_ascii=False))
|