Files
ops-assistant/scripts/rate_query.py
T

87 lines
4.0 KiB
Python

#!/usr/bin/env python3
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone
import requests
CACHE = "/root/.openclaw/workspace/fx_realtime.json"
FETCH = "/root/.openclaw/workspace/web_tasks/fx_fetch_real.js"
PAIRS = {"USD/CNY", "EUR/CNY", "SGD/CNY", "GBP/CNY", "USD/EUR", "GBP/EUR"}
PAT = re.compile(r'\[\[(20\d{2}),(\d{1,2}),(\d{1,2}),(\d{1,2}),(\d{1,2}),null,null,\[\]\],\[([0-9]+\.[0-9]+),')
def google_rate(base, quote):
s = requests.Session()
s.headers.update({"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/126.0 Safari/537.36", "Accept-Language":"en-US,en;q=0.9"})
s.cookies.set("SOCS", "CAESHAgBEhJnd3NfMjAyMzA4MTAtMF9SQzIaAmVuIAEaBgiAo_CmBg", domain=".google.com")
u = f"https://www.google.com/finance/quote/{base}-{quote}"
r = s.get(u, timeout=15, allow_redirects=True)
hits = PAT.findall(r.text)
if not hits:
return None, None
y, mo, d, h, mi, price = max(hits, key=lambda x: tuple(int(v) for v in x[:5]))
return float(price), f"{y}-{int(mo):02d}-{int(d):02d} {int(h):02d}:{int(mi):02d} UTC"
def fresh_cache():
try:
with open(CACHE, encoding="utf-8") as f: raw = json.load(f)
gen = datetime.fromisoformat(raw.get("generatedAt", "").replace("Z", "+00:00"))
return raw if (datetime.now(timezone.utc) - gen).total_seconds() <= 600 else None
except Exception:
return None
def ensure_real():
raw = fresh_cache()
if raw:
return raw
try:
subprocess.run(["node", FETCH], cwd=os.path.dirname(FETCH), timeout=150, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
except Exception:
return None
return fresh_cache()
def network_rates(raw, base, quote):
key = f"{base}/{quote}"
rev = f"{quote}/{base}"
for name, data in (("Visa", (raw.get("visa") or {}).get(key)), ("Mastercard", (raw.get("mastercard") or {}).get(key))):
if data and data.get("ok"):
try:
orig = data["data"]["originalValues"]
fx = float(orig["fxRateVisa"])
f, t = orig.get("fromCurrency"), orig.get("toCurrency")
if f == base and t == quote: yield name, fx, data.get("queryDate") or data.get("data", {}).get("disclaimerDate", "")
elif f == quote and t == base: yield name, 1/fx, data.get("queryDate", "")
except Exception: pass
data = (raw.get("mastercard") or {}).get(key)
if data and data.get("data") and data["data"].get("data"):
try: yield "Mastercard", float(data["data"]["data"]["conversionRate"]), data["data"]["data"].get("fxDate", "")
except Exception: pass
data = (raw.get("mastercard") or {}).get(rev)
if data and data.get("data") and data["data"].get("data"):
try: yield "Mastercard", 1/float(data["data"]["data"]["conversionRate"]), data["data"]["data"].get("fxDate", "")
except Exception: pass
def main():
if len(sys.argv) != 4: raise SystemExit("usage: rate_query.py AMOUNT BASE QUOTE")
amount = float(sys.argv[1]); base = sys.argv[2].upper(); quote = sys.argv[3].upper()
g, gt = google_rate(base, quote)
raw = ensure_real() if f"{base}/{quote}" in PAIRS or f"{quote}/{base}" in PAIRS else None
lines = [f"💱 {amount:g} {base} → {quote}", ""]
if g is not None:
lines += [f"Google 1 {base} = {g:.6f} {quote}", f" {amount:g} {base} = {amount*g:.2f} {quote}"]
else: lines.append("Google 暂时无法获取")
seen = set()
if raw:
for name, rate, date in network_rates(raw, base, quote):
if name in seen: continue
seen.add(name)
lines += [f"{name:<12}1 {base} = {rate:.6f} {quote}", f" {amount:g} {base} = {amount*rate:.2f} {quote} ({date})"]
if not seen: lines.append("Visa/MC 暂无最新官方结算数据")
lines += ["", "Google 为参考市场价;Visa/万事达为官方计算器结算汇率。"]
print("\n".join(lines))
if __name__ == "__main__":
main()