ops-assistant/scripts/rate_query.py

#!/usr/bin/env python3
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone, timedelta
import requests

CACHE = "/root/.openclaw/workspace/fx_realtime.json"
REPORT_TZ = timezone(timedelta(hours=8))
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 parse_date(value):
    if not value:
        return None
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%Y/%m/%d"):
        try:
            return datetime.strptime(value[:10], fmt).date()
        except ValueError:
            continue
    return None

def age_label(date_text):
    d = parse_date(date_text)
    if not d:
        return ""
    days = (datetime.now(REPORT_TZ).date() - d).days
    if days <= 0:
        return "🟢 最新"
    if days == 1:
        return "🟢 1天前"
    if days <= 3:
        return f"🟠 {days}天前"
    return f"🔴 {days}天前"

def fmt_date(date_text):
    d = parse_date(date_text)
    return d.isoformat() if d else (date_text or "未知日期")

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}", ""]
    rates = []
    if g is not None:
        lines += ["市场参考", f"Google      1 {base} = {g:.6f} {quote}", f"            {amount:g} {base} = {amount*g:.2f} {quote}"]
        rates.append(("Google", g))
    else:
        lines += ["市场参考", "Google      暂时无法获取"]

    network = []
    if raw:
        for name, rate, date in network_rates(raw, base, quote):
            if any(n == name for n, _, _ in network):
                continue
            network.append((name, rate, date))
    lines += ["", "卡组织汇率"]
    if network:
        for name, rate, date in network:
            freshness = age_label(date)
            date_part = f"📅 {fmt_date(date)}"
            if freshness:
                date_part += f" · {freshness}"
            lines += [f"{name:<12}1 {base} = {rate:.6f} {quote}", f"            {amount:g} {base} = {amount*rate:.2f} {quote}", f"            {date_part}"]
            rates.append((name, rate))
    else:
        lines.append("Visa/MC      暂无最新官方结算数据")

    if network and g is not None:
        lines += ["", "📊 相对 Google"]
        for name, rate, _ in network:
            diff = (rate / g - 1.0) * 100.0
            sign = "+" if diff >= 0 else ""
            lines.append(f"{name:<12}{sign}{diff:.3f}%")
        if len(network) >= 1:
            card_rates = [r for _, r, _ in network]
            lo, hi = min(card_rates), max(card_rates)
            if len(card_rates) == 1:
                lines += ["", f"💳 卡组织参考：{amount*lo:.2f} {quote}"]
            else:
                lines += ["", f"💳 卡组织参考区间：{amount*lo:.2f}–{amount*hi:.2f} {quote}"]

        stale = [(n, d) for n, _, d in network if parse_date(d) and (datetime.now(timezone.utc).date() - parse_date(d)).days >= 2]
        if stale:
            lines += ["", "⚠️ " + "；".join(f"{n} 数据距今天 {(datetime.now(REPORT_TZ).date()-parse_date(d)).days} 天" for n, d in stale) + "，不要将其视为当前实时市场价。"]

    lines += ["", "Google 为当前市场参考价；Visa/万事达为官方卡组织结算汇率。实际扣款还可能受发卡行费用、入账时间等影响。"]
    print("\n".join(lines))

if __name__ == "__main__":
    main()