fx.py

"""
汇率换算 — 把各卡残值按显示货币汇总。
汇率从 open.er-api.com 拉取(无 key), 本地缓存 1 天; 拉取失败用内置近似汇率兜底。
"""
from __future__ import annotations
import json
import os
import threading
import time
import urllib.request

import paths
_CACHE = paths.data_path("fx_cache.json")
_URL = "https://open.er-api.com/v6/latest/USD"
_mem = {"rates": None}          # 进程内已加载的汇率
_fetching = {"on": False}       # 是否已在后台拉取

SYMBOL = {"CNY": "元", "JPY": "円", "USD": "$", "EUR": "€", "GBP": "£",
          "KRW": "₩", "HKD": "HK$", "TWD": "NT$", "AUD": "A$", "CAD": "C$",
          "SGD": "S$", "THB": "฿", "INR": "₹", "RUB": "₽", "VND": "₫"}
# 内置近似汇率(USD 基准, 拉取失败时兜底)
_FALLBACK = {"USD": 1.0, "CNY": 7.2, "JPY": 155.0}


def _fetch_bg():
    """后台拉取汇率并写缓存 —— 绝不在 UI 线程里联网, 避免卡死。"""
    try:
        req = urllib.request.Request(_URL, headers={"User-Agent": "SFCardViewerPro/1.0"})
        with urllib.request.urlopen(req, timeout=15) as r:
            data = json.loads(r.read().decode("utf-8"))
        rates = data.get("rates") or {}
        if rates.get("CNY") and rates.get("JPY"):
            _mem["rates"] = rates
            try:
                with open(_CACHE, "w", encoding="utf-8") as f:
                    json.dump({"ts": time.time(), "rates": rates}, f)
            except OSError:
                pass
    except Exception:
        pass
    finally:
        _fetching["on"] = False


def _rates() -> dict:
    """立即返回可用汇率(内存/磁盘缓存/兜底), 永不阻塞; 过期时后台异步刷新。"""
    if _mem["rates"]:
        return _mem["rates"]
    stale = True
    if os.path.exists(_CACHE):
        try:
            with open(_CACHE, "r", encoding="utf-8") as f:
                c = json.load(f)
            if c.get("rates"):
                _mem["rates"] = c["rates"]
                stale = time.time() - c.get("ts", 0) >= 86400
        except (OSError, json.JSONDecodeError):
            pass
    # 没缓存或已过期 → 后台拉一次(只起一个线程)
    if stale and not _fetching["on"]:
        _fetching["on"] = True
        threading.Thread(target=_fetch_bg, daemon=True).start()
    return _mem["rates"] or _FALLBACK


def available() -> list:
    """所有可换算货币代码(已排序)。常用币种排前面。"""
    codes = sorted(_rates().keys())
    head = [c for c in ("CNY", "JPY", "USD", "EUR", "GBP", "HKD", "TWD", "KRW") if c in codes]
    rest = [c for c in codes if c not in head]
    return head + rest


def label(code: str) -> str:
    s = SYMBOL.get(code)
    return f"{code} {s}" if s else code


def convert(amount: float, from_cur: str, to_cur: str) -> float:
    if from_cur == to_cur:
        return amount
    r = _rates()
    rf = r.get(from_cur); rt = r.get(to_cur)
    if not rf or not rt:
        rf = _FALLBACK.get(from_cur, 1); rt = _FALLBACK.get(to_cur, 1)
    return amount / rf * rt   # USD 基准: amount/rf=USD, *rt=目标


if __name__ == "__main__":
    print("100 元 ->", round(convert(100, "CNY", "JPY"), 1), "円")
    print("1000 円 ->", round(convert(1000, "JPY", "CNY"), 2), "元")
    print("100 元 ->", round(convert(100, "CNY", "USD"), 2), "USD")