dalongxia/stock-risk-to-feishu
2026-03-04 13:25:48 +08:00

307 lines
11 KiB
Python
Executable File

#!/usr/bin/env python3
import argparse
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Dict, List, Tuple
DEFAULT_APP_TOKEN = "SZ4ebCKU3a9OdfspetxcSeg1n4g"
DEFAULT_TABLE_ID = "tblzuMy85WiRAUX5"
DEFAULT_TABLE_LINK = (
"https://acnuoqgha8fk.feishu.cn/wiki/"
"RBSxwq8woiQDCUkRZfkcg7x8nMb?table=tblzuMy85WiRAUX5&view=vewuydm9EI"
)
OPENCLAW_CONFIG = "/root/.openclaw/openclaw.json"
REQUIRED_FIELDS = [
"记录类型",
"批次ID",
"报表时间",
"统计周期天数",
"补货提前期天数",
"预警阈值天数",
"在售SKU数",
"风险SKU数",
"紧急SKU数",
"预警SKU数",
"SKU编码",
"SKU名称",
"SPU",
"风险等级",
"预计缺货天数",
"建议补货件数",
"可用库存",
"近7天发货件数",
"近30天发货件数",
"近30天日均发货",
"渠道",
"店铺",
"采购",
"运营",
"工厂",
"仓库",
]
def _http_json(method: str, url: str, headers: Dict[str, str], payload: Dict = None) -> Dict:
data = None
req_headers = dict(headers)
if payload is not None:
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req_headers["Content-Type"] = "application/json; charset=utf-8"
req = urllib.request.Request(url, data=data, headers=req_headers, method=method)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read().decode("utf-8")
if not raw:
return {}
return json.loads(raw)
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", errors="ignore")
raise RuntimeError(f"HTTP {e.code} {url}: {body[:1000]}") from e
except Exception as e:
raise RuntimeError(f"request failed {method} {url}: {e}") from e
def _load_feishu_app() -> Tuple[str, str]:
app_id = os.getenv("FEISHU_APP_ID", "").strip()
app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
if app_id and app_secret:
return app_id, app_secret
cfg_path = Path(OPENCLAW_CONFIG)
if not cfg_path.exists():
raise RuntimeError("missing feishu credentials (env and openclaw.json both unavailable)")
cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
ch = (cfg.get("channels") or {}).get("feishu") or {}
app_id = (ch.get("appId") or "").strip()
app_secret = (ch.get("appSecret") or "").strip()
if not app_id or not app_secret:
raise RuntimeError("feishu appId/appSecret is empty in openclaw.json")
return app_id, app_secret
def _tenant_token(app_id: str, app_secret: str) -> str:
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
res = _http_json(
"POST",
url,
headers={},
payload={"app_id": app_id, "app_secret": app_secret},
)
if res.get("code") != 0:
raise RuntimeError(f"feishu auth failed: {res}")
token = res.get("tenant_access_token")
if not token:
raise RuntimeError("feishu auth succeeded but tenant_access_token missing")
return token
def _parse_company_rows(output: str) -> Dict[int, Dict]:
result_sets: Dict[int, Dict] = {}
current_idx = None
current_cols: List[str] = []
current_rows: List[Dict] = []
def flush():
nonlocal current_idx, current_cols, current_rows
if current_idx is not None:
result_sets[current_idx] = {"columns": current_cols, "rows": current_rows}
current_idx = None
current_cols = []
current_rows = []
for raw_line in output.splitlines():
line = raw_line.strip()
if not line:
continue
if line.startswith("RESULT_SET "):
flush()
current_idx = int(line.split()[1])
continue
if line.startswith("COLUMNS:"):
cols = line.split(":", 1)[1].strip()
current_cols = [c.strip() for c in cols.split(",")]
continue
if line.startswith("{") and line.endswith("}"):
try:
current_rows.append(json.loads(line))
except Exception:
pass
continue
flush()
return result_sets
def _run_stock_risk(days: int, lead_days: int, warn_days: int, limit: int) -> Tuple[Dict, List[Dict]]:
cmd = [
"stock-risk-alerts",
"--days",
str(days),
"--lead-days",
str(lead_days),
"--warn-days",
str(warn_days),
"--limit",
str(limit),
]
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
err = (proc.stderr or proc.stdout or "stock-risk-alerts failed").strip()
raise RuntimeError(err)
parsed = _parse_company_rows(proc.stdout or "")
summary_rows = parsed.get(1, {}).get("rows", [])
detail_rows = parsed.get(2, {}).get("rows", [])
if not summary_rows:
raise RuntimeError("stock-risk-alerts returned no summary rows")
return summary_rows[0], detail_rows
def _list_fields(token: str, app_token: str, table_id: str) -> Dict[str, str]:
url = (
f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/fields"
"?page_size=500"
)
res = _http_json("GET", url, headers={"Authorization": f"Bearer {token}"})
if res.get("code") != 0:
raise RuntimeError(f"list fields failed: {res}")
items = (((res.get("data") or {}).get("items")) or [])
out: Dict[str, str] = {}
for it in items:
name = it.get("field_name")
fid = it.get("field_id")
if name and fid:
out[name] = fid
return out
def _create_field(token: str, app_token: str, table_id: str, field_name: str) -> None:
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/fields"
payload = {"field_name": field_name, "type": 1}
res = _http_json("POST", url, headers={"Authorization": f"Bearer {token}"}, payload=payload)
if res.get("code") != 0:
raise RuntimeError(f"create field failed ({field_name}): {res}")
def _ensure_fields(token: str, app_token: str, table_id: str) -> None:
fields = _list_fields(token, app_token, table_id)
for name in REQUIRED_FIELDS:
if name not in fields:
_create_field(token, app_token, table_id, name)
# wait a bit for schema consistency
time.sleep(0.5)
def _create_record(token: str, app_token: str, table_id: str, fields: Dict[str, str]) -> str:
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records"
payload = {"fields": fields}
res = _http_json("POST", url, headers={"Authorization": f"Bearer {token}"}, payload=payload)
if res.get("code") != 0:
raise RuntimeError(f"create record failed: {res}")
rec = (res.get("data") or {}).get("record") or {}
return rec.get("record_id") or rec.get("id") or ""
def _fmt(v) -> str:
if v is None:
return ""
if isinstance(v, float):
return f"{v:.2f}".rstrip("0").rstrip(".")
return str(v)
def main() -> int:
parser = argparse.ArgumentParser(description="Push stock risk alerts to Feishu as structured rows")
parser.add_argument("--days", type=int, default=30)
parser.add_argument("--lead-days", type=int, default=10)
parser.add_argument("--warn-days", type=int, default=15)
parser.add_argument("--top", type=int, default=50, help="top N detail rows to push")
parser.add_argument("--app-token", default=DEFAULT_APP_TOKEN)
parser.add_argument("--table-id", default=DEFAULT_TABLE_ID)
parser.add_argument("--table-link", default=DEFAULT_TABLE_LINK)
args = parser.parse_args()
days = max(7, min(args.days, 180))
lead_days = max(1, min(args.lead_days, 60))
warn_days = max(lead_days, min(args.warn_days, 90))
top = max(1, min(args.top, 500))
summary, details = _run_stock_risk(days, lead_days, warn_days, limit=max(top, 50))
app_id, app_secret = _load_feishu_app()
token = _tenant_token(app_id, app_secret)
_ensure_fields(token, args.app_token, args.table_id)
batch_id = time.strftime("%Y%m%d%H%M%S")
now_text = time.strftime("%Y-%m-%d %H:%M:%S")
summary_record = {
"记录类型": "汇总",
"批次ID": batch_id,
"报表时间": now_text,
"统计周期天数": _fmt(summary.get("lookback_days", days)),
"补货提前期天数": _fmt(summary.get("replenishment_lead_days", lead_days)),
"预警阈值天数": _fmt(summary.get("warning_days", warn_days)),
"在售SKU数": _fmt(summary.get("sold_sku_count_30d")),
"风险SKU数": _fmt(summary.get("stock_risk_sku_count")),
"紧急SKU数": _fmt(summary.get("urgent_sku_count")),
"预警SKU数": _fmt(summary.get("warning_sku_count")),
}
summary_record_id = _create_record(token, args.app_token, args.table_id, summary_record)
detail_record_ids: List[str] = []
for row in details[:top]:
fields = {
"记录类型": "明细",
"批次ID": batch_id,
"报表时间": now_text,
"统计周期天数": _fmt(days),
"补货提前期天数": _fmt(lead_days),
"预警阈值天数": _fmt(warn_days),
"SKU编码": _fmt(row.get("sku_id")),
"SKU名称": _fmt(row.get("sku_name")),
"SPU": _fmt(row.get("spu")),
"风险等级": _fmt(row.get("stock_risk_level")),
"预计缺货天数": _fmt(row.get("estimated_stockout_days")),
"建议补货件数": _fmt(row.get("suggest_replenish_qty_to_lead_days")),
"可用库存": _fmt(row.get("available_qty")),
"近7天发货件数": _fmt(row.get("shipped_qty_7d")),
"近30天发货件数": _fmt(row.get("shipped_qty_30d")),
"近30天日均发货": _fmt(row.get("avg_daily_shipped_30d")),
"渠道": _fmt(row.get("top_shop_channel")),
"店铺": _fmt(row.get("top_shop_name")),
"采购": _fmt(row.get("purchaser_name")),
"运营": _fmt(row.get("operator_name")),
"工厂": _fmt(row.get("factory_name")),
"仓库": _fmt(row.get("warehouse_name")),
}
rec_id = _create_record(token, args.app_token, args.table_id, fields)
detail_record_ids.append(rec_id)
print("PUSH_STATUS: success")
print(f"TABLE_LINK: {args.table_link}")
print(f"BATCH_ID: {batch_id}")
print(f"SUMMARY_RECORD_ID: {summary_record_id}")
print(f"DETAIL_RECORD_COUNT: {len(detail_record_ids)}")
if detail_record_ids:
print(f"FIRST_DETAIL_RECORD_ID: {detail_record_ids[0]}")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as e:
print(f"PUSH_STATUS: failed\nERROR: {e}", file=sys.stderr)
raise