2c6281d041
Expand manhuang collect/UI (tabs, markets, guishi/quote probes), add configs and launchd helpers, ops handbooks, and scripts for channel validation and price/threat enrichment. Keys stay in keys.env.example only.
204 lines
7.5 KiB
Python
Executable file
204 lines
7.5 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""拉取 ai-team ChangeDetection 状态 → 本地缓存。
|
||
|
||
环境变量：
|
||
CHANGEDTECTION_URL 默认 http://10.211.55.107:5000
|
||
CHANGEDTECTION_API_KEY 若设置则走 /api/v1/watch
|
||
|
||
无 key 时：只探活首页 + 记「需 key」；不炸外网。
|
||
写出：~/.cache/manhuang/changedetection/status.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
UA = "joy-manhuang-cd-probe/1.0"
|
||
OUT_DIR = Path.home() / ".cache" / "manhuang" / "changedetection"
|
||
OUT = OUT_DIR / "status.json"
|
||
DEFAULT_URL = "http://10.211.55.107:5000"
|
||
KEYS = Path.home() / ".config" / "manhuang" / "keys.env"
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _load_keys_env() -> None:
|
||
import os
|
||
if not KEYS.is_file():
|
||
return
|
||
try:
|
||
for line in KEYS.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
k, v = k.strip(), v.strip().strip('"').strip("'")
|
||
if k and k not in os.environ:
|
||
os.environ[k] = v
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _opener_for(url: str) -> urllib.request.OpenerDirector:
|
||
"""内网 10.x / 集群 IP 直连：Python urllib 不认 NO_PROXY 的 CIDR，会被本机 7893 代理劫持成 502。"""
|
||
host = urllib.request.urlparse(url).hostname or ""
|
||
direct = (
|
||
host.startswith("10.")
|
||
or host.startswith("192.168.")
|
||
or host in ("127.0.0.1", "localhost")
|
||
or host.endswith(".local")
|
||
)
|
||
if direct:
|
||
return urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
return urllib.request.build_opener()
|
||
|
||
|
||
def _fetch(url: str, headers: dict[str, str] | None = None, timeout: int = 12) -> tuple[int, str]:
|
||
h = {"User-Agent": UA, "Accept": "application/json,text/html,*/*"}
|
||
if headers:
|
||
h.update(headers)
|
||
req = urllib.request.Request(url, headers=h)
|
||
opener = _opener_for(url)
|
||
try:
|
||
with opener.open(req, timeout=timeout) as r:
|
||
return int(getattr(r, "status", 200) or 200), r.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
body = e.read().decode("utf-8", "replace")
|
||
except Exception:
|
||
body = ""
|
||
return int(e.code), body
|
||
except Exception as e:
|
||
return 0, str(e)[:200]
|
||
|
||
|
||
def probe() -> dict[str, Any]:
|
||
_load_keys_env()
|
||
base = (os.environ.get("CHANGEDTECTION_URL") or DEFAULT_URL).rstrip("/")
|
||
key = (os.environ.get("CHANGEDTECTION_API_KEY") or os.environ.get("CD_API_KEY") or "").strip()
|
||
out: dict[str, Any] = {
|
||
"ts": _now(),
|
||
"ok": False,
|
||
"base": base,
|
||
"mode": "api" if key else "health_only",
|
||
"watches": [],
|
||
"counts": {"n": 0, "changed": 0, "errors": 0, "manhuang": 0},
|
||
}
|
||
|
||
# health
|
||
code, body = _fetch(base + "/")
|
||
out["health_http"] = code
|
||
out["health_ok"] = 200 <= code < 500 and code != 0
|
||
|
||
# 推荐监控（ensure-cd-watches.py 同源）
|
||
out["suggested_watches"] = [
|
||
{"url": "https://openai.com/chatgpt/pricing/", "tag": "msrp", "sku": "openai"},
|
||
{"url": "https://claude.ai/pricing", "tag": "msrp", "sku": "claude"},
|
||
{"url": "https://cursor.com/pricing", "tag": "msrp", "sku": "cursor"},
|
||
{"url": "https://openrouter.ai/models", "tag": "relay", "sku": "openrouter"},
|
||
{"url": "https://ggsel.net/en/catalog/grok-accounts", "tag": "market", "sku": "supergrok"},
|
||
{"url": "https://ggsel.net/en/catalog?query=chatgpt%20plus", "tag": "market", "sku": "openai"},
|
||
{"url": "https://ggsel.net/en/catalog?query=claude%20pro", "tag": "market", "sku": "claude"},
|
||
{"url": "https://ggsel.net/en/catalog?query=cursor%20pro", "tag": "market", "sku": "cursor"},
|
||
{"url": "https://5sim.net/", "tag": "sms", "sku": "sms_otp"},
|
||
{"url": "https://tokenx24.com", "tag": "relay", "sku": "relay_premium"},
|
||
{"url": "https://howardpen9.github.io/awesome-ai-api-proxy/", "tag": "relay_index", "sku": "newapi_relay"},
|
||
{"url": "https://x.ai/grok", "tag": "msrp", "sku": "supergrok"},
|
||
]
|
||
|
||
if key:
|
||
# ChangeDetection.io API: x-api-key header
|
||
code, body = _fetch(
|
||
base + "/api/v1/watch",
|
||
headers={"x-api-key": key, "Accept": "application/json"},
|
||
)
|
||
out["api_http"] = code
|
||
if code == 200:
|
||
try:
|
||
data = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
data = None
|
||
watches = []
|
||
# API may return dict uuid→watch or list
|
||
if isinstance(data, dict):
|
||
items = data.items()
|
||
for uid, w in items:
|
||
if not isinstance(w, dict):
|
||
continue
|
||
watches.append(_norm_watch(uid, w))
|
||
elif isinstance(data, list):
|
||
for i, w in enumerate(data):
|
||
if isinstance(w, dict):
|
||
watches.append(_norm_watch(w.get("uuid") or str(i), w))
|
||
out["watches"] = watches
|
||
out["counts"]["n"] = len(watches)
|
||
out["counts"]["changed"] = sum(1 for w in watches if w.get("last_changed"))
|
||
out["counts"]["errors"] = sum(1 for w in watches if w.get("has_error"))
|
||
out["counts"]["manhuang"] = sum(
|
||
1 for w in watches
|
||
if "manhuang" in str(w.get("title") or "").lower()
|
||
or any(
|
||
x in str(w.get("url") or "")
|
||
for x in ("ggsel.net", "openrouter", "5sim.net", "tokenx24", "chatgpt/pricing", "claude.ai/pricing")
|
||
)
|
||
)
|
||
out["ok"] = True
|
||
out["needs_api_key"] = False
|
||
else:
|
||
out["error"] = f"api_http={code} body={body[:120]}"
|
||
out["ok"] = out["health_ok"]
|
||
out["needs_api_key"] = code == 403
|
||
else:
|
||
# 无 key：从 HTML 粗提取标题数（尽力）
|
||
titles = re.findall(r'class=["\'][^"\']*watch-title[^"\']*["\'][^>]*>([^<]+)', body, re.I)
|
||
if not titles:
|
||
titles = re.findall(r'data-watch-uuid=["\'][^"\']+["\']', body)
|
||
out["html_hint_n"] = len(titles)
|
||
out["needs_api_key"] = True
|
||
out["hint"] = (
|
||
"设置 CHANGEDTECTION_API_KEY（~/.config/manhuang/keys.env）后可拉 /api/v1/watch；"
|
||
"或运行 ensure-cd-watches.py"
|
||
)
|
||
out["ok"] = out["health_ok"]
|
||
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
OUT.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return out
|
||
|
||
|
||
def _norm_watch(uid: str, w: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"uuid": str(uid),
|
||
"title": w.get("title") or w.get("url") or uid,
|
||
"url": w.get("url"),
|
||
"last_checked": w.get("last_checked") or w.get("last_checked_text"),
|
||
"last_changed": w.get("last_changed") or w.get("last_changed_text"),
|
||
"has_error": bool(w.get("has_error") or w.get("error")),
|
||
"tags": w.get("tags") or [],
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
r = probe()
|
||
print(
|
||
"wrote", OUT,
|
||
"ok", r.get("ok"),
|
||
"mode", r.get("mode"),
|
||
"watches", r.get("counts", {}).get("n"),
|
||
"health", r.get("health_http"),
|
||
)
|
||
if r.get("needs_api_key"):
|
||
print("hint:", r.get("hint"))
|
||
return 0 if r.get("ok") else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
