af1e82e9a5
- index.html manhuang.js 版本对齐内容 hash，避免面板吃到旧 JS - derive_shelf_events(limit=0) 返回空列表；key 查找兼容非 str - --record-outcome 同时 --fail/--dispute 只计一次 fail - build-tab-caches 价差与 shelf_events 共用一次差分
301 lines
9.8 KiB
Python
Executable file
301 lines
9.8 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""鬼市本地店表：初始化 / 校验 / 可选同步 scout。
|
||
|
||
路径（优先）：
|
||
~/.config/manhuang/shops.json
|
||
~/data/manhuang/shops.json（scout）
|
||
|
||
用法：
|
||
python3 scripts/manhuang/ensure-shops.py # 无文件则从 example 建空壳
|
||
python3 scripts/manhuang/ensure-shops.py --validate
|
||
python3 scripts/manhuang/ensure-shops.py --sync-scout # scp 到 scout:data/manhuang/
|
||
python3 scripts/manhuang/ensure-shops.py --add-stub # 加一条可编辑空店位（非 demo）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
# ensure-shops 可 --record-outcome 时 import manhuang_shelf
|
||
_SRC = ROOT / "src"
|
||
if _SRC.is_dir() and str(_SRC) not in sys.path:
|
||
sys.path.insert(0, str(_SRC))
|
||
EXAMPLE = ROOT / "configs" / "manhuang" / "shops.example.json"
|
||
MAC_SHOPS = Path.home() / ".config" / "manhuang" / "shops.json"
|
||
SCOUT_REMOTE = "scout:data/manhuang/shops.json"
|
||
|
||
|
||
def _default_doc() -> dict[str, Any]:
|
||
return {
|
||
"version": 1,
|
||
"updated": str(date.today()),
|
||
"note": "本地店表：填 TG handle / 发卡 URL；频道名不进 git。demo 条目 collect 会跳过。",
|
||
"brands": [
|
||
"feibisi.com",
|
||
"weixiaoduo.com",
|
||
"wenpai.org",
|
||
"cravatar.com",
|
||
"feicode.com",
|
||
],
|
||
"rsshub_base": "",
|
||
"shops": [],
|
||
}
|
||
|
||
|
||
def _load(path: Path) -> dict[str, Any]:
|
||
if not path.is_file():
|
||
return _default_doc()
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
if isinstance(data, dict):
|
||
return data
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
return _default_doc()
|
||
|
||
|
||
def _write(path: Path, data: dict[str, Any]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
data["updated"] = str(date.today())
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def ensure(path: Path) -> dict[str, Any]:
|
||
if path.is_file():
|
||
return _load(path)
|
||
data = _default_doc()
|
||
if EXAMPLE.is_file():
|
||
try:
|
||
ex = json.loads(EXAMPLE.read_text(encoding="utf-8"))
|
||
if isinstance(ex, dict) and ex.get("brands"):
|
||
data["brands"] = list(ex["brands"])
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
_write(path, data)
|
||
print("created", path)
|
||
return data
|
||
|
||
|
||
def validate(data: dict[str, Any]) -> list[str]:
|
||
errs: list[str] = []
|
||
shops = data.get("shops")
|
||
if shops is None:
|
||
errs.append("missing shops[]")
|
||
return errs
|
||
if not isinstance(shops, list):
|
||
errs.append("shops must be list")
|
||
return errs
|
||
seen: set[str] = set()
|
||
real = 0
|
||
for i, s in enumerate(shops):
|
||
if not isinstance(s, dict):
|
||
errs.append(f"shops[{i}] not object")
|
||
continue
|
||
sid = str(s.get("id") or "")
|
||
if not sid:
|
||
errs.append(f"shops[{i}] missing id")
|
||
continue
|
||
if sid in seen:
|
||
errs.append(f"duplicate id {sid}")
|
||
seen.add(sid)
|
||
if "example" in sid.lower() or "demo" in sid.lower():
|
||
continue
|
||
url = str(s.get("url") or "")
|
||
if ".invalid" in url:
|
||
errs.append(f"{sid}: url uses .invalid (skipped by collect)")
|
||
continue
|
||
if not (url.startswith("http") or s.get("handle") or s.get("rsshub_path")):
|
||
errs.append(f"{sid}: need url or handle or rsshub_path")
|
||
real += 1
|
||
if not real:
|
||
errs.append("no real shops (only empty/demo) — add TG handle or faka URL")
|
||
return errs
|
||
|
||
|
||
def add_stub(data: dict[str, Any]) -> dict[str, Any]:
|
||
shops = list(data.get("shops") or [])
|
||
n = sum(1 for s in shops if isinstance(s, dict) and str(s.get("id") or "").startswith("local-shop-"))
|
||
sid = f"local-shop-{n + 1}"
|
||
shops.append({
|
||
"id": sid,
|
||
"name": "待填 · 本地店",
|
||
"kind": "telegram",
|
||
"platform": "telegram",
|
||
"handle": "",
|
||
"url": None,
|
||
"rsshub_path": "",
|
||
"goods": ["ChatGPT", "Claude"],
|
||
"sku_for": ["chatgpt", "claude", "codex_quota"],
|
||
"status": "watch",
|
||
"risk": "high",
|
||
"trade": True,
|
||
"depth": "mid",
|
||
"opacity": "opaque",
|
||
"region": "global",
|
||
"layer": "trade_grey",
|
||
"trust": {
|
||
"platform": {"rating": None, "sales_n": None},
|
||
"ours": {"orders_ok": 0, "orders_fail": 0, "dispute": False, "quality": None},
|
||
},
|
||
"next": "填 handle 或 url 后跑 ensure-cd-watches --shops",
|
||
})
|
||
data["shops"] = shops
|
||
return data
|
||
|
||
|
||
def record_outcome(
|
||
data: dict[str, Any],
|
||
shop_id: str,
|
||
*,
|
||
ok: bool = False,
|
||
fail: bool = False,
|
||
dispute: bool = False,
|
||
quality: float | None = None,
|
||
note: str | None = None,
|
||
) -> tuple[dict[str, Any], str]:
|
||
"""成交回填 trust.ours（不成交系统外；人记一笔）。"""
|
||
shops = list(data.get("shops") or [])
|
||
sid = str(shop_id or "").strip()
|
||
if not sid:
|
||
return data, "missing shop id"
|
||
found = None
|
||
for s in shops:
|
||
if isinstance(s, dict) and str(s.get("id") or "") == sid:
|
||
found = s
|
||
break
|
||
if found is None:
|
||
return data, f"shop not found: {sid}"
|
||
trust = dict(found.get("trust") or {}) if isinstance(found.get("trust"), dict) else {}
|
||
ours = dict(trust.get("ours") or {}) if isinstance(trust.get("ours"), dict) else {}
|
||
if ok:
|
||
ours["orders_ok"] = int(ours.get("orders_ok") or 0) + 1
|
||
# dispute 视为失败；与 --fail 同时传时只计一次 fail
|
||
if fail or dispute:
|
||
ours["orders_fail"] = int(ours.get("orders_fail") or 0) + 1
|
||
if dispute:
|
||
ours["dispute"] = True
|
||
if quality is not None:
|
||
try:
|
||
ours["quality"] = float(quality)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if note:
|
||
ours["last_note"] = str(note)[:200]
|
||
ours["last_outcome_at"] = str(date.today())
|
||
trust["ours"] = ours
|
||
found["trust"] = trust
|
||
# 即时算分（可选，collect 也会再算）
|
||
try:
|
||
sys.path.insert(0, str(ROOT / "src"))
|
||
from joy.dashboard.manhuang_shelf import score_shop_trust
|
||
|
||
scored = score_shop_trust(found)
|
||
trust["score"] = scored.get("score")
|
||
trust["band"] = scored.get("band")
|
||
found["trust"] = trust
|
||
found["trust_score"] = scored.get("score")
|
||
found["trust_band"] = scored.get("band")
|
||
except Exception as e:
|
||
return data, f"recorded but score failed: {e}"
|
||
data["shops"] = shops
|
||
return data, (
|
||
f"ok shop={sid} score={found.get('trust_score')} band={found.get('trust_band')} "
|
||
f"ours={ours}"
|
||
)
|
||
|
||
|
||
def sync_scout(path: Path) -> int:
|
||
if not path.is_file():
|
||
print("no local shops file", path)
|
||
return 1
|
||
if not shutil.which("scp"):
|
||
print("scp not found")
|
||
return 1
|
||
r = subprocess.run(
|
||
["scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=12", str(path), SCOUT_REMOTE],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if r.returncode != 0:
|
||
print("scp fail", (r.stderr or r.stdout)[:200])
|
||
return r.returncode
|
||
print("synced", path, "→", SCOUT_REMOTE)
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--path", type=Path, default=MAC_SHOPS)
|
||
ap.add_argument("--validate", action="store_true")
|
||
ap.add_argument("--sync-scout", action="store_true")
|
||
ap.add_argument("--add-stub", action="store_true")
|
||
ap.add_argument("--record-outcome", metavar="SHOP_ID", help="成交回填 trust.ours")
|
||
ap.add_argument("--ok", action="store_true", help="with --record-outcome: 成功一单")
|
||
ap.add_argument("--fail", action="store_true", help="with --record-outcome: 失败一单")
|
||
ap.add_argument("--dispute", action="store_true", help="with --record-outcome: 争议/假货 → block")
|
||
ap.add_argument("--quality", type=float, default=None, help="0-1 或 0-100 货质")
|
||
ap.add_argument("--note", type=str, default=None, help="短备注")
|
||
args = ap.parse_args()
|
||
|
||
data = ensure(args.path)
|
||
if args.add_stub:
|
||
data = add_stub(data)
|
||
_write(args.path, data)
|
||
print("added stub shop →", args.path)
|
||
|
||
if args.record_outcome:
|
||
if not (args.ok or args.fail or args.dispute):
|
||
print("need --ok and/or --fail and/or --dispute with --record-outcome")
|
||
return 2
|
||
data, msg = record_outcome(
|
||
data,
|
||
args.record_outcome,
|
||
ok=args.ok,
|
||
fail=args.fail,
|
||
dispute=args.dispute,
|
||
quality=args.quality,
|
||
note=args.note,
|
||
)
|
||
if msg.startswith("shop not found") or msg.startswith("missing"):
|
||
print(msg)
|
||
return 1
|
||
_write(args.path, data)
|
||
print("record-outcome", msg)
|
||
print("next: python3 scripts/manhuang/build-tab-caches.py")
|
||
|
||
errs = validate(data)
|
||
n = len(data.get("shops") or [])
|
||
real = sum(
|
||
1 for s in (data.get("shops") or [])
|
||
if isinstance(s, dict)
|
||
and "demo" not in str(s.get("id") or "").lower()
|
||
and "example" not in str(s.get("id") or "").lower()
|
||
and ".invalid" not in str(s.get("url") or "")
|
||
)
|
||
print("path", args.path)
|
||
print("shops_total", n, "realish", real, "brands", len(data.get("brands") or []))
|
||
if errs:
|
||
print("validate:")
|
||
for e in errs:
|
||
print(" -", e)
|
||
if args.validate and any("no real shops" not in e for e in errs):
|
||
return 1
|
||
if args.validate and real == 0:
|
||
return 2 # soft: empty shops
|
||
else:
|
||
print("validate: ok")
|
||
|
||
if args.sync_scout:
|
||
return sync_scout(args.path)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
