172 lines
5.2 KiB
Python
172 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
hk_rate.py — 每日刷新HKD/CNY汇率
|
|
|
|
用法：
|
|
from hk_rate import hkd_to_cny, refresh_rate
|
|
|
|
rate = hkd_to_cny() # 自动使用缓存，过期则刷新
|
|
refresh_rate() # 强制刷新
|
|
|
|
缓存文件：~/.cache/hk_exchange_rate.json
|
|
有效期：24小时
|
|
"""
|
|
|
|
import json, os, time, sys
|
|
from datetime import date
|
|
|
|
CACHE_PATH = os.path.expanduser("~/.cache/hk_exchange_rate.json")
|
|
CACHE_TTL = 86400 # 24小时，合理配置常量
|
|
|
|
# 数据库路径 - 使用绝对路径
|
|
DB_PATH = "/home/hmo/MoFin/data/mofin.db"
|
|
|
|
def _get_rate_from_db():
|
|
"""从 fx_rate 表获取汇率"""
|
|
try:
|
|
import sqlite3
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.execute(
|
|
"SELECT rate FROM fx_rate WHERE currency_pair='HKD_CNY' AND is_active=1 ORDER BY created_at DESC LIMIT 1"
|
|
)
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
if row and 0.7 < row[0] < 1.0:
|
|
return round(float(row[0]), 6)
|
|
except Exception as e:
|
|
print(f"[hk_rate] 数据库读取失败: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
def _set_rate_in_db(rate, source='api_refresh'):
|
|
"""将汇率写入 fx_rate 表"""
|
|
try:
|
|
import sqlite3
|
|
conn = sqlite3.connect(DB_PATH)
|
|
# 先将旧的设为 inactive
|
|
conn.execute("UPDATE fx_rate SET is_active=0 WHERE currency_pair='HKD_CNY'")
|
|
# 插入新汇率
|
|
conn.execute(
|
|
"INSERT INTO fx_rate (currency_pair, rate, source, created_at, is_active) VALUES (?, ?, ?, datetime('now','localtime'), 1)",
|
|
('HKD_CNY', rate, source)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return True
|
|
except Exception as e:
|
|
print(f"[hk_rate] 数据库写入失败: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
# 不再硬编码备用值，每次取缓存中的最近一次有效汇率
|
|
def _load_last_rate():
|
|
"""从缓存文件读取上次已知有效汇率"""
|
|
try:
|
|
if os.path.exists(CACHE_PATH):
|
|
with open(CACHE_PATH) as f:
|
|
data = json.load(f)
|
|
rate = data.get("rate", 0)
|
|
if 0.7 < rate < 1.0:
|
|
return round(float(rate), 6)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
PRIMARY_API = "https://api.exchangerate-api.com/v4/latest/HKD"
|
|
BACKUP_API = "https://api.exchangerate-api.com/v4/latest/USD"
|
|
|
|
def _fetch_rate():
|
|
"""从API获取 HKD/CNY 汇率"""
|
|
import urllib.request
|
|
ua = "Mozilla/5.0"
|
|
|
|
# 主API：直接用HKD→CNY
|
|
try:
|
|
req = urllib.request.Request(PRIMARY_API, headers={"User-Agent": ua})
|
|
with urllib.request.urlopen(req, timeout=8) as r:
|
|
data = json.loads(r.read())
|
|
cny = data.get("rates", {}).get("CNY")
|
|
if cny and 0.7 < cny < 1.0: # 合理性检查
|
|
return round(float(cny), 6)
|
|
except Exception:
|
|
pass
|
|
|
|
# 备用：USD→HKD + USD→CNY 间接计算
|
|
try:
|
|
req = urllib.request.Request(BACKUP_API, headers={"User-Agent": ua})
|
|
with urllib.request.urlopen(req, timeout=8) as r:
|
|
data = json.loads(r.read())
|
|
rates = data.get("rates", {})
|
|
hkd = rates.get("HKD")
|
|
cny = rates.get("CNY")
|
|
if hkd and cny:
|
|
rate = cny / hkd
|
|
if 0.7 < rate < 1.0:
|
|
return round(rate, 6)
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def hkd_to_cny(force_refresh=False):
|
|
"""获取 HKD→CNY 汇率，优先使用数据库中的权威汇率"""
|
|
os.makedirs(os.path.dirname(CACHE_PATH), exist_ok=True)
|
|
now = time.time()
|
|
rate = None
|
|
|
|
# 1. 优先从数据库获取（权威源）
|
|
if not force_refresh:
|
|
rate = _get_rate_from_db()
|
|
if rate is not None:
|
|
return rate
|
|
|
|
# 2. 读本地缓存
|
|
if not force_refresh:
|
|
try:
|
|
with open(CACHE_PATH) as f:
|
|
cached = json.load(f)
|
|
cache_date = cached.get("date", "")
|
|
rate = cached.get("rate")
|
|
cached_at = cached.get("cached_at", 0)
|
|
# 同一天且未过期
|
|
if (cache_date == date.today().isoformat()
|
|
and rate is not None
|
|
and (now - cached_at) < CACHE_TTL):
|
|
# 写入数据库作为权威源
|
|
_set_rate_in_db(rate, 'cache_sync')
|
|
return rate
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. 刷新API
|
|
rate = _fetch_rate()
|
|
if rate is None:
|
|
# API全挂，用缓存中的上次有效汇率
|
|
rate = _load_last_rate()
|
|
if rate is None:
|
|
rate = 0.87 # 极限兜底，纯预防
|
|
print(f"[hk_rate] API不可达，使用 {rate} (fallback)", file=sys.stderr)
|
|
else:
|
|
# 写本地缓存
|
|
try:
|
|
with open(CACHE_PATH, "w") as f:
|
|
json.dump({
|
|
"rate": rate,
|
|
"date": date.today().isoformat(),
|
|
"cached_at": now,
|
|
"source": "exchangerate-api.com",
|
|
}, f)
|
|
except Exception:
|
|
pass
|
|
# 写入数据库作为权威源
|
|
_set_rate_in_db(rate, 'api_refresh')
|
|
|
|
return rate
|
|
|
|
|
|
def refresh_rate():
|
|
return hkd_to_cny(force_refresh=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
r = hkd_to_cny()
|
|
print(f"HKD/CNY = {r}") |