4bd5c2abab
- ensure-cd-watches：默认并入 DEMAND_WATCHES（Plati/GGSel/FunPay 需求页） - notify-chain：忽略 noise 价差；推送 sold_out/restock 等货架高信号 - record-shop-outcome.sh + 总览三利害卡；post-pay 可接 MANHUANG_SHOP_ID
262 lines
9.5 KiB
Bash
Executable file
262 lines
9.5 KiB
Bash
Executable file
#!/usr/bin/env bash
|
||
# 蛮荒链危机 / 重要价差 → 通知 konqi（走现有 vm-dm，不改 NATS ACL）
|
||
# 节流：同 mode 2h 内不重复
|
||
#
|
||
# 用法：
|
||
# bash notify-chain.sh # 读 /tmp/joy-dashboard.json 或最新 collect 摘要
|
||
# bash notify-chain.sh --force # 忽略节流
|
||
set -euo pipefail
|
||
|
||
ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
|
||
CACHE="${HOME}/.cache/manhuang"
|
||
STAMP="${CACHE}/.chain-notify.last"
|
||
LOG="${HOME}/logs/manhuang-quotes.log"
|
||
DASH="${DASH:-/tmp/joy-dashboard.json}"
|
||
THROTTLE_S="${CHAIN_NOTIFY_THROTTLE_S:-7200}"
|
||
FORCE=0
|
||
[[ "${1:-}" == "--force" ]] && FORCE=1
|
||
|
||
mkdir -p "$CACHE" "$(dirname "$LOG")"
|
||
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] notify-chain: $*" | tee -a "$LOG"; }
|
||
|
||
# 解析 chain 状态
|
||
read_state() {
|
||
python3 - <<'PY'
|
||
import json, os, sys
|
||
from pathlib import Path
|
||
dash = Path(os.environ.get("DASH", "/tmp/joy-dashboard.json"))
|
||
m = {}
|
||
if dash.is_file():
|
||
try:
|
||
m = (json.loads(dash.read_text()).get("manhuang") or {})
|
||
except Exception as e:
|
||
print(json.dumps({"ok": False, "error": str(e)}))
|
||
sys.exit(0)
|
||
chain = m.get("chain") or {}
|
||
s = m.get("summary") or {}
|
||
bm = m.get("blackmarket") or {}
|
||
tc = m.get("tab_caches") or {}
|
||
moves = bm.get("price_moves") or tc.get("price_moves") or []
|
||
# 噪声（面值假差）不进通知
|
||
big = [
|
||
x for x in moves
|
||
if isinstance(x, dict)
|
||
and not x.get("noise")
|
||
and abs(float(x.get("delta") or 0)) >= 1.0
|
||
]
|
||
# 货架事件：售罄/下架/补货 + 非噪声价变（abs≥1）
|
||
shelf_raw = bm.get("shelf_events") or m.get("shelf_events") or tc.get("shelf_events") or []
|
||
shelf_hi = []
|
||
for e in shelf_raw:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
t = str(e.get("type") or "")
|
||
if e.get("noise"):
|
||
continue
|
||
if t in ("sold_out", "delist", "restock", "listed"):
|
||
shelf_hi.append(e)
|
||
elif t in ("price_up", "price_down") and abs(float(e.get("delta") or 0)) >= 1.0:
|
||
shelf_hi.append(e)
|
||
relays = m.get("relays") or {}
|
||
down = [r for r in (relays.get("relays") or []) if r.get("status") == "down"]
|
||
# demand → 实价
|
||
demand_rows = []
|
||
gap_total = 0
|
||
priced_n = 0
|
||
for drow in (bm.get("demand") or []):
|
||
if not isinstance(drow, dict):
|
||
continue
|
||
hits = int(drow.get("sku_price_hits") or 0)
|
||
gap = int(drow.get("sku_gap") or 0)
|
||
gap_total += gap
|
||
priced_n += hits
|
||
prices = []
|
||
for p in (drow.get("sku_prices") or [])[:4]:
|
||
if isinstance(p, dict) and p.get("price") is not None:
|
||
prices.append("%s $%s@%s" % (p.get("id"), p.get("price"), p.get("channel") or "?"))
|
||
demand_rows.append({
|
||
"capability": drow.get("capability"),
|
||
"status": drow.get("status"),
|
||
"hits": hits,
|
||
"gap": gap,
|
||
"prices": prices,
|
||
})
|
||
pr = m.get("probe_request") or {}
|
||
out = {
|
||
"ok": True,
|
||
"crisis": bool(chain.get("crisis")),
|
||
"mode": chain.get("mode") or "ok",
|
||
"headline": chain.get("headline") or "",
|
||
"err_n": chain.get("err_n") or 0,
|
||
"codex_total": s.get("codex_total"),
|
||
"codex_quota": s.get("codex_quota"),
|
||
"attention": s.get("attention_score"),
|
||
"price_moves_big": big[:5],
|
||
"shelf_events_hi": [
|
||
{
|
||
"sku_id": e.get("sku_id") or e.get("id"),
|
||
"type": e.get("type"),
|
||
"prev": e.get("prev"),
|
||
"cur": e.get("cur"),
|
||
"delta": e.get("delta"),
|
||
"ch": e.get("ch"),
|
||
}
|
||
for e in shelf_hi[:8]
|
||
],
|
||
"relays_down": [{"id": r.get("id"), "label": r.get("label")} for r in down[:5]],
|
||
"rings_bad": [
|
||
{"id": r.get("id"), "level": r.get("level"), "detail": r.get("detail")}
|
||
for r in (chain.get("rings") or [])
|
||
if r.get("level") in ("err", "warn")
|
||
][:6],
|
||
"demand_rows": demand_rows[:6],
|
||
"demand_priced_n": priced_n,
|
||
"demand_gap_total": gap_total,
|
||
"probe_request": {
|
||
"requested": pr.get("requested"),
|
||
"throttled": pr.get("throttled"),
|
||
"reasons": pr.get("reasons") or [],
|
||
"gap_skus": (pr.get("gap_skus") or [])[:8],
|
||
},
|
||
}
|
||
# 是否值得通知（危机 + 非噪声价差 + 货架事件 + 中转 + 需求情报）
|
||
notify = False
|
||
reasons = []
|
||
brief = False # 非危机情报 brief（更长节流）
|
||
if out["crisis"] or (out["codex_total"] or 0) > 0 and (out["codex_quota"] or 0) == 0:
|
||
notify = True
|
||
reasons.append("chain_crisis")
|
||
if out["price_moves_big"]:
|
||
notify = True
|
||
reasons.append("price_move")
|
||
# 售罄/补货等货架事件（噪声已滤）
|
||
_av = [e for e in shelf_hi if str(e.get("type") or "") in ("sold_out", "delist", "restock", "listed")]
|
||
if _av:
|
||
notify = True
|
||
reasons.append("shelf_availability")
|
||
elif shelf_hi and not out["price_moves_big"]:
|
||
# 仅有已在 big 中的价变时不重复；有独立 shelf 价变时 brief
|
||
notify = True
|
||
brief = True
|
||
reasons.append("shelf_events")
|
||
if out["relays_down"]:
|
||
notify = True
|
||
reasons.append("relay_down")
|
||
if out["crisis"] and gap_total > 0:
|
||
reasons.append("demand_sku_gap")
|
||
if out["crisis"] and priced_n > 0:
|
||
reasons.append("demand_prices")
|
||
# 大胆：有密令需求 + 已采到灰市价，即推采购情报（不成交，仅观察/比价）
|
||
if not notify and priced_n >= 2 and demand_rows:
|
||
notify = True
|
||
brief = True
|
||
reasons.append("heishi_price_brief")
|
||
if not notify and gap_total >= 3 and priced_n >= 1:
|
||
notify = True
|
||
brief = True
|
||
reasons.append("heishi_gap_brief")
|
||
out["should_notify"] = notify
|
||
out["brief"] = brief
|
||
out["reasons"] = reasons
|
||
print(json.dumps(out, ensure_ascii=False))
|
||
PY
|
||
}
|
||
|
||
export DASH
|
||
STATE=$(read_state)
|
||
echo "$STATE" | python3 -c "import sys,json; d=json.load(sys.stdin); print('should', d.get('should_notify'), d.get('reasons'), d.get('headline','')[:40])"
|
||
|
||
SHOULD=$(echo "$STATE" | python3 -c "import sys,json; print('1' if json.load(sys.stdin).get('should_notify') else '0')")
|
||
if [[ "$SHOULD" != "1" ]]; then
|
||
log "skip: no crisis/price/relay signal"
|
||
exit 0
|
||
fi
|
||
|
||
MODE=$(echo "$STATE" | python3 -c "import sys,json; d=json.load(sys.stdin); print(('brief:' if d.get('brief') else '') + (d.get('mode') or 'ok'))")
|
||
BRIEF=$(echo "$STATE" | python3 -c "import sys,json; print('1' if json.load(sys.stdin).get('brief') else '0')")
|
||
# brief 情报 6h 节流；危机仍用 THROTTLE_S（默认 2h）
|
||
EFF_THROTTLE=$THROTTLE_S
|
||
[[ "$BRIEF" == "1" ]] && EFF_THROTTLE="${CHAIN_BRIEF_THROTTLE_S:-21600}"
|
||
if [[ "$FORCE" != "1" && -f "$STAMP" ]]; then
|
||
now=$(date +%s)
|
||
last_epoch=$(python3 -c "import json;print(int(json.load(open('$STAMP')).get('epoch') or 0))" 2>/dev/null || echo 0)
|
||
last_mode=$(python3 -c "import json;print(json.load(open('$STAMP')).get('mode') or '')" 2>/dev/null || echo "")
|
||
if [[ "$last_mode" == "$MODE" && $((now - last_epoch)) -lt $EFF_THROTTLE ]]; then
|
||
log "skip: throttled mode=$MODE age=$((now - last_epoch))s thr=$EFF_THROTTLE"
|
||
exit 0
|
||
fi
|
||
fi
|
||
|
||
MSG=$(echo "$STATE" | python3 -c '
|
||
import json,sys
|
||
d=json.load(sys.stdin)
|
||
tag = "【黑市比价情报】" if d.get("brief") else "【蛮荒供应链】"
|
||
lines=[
|
||
"%s%s" % (tag, d.get("headline") or d.get("mode")),
|
||
"原因: %s" % (", ".join(d.get("reasons") or [])),
|
||
"注意力=%s Codex额=%s/%s · 观察不成交" % (d.get("attention"), d.get("codex_quota"), d.get("codex_total")),
|
||
]
|
||
for r in d.get("rings_bad") or []:
|
||
lines.append("- %s/%s: %s" % (r.get("id"), r.get("level"), r.get("detail")))
|
||
for r in d.get("relays_down") or []:
|
||
lines.append("- 中转 down: %s" % (r.get("label") or r.get("id")))
|
||
for p in d.get("price_moves_big") or []:
|
||
lines.append("- 价差 %s: %s→%s (Δ%s)" % (p.get("id"), p.get("prev"), p.get("cur"), p.get("delta")))
|
||
for e in d.get("shelf_events_hi") or []:
|
||
lines.append("- 货架 %s %s: %s→%s @%s" % (
|
||
e.get("type"), e.get("sku_id"), e.get("prev"), e.get("cur"), e.get("ch") or "?"))
|
||
# demand 实价 / 缺口
|
||
if d.get("demand_rows"):
|
||
lines.append("需求实价 %s · 缺口SKU %s" % (d.get("demand_priced_n"), d.get("demand_gap_total")))
|
||
for row in d.get("demand_rows") or []:
|
||
lines.append("- %s [%s] 实价%s 缺口%s" % (
|
||
row.get("capability"), row.get("status"), row.get("hits"), row.get("gap")))
|
||
for pr in (row.get("prices") or [])[:3]:
|
||
lines.append(" · %s" % pr)
|
||
pr = d.get("probe_request") or {}
|
||
if pr.get("requested"):
|
||
lines.append("已挂 demand-first 探针旗: %s" % ", ".join(pr.get("reasons") or []))
|
||
if pr.get("gap_skus"):
|
||
lines.append("gap SKU: %s" % ", ".join(pr.get("gap_skus") or []))
|
||
elif pr.get("throttled"):
|
||
lines.append("探针旗节流中: %s" % ", ".join(pr.get("reasons") or []))
|
||
lines.append("面板: http://localhost:8899/#tab=mh-heishi")
|
||
print("\n".join(lines))
|
||
')
|
||
|
||
# 优先 vm-dm konqi；失败写本地 outbox
|
||
sent=0
|
||
if command -v vm-dm >/dev/null 2>&1; then
|
||
if vm-dm konqi "$MSG" --type notification --subject "manhuang-chain" 2>>"$LOG"; then
|
||
sent=1
|
||
log "sent via vm-dm konqi"
|
||
else
|
||
log "vm-dm konqi failed, try outbox"
|
||
fi
|
||
fi
|
||
|
||
if [[ "$sent" != "1" ]]; then
|
||
OUTBOX="${HOME}/office/konqi/inbox"
|
||
mkdir -p "$OUTBOX"
|
||
FN="${OUTBOX}/manhuang-chain-$(date +%s).json"
|
||
python3 - <<PY
|
||
import json, time
|
||
from pathlib import Path
|
||
msg = '''$MSG'''
|
||
Path("$FN").write_text(json.dumps({
|
||
"from": "manhuang",
|
||
"to": "konqi",
|
||
"type": "notification",
|
||
"subject": "manhuang-chain",
|
||
"content": msg,
|
||
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print("wrote", "$FN")
|
||
PY
|
||
log "wrote local inbox $FN"
|
||
sent=1
|
||
fi
|
||
|
||
python3 -c "import json,time; open('$STAMP','w').write(json.dumps({'epoch':time.time(),'mode':'$MODE','ts':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())}))"
|
||
log "done mode=$MODE"
|
||
exit 0
|
