diff --git a/backend/README.md b/backend/README.md
index ab2996a2..16591a95 100644
--- a/backend/README.md
+++ b/backend/README.md
@@ -29,6 +29,26 @@ uv run ruff check .
- `core/basic/equipment_registry.py`: equipment version registry and cache.
- `database/connect.py`: SQLite and Redis connections.
+## Gold Cache and Notifications
+
+- Application startup checks both realtime and official gold caches, then checks
+ them every minute. Data older than 30 minutes refreshes without HTTP traffic.
+ Redis locks prevent duplicate work across workers. Realtime region pages stay
+ 8 seconds apart. Failures retain the last successful data for up to 24 hours.
+- Official authentication uses `HELPER_TOKEN` and `HELPER_ID`. Authentication
+ failures (`returnCode=-30003`, explicit token-expiry messages or HTTP 401) can
+ notify via Bark. Network errors do not trigger token-expiry notifications.
+- Set optional `BARK_URL` in `config/.env`, `.env.dev`, `.env.prod` or the process
+ environment to your full Bark push URL, including its device key. There is no
+ default destination; unset/blank means no Bark request. Restart after changing
+ credentials. A successful alert is deduplicated per token for 24 hours; a new
+ token permits a new alert. Failed pushes remain retryable. No credentials are
+ included in notification text or logged push errors.
+- Official gold responses also contain `max` and `min`: the largest/smallest
+ values of `10000 / price` across both rank arrays, in 万金币 per 100点券,
+ rounded to 4 decimals. Nonpositive/invalid prices are excluded; absent valid
+ quotes yield `null`. Existing rank arrays retain their 点券 per 100万金币 unit.
+
## Compatibility Rules
- Do not change `/calc` request and response contract without frontend migration approval.
diff --git a/backend/api/routes/colg.py b/backend/api/routes/colg.py
index 5875ff86..c721e7f1 100644
--- a/backend/api/routes/colg.py
+++ b/backend/api/routes/colg.py
@@ -1,19 +1,17 @@
-import asyncio
import logging
from api.types.colg import HistoryGold, OfficialGold, RealTimeGold
+from api.utils import official_gold, thirdpart_gold
from api.core.Gzip import register_gzip_request
import json
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, BackgroundTasks, Depends, Query
+from redis.asyncio import Redis as AsyncRedis
from api.core.Response import response
-from api.core.Redis import get_redis_info, get_redis_info_async
-from api.dp import AsyncRedisDep, RedisDep
from typing import Annotated
import requests
from config.main import config
from datetime import datetime, timezone, timedelta
from api.core.Response import Return as Response
-import aiohttp
router = APIRouter()
logger = logging.getLogger('dcalc.colg')
@@ -35,36 +33,6 @@ url = 'https://poservice.colg.cn/api/v1/trend/getCoinTrendDataKol'
payload = {'token': config.COLG_TOKEN, 'area': 0, 'time_type': 'other', 'start_time': '2026-01-06', 'end_time': '2026-01-06'}
-payload_helper = {
- 'r': 'godRank/godDealRank',
- 'userId': config.HELPER_ID,
- 'token': config.HELPER_TOKEN,
-}
-
-OFFICIAL_GOLD_URL = 'https://dzhu.qq.com/zangyi/activity/app'
-OFFICIAL_GOLD_REGIONS = ('1', '2', '3A', '3B', '4', '5', '6', '7')
-
-
-async def _fetch_official_gold() -> list[dict]:
- timeout = aiohttp.ClientTimeout(total=10)
- async with aiohttp.ClientSession(timeout=timeout) as session:
-
- async def fetch_region(region: str) -> dict:
- form = {**payload_helper, 'serverNo': region}
- async with session.post(OFFICIAL_GOLD_URL, data=form) as resp:
- resp.raise_for_status()
- payload = await resp.json(content_type=None)
- data = payload['data']
- return {
- 'updataTime': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
- 'region': f'跨{region}',
- 'goldDealRank': [int(item['price']) for item in data['godDealRank']],
- 'goldTopDealRank': [int(item['price']) for item in data['godTopDealRank']],
- }
-
- return await asyncio.gather(*(fetch_region(region) for region in OFFICIAL_GOLD_REGIONS))
-
-
def get_date_str(timestamp: int) -> str:
utc_time = datetime.fromtimestamp(timestamp, tz=timezone.utc)
beijing_time = utc_time + timedelta(hours=8)
@@ -72,30 +40,13 @@ def get_date_str(timestamp: int) -> str:
@router.get('/gold/realtime/', operation_id='realtimeGold', response_model=Response[list[RealTimeGold]])
-def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
- """
- 返回所有跨区的实时金价信息
- """
- key = 'openapi:colg:real_time_gold'
- try:
+async def get_real_time_gold(redis: Annotated[AsyncRedis, Depends(thirdpart_gold.get_redis)], background_tasks: BackgroundTasks) -> Response[list[RealTimeGold]]:
+ """获取网站金币比例，数据为各跨区首次报价前200条的平均万金币/元。
- def get_gold_data():
- res = []
- for region_name, region_id in regionMap.items():
- payload['area'] = region_id
- payload['start_time'] = datetime.now().strftime('%Y-%m-%d')
- payload['end_time'] = datetime.now().strftime('%Y-%m-%d')
- response = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(payload))
- response.raise_for_status()
- resp = response.json().get('data', {})
- res.append({'updataTime': get_date_str(resp['x'][-1]), 'region': region_name, 'goldPrice': resp['y'][-1]})
- return res
- except Exception:
- # print(f"File not found: {e}")
- return response(data=[])
- # 设置半小时过期一次
- data = get_redis_info(redis, key, get_gold_data, 30 * 60)
- return response(data=data)
+ updataTime为数字秒级时间戳，code为跨区标识。与thirdpart接口共用Redis缓存和
+ 后台刷新任务：30分钟自动更新，也支持按需补齐，逐区间隔8秒；初始化未齐时响应体code=202。
+ """
+ return await get_thirdpart_gold(redis, background_tasks)
@router.get('/gold/history/', operation_id='historyGold', response_model=Response[HistoryGold] | Response[None])
@@ -171,7 +122,7 @@ def get_history_gold(region: Annotated[int, Query(..., description='跨区id')],
@router.get('/gold/official/', operation_id='officialGold', response_model=Response[list[OfficialGold]])
-async def get_official_gold(redis: AsyncRedisDep) -> Response[list[OfficialGold]]:
+async def get_official_gold(redis: Annotated[AsyncRedis, Depends(thirdpart_gold.get_redis)]) -> Response[list[OfficialGold]]:
"""
获取DNF官方金币寄售实时数据
@@ -180,11 +131,43 @@ async def get_official_gold(redis: AsyncRedisDep) -> Response[list[OfficialGold]
Note:
- 该接口会返回所有跨区的官方金币寄售数据，包括最低成交价排名前5的价格和最高成交价排名前5的价格
- 数据来源于DNF官方提供的接口，可能会有一定的延迟和不稳定性
+ - 30分钟自动刷新，失败保留最多24小时的旧值；认证失败且配置了BARK_URL时发送通知
"""
- key = 'openapi:colg:real_time_gold_officical'
try:
- data = await get_redis_info_async(redis, key, _fetch_official_gold, 30 * 60)
+ data = await official_gold.get_data(redis)
except Exception:
logger.exception('fetch official gold failed')
- return response(data=[])
+ return response(code=503, data=[], message='官方金价缓存暂不可用')
+ if not data:
+ return response(code=503, data=[], message='官方金价暂不可用，请稍后重试')
+ return response(data=data)
+
+
+@router.get('/gold/thirdpart/', operation_id='thirdpartGold', response_model=Response[list[RealTimeGold]])
+async def get_thirdpart_gold(redis: Annotated[AsyncRedis, Depends(thirdpart_gold.get_redis)], background_tasks: BackgroundTasks) -> Response[list[RealTimeGold]]:
+ """获取跨1～跨8（含跨3A/3B）的第三方首批报价均值，单位：万金币/元。
+
+ 不限制最低总价；各平台首次返回的有效报价汇总、去重、按单价取前200条，
+ 对每条报价的1/Price取算术平均，不按库存或份数加权。保留上游原始价格，
+ 不额外计算交易手续费。该接口无上游分页参数，前200条是本接口的样本口径。
+
+ 每30分钟自动后台刷新，跨区页面最短间隔8秒（每分钟不超过8次），Redis锁避免重复刷新。
+ 初始化跨区未齐时响应体code=202并注明进度；9区缓存齐全才返回code=200。
+ 刷新中可返回部分已完成跨区，缺失跨区不补0。
+ 已有缓存即时返回，失败保留原更新时间，旧值最多保留24小时。
+ 该接口在DEBUG模式下也使用缓存，Redis不可用时不直接抓取上游。
+ """
+ try:
+ data = await thirdpart_gold.read_cache(redis)
+ except Exception:
+ logger.exception('read thirdpart gold cache failed')
+ return response(code=503, data=[], message='第三方金价缓存暂不可用')
+ try:
+ await thirdpart_gold.schedule_refresh(redis, background_tasks, data)
+ except Exception:
+ logger.exception('schedule thirdpart gold refresh failed')
+ return response(code=200 if data else 503, data=data, message='第三方金价刷新暂不可用')
+ if len(data) < len(thirdpart_gold.REGIONS):
+ return response(code=202, data=data,
+ message=f'第三方金价缓存尚未齐全（{len(data)}/{len(thirdpart_gold.REGIONS)}跨区），后台刷新中或等待重试')
return response(data=data)
diff --git a/backend/api/types/colg.py b/backend/api/types/colg.py
index dd2c14df..78cca932 100644
--- a/backend/api/types/colg.py
+++ b/backend/api/types/colg.py
@@ -1,14 +1,30 @@
from pydantic import BaseModel, Field
+GOLD_REGION_CODES = {
+ '跨1': 'k1', '跨2': 'k2', '跨3A': 'k3a', '跨3B': 'k3b',
+ '跨4': 'k4', '跨5': 'k5', '跨6': 'k6', '跨7': 'k7', '跨8': 'k8',
+}
+
+
class RealTimeGold(BaseModel):
- updataTime: str = Field(..., description='更新时间')
+ updataTime: int = Field(..., ge=0, description='更新时间，Unix秒级时间戳')
"""更新时间"""
region: str = Field(..., description='跨区')
"""跨区"""
- goldPrice: float|str = Field(..., description='金价')
+ code: str = Field(..., description='跨区code：k1、k2、k3a、k3b、k4、k5、k6、k7、k8')
+ goldPrice: float = Field(..., gt=0, allow_inf_nan=False, description='金币比例，1元对应的万金币数量')
"""金价"""
+
+class ThirdPartGoldCache(RealTimeGold):
+ """内部缓存数据，不作为接口返回模型。"""
+
+ goldPrice: float = Field(..., gt=0, allow_inf_nan=False, description='首页有效报价换算为万金币/元后的算术平均值')
+ sampleCount: int = Field(..., ge=1, le=200, description='参与均值计算的报价条数')
+ stale: bool = Field(..., description='数据是否已超过30分钟刷新周期')
+
+
class DayGold(BaseModel):
date: str = Field(..., description='日期')
"""日期"""
@@ -25,11 +41,14 @@ class HistoryGold(BaseModel):
"""echarts数据"""
class OfficialGold(BaseModel):
- updataTime: str = Field(..., description='更新时间')
+ updataTime: str = Field(..., pattern=r'^\d+$', description='更新时间，Unix秒级时间戳字符串')
"""更新时间"""
region: str = Field(..., description='跨区')
"""跨区"""
- goldDealRank: list[float|str] = Field(..., description='金币寄售最低成交价排名前5的价格')
+ code: str = Field(..., description='跨区code：k1、k2、k3a、k3b、k4、k5、k6、k7')
+ goldDealRank: list[float] = Field(..., max_length=5, description='金币寄售最低成交价，最多5项，每100万金币对应的点券数量')
"""金币寄售最低成交价排名前5的价格"""
- goldTopDealRank: list[float|str] = Field(..., description='金币寄售最高成交价排名前5的价格')
+ goldTopDealRank: list[float] = Field(..., max_length=5, description='金币寄售最高成交价，最多5项，每100万金币对应的点券数量')
"""金币寄售最高成交价排名前5的价格"""
+ max: float | None = Field(..., description='100点券可兑换的最大万金币数量，无有效报价时为null')
+ min: float | None = Field(..., description='100点券可兑换的最小万金币数量，无有效报价时为null')
diff --git a/backend/api/utils/gold_refresh.py b/backend/api/utils/gold_refresh.py
new file mode 100644
index 00000000..6e765335
--- /dev/null
+++ b/backend/api/utils/gold_refresh.py
@@ -0,0 +1,45 @@
+"""随应用生命周期运行的金币缓存定时检查，无需外部请求触发。"""
+
+import asyncio
+import logging
+
+from api.utils import official_gold, thirdpart_gold
+
+logger = logging.getLogger('dcalc.gold_refresh')
+CHECK_INTERVAL_SECONDS = 60
+_periodic_tasks: list[asyncio.Task] = []
+
+
+async def refresh_realtime():
+ async with thirdpart_gold.create_redis() as redis:
+ data = await thirdpart_gold.read_cache(redis)
+ await thirdpart_gold.schedule_refresh(redis, None, data)
+
+
+async def refresh_official():
+ async with thirdpart_gold.create_redis() as redis:
+ await official_gold.refresh_cache(redis)
+
+
+async def _periodic(job):
+ while True:
+ try:
+ await job()
+ except Exception:
+ logger.exception('periodic gold refresh check failed job=%s', job.__name__)
+ await asyncio.sleep(CHECK_INTERVAL_SECONDS)
+
+
+async def start_periodic_refresh():
+ if _periodic_tasks:
+ return
+ for job in (refresh_realtime, refresh_official):
+ _periodic_tasks.append(asyncio.create_task(_periodic(job), name=f'gold-periodic-{job.__name__}'))
+
+
+async def stop_periodic_refresh():
+ tasks = list(_periodic_tasks)
+ for task in tasks:
+ task.cancel()
+ await asyncio.gather(*tasks, return_exceptions=True)
+ _periodic_tasks.clear()
diff --git a/backend/api/utils/official_gold.py b/backend/api/utils/official_gold.py
new file mode 100644
index 00000000..0faefdff
--- /dev/null
+++ b/backend/api/utils/official_gold.py
@@ -0,0 +1,201 @@
+"""官方寄售缓存、认证失败识别及可选BARK通知。"""
+
+import asyncio
+from datetime import datetime, timezone
+import hashlib
+import json
+import logging
+import math
+import time
+from uuid import uuid4
+
+import aiohttp
+
+from api.core.Gzip import gzip_unzip, gzip_zip
+from api.types.colg import GOLD_REGION_CODES, OfficialGold
+from api.utils.thirdpart_gold import RELEASE_LOCK
+from config.main import config
+
+logger = logging.getLogger('dcalc.official_gold')
+URL = 'https://dzhu.qq.com/zangyi/activity/app'
+REGIONS = ('1', '2', '3A', '3B', '4', '5', '6', '7')
+CACHE_KEY = 'openapi:colg:real_time_gold_official:v2'
+LOCK_KEY = f'{CACHE_KEY}:lock'
+COOLDOWN_KEY = f'{CACHE_KEY}:cooldown'
+FRESH_SECONDS = 30 * 60
+RETENTION_SECONDS = 24 * 60 * 60
+ERROR_COOLDOWN_SECONDS = 5 * 60
+ALERT_RETENTION_SECONDS = 24 * 60 * 60
+
+
+class TokenExpired(RuntimeError):
+ pass
+
+
+def normalize_row(item: dict) -> dict:
+ """原报价为点券/100万金币，换算为每100点券的万金币为10000/报价。"""
+ ranks = {}
+ for name in ('goldDealRank', 'goldTopDealRank'):
+ prices = []
+ for raw in item[name][:5]:
+ try:
+ price = float(raw)
+ except (TypeError, ValueError, OverflowError):
+ continue
+ if math.isfinite(price):
+ prices.append(price)
+ ranks[name] = prices
+ ratios = [10000 / price for prices in ranks.values() for price in prices
+ if price > 0 and math.isfinite(10000 / price)]
+ return OfficialGold.model_validate({
+ **item, **ranks,
+ 'max': round(max(ratios), 4) if ratios else None,
+ 'min': round(min(ratios), 4) if ratios else None,
+ }).model_dump()
+
+
+def is_auth_failure(payload: dict) -> bool:
+ if any(str(payload.get(key)) == '-30003' for key in ('result', 'returnCode')):
+ return True
+ message = ' '.join(str(payload.get(key, '')) for key in ('returnMsg', 'message', 'msg')).lower()
+ return any(marker in message for marker in (
+ 'auth verification failed', 'token expired', 'token invalid', 'invalid token',
+ 'token失效', 'token已过期', '登录失效', '登录已过期', '请重新登录',
+ ))
+
+
+async def fetch_gold() -> list[dict]:
+ timeout = aiohttp.ClientTimeout(total=10)
+ async with aiohttp.ClientSession(timeout=timeout) as session:
+ async def fetch_region(region):
+ form = {'r': 'godRank/godDealRank', 'userId': config.HELPER_ID,
+ 'token': config.HELPER_TOKEN, 'serverNo': region}
+ async with session.post(URL, data=form) as resp:
+ if resp.status == 401:
+ raise TokenExpired('官方金币接口认证失败')
+ payload = await resp.json(content_type=None)
+ if not isinstance(payload, dict):
+ raise ValueError('官方金币接口返回格式异常')
+ if is_auth_failure(payload):
+ raise TokenExpired('官方金币接口认证失败')
+ resp.raise_for_status()
+ data = payload['data']
+ return normalize_row({
+ 'updataTime': str(int(datetime.now(timezone.utc).timestamp())),
+ 'region': f'跨{region}', 'code': GOLD_REGION_CODES[f'跨{region}'],
+ 'goldDealRank': [float(item['price']) for item in data['godDealRank'][:5]],
+ 'goldTopDealRank': [float(item['price']) for item in data['godTopDealRank'][:5]],
+ })
+
+ results = await asyncio.gather(*(fetch_region(region) for region in REGIONS),
+ return_exceptions=True)
+ # 任一跨区认证失败只向刷新调用方抛出一次；优先于其他渠道的网络异常。
+ for result in results:
+ if isinstance(result, TokenExpired):
+ raise result
+ for result in results:
+ if isinstance(result, BaseException):
+ raise result
+ return results
+
+
+async def notify_token_expired(redis) -> bool:
+ bark_url = (config.BARK_URL or '').strip()
+ if not bark_url:
+ return False
+ # 不将Token或BARK设备key写入缓存键、通知正文或日志。
+ fingerprint = hashlib.sha256(config.HELPER_TOKEN.encode()).hexdigest()
+ key = f'{CACHE_KEY}:token_alert:{fingerprint}'
+ owner = uuid4().hex
+ try:
+ if not await redis.set(key, owner, nx=True, ex=60):
+ return False
+ except Exception:
+ logger.warning('unable to acquire official token notification lock')
+ return False
+ sent = False
+ try:
+ timeout = aiohttp.ClientTimeout(total=5)
+ async with aiohttp.ClientSession(timeout=timeout) as session:
+ async with session.post(bark_url, json={
+ 'title': '助手Token失效',
+ 'body': '官方接口认证失败，请检查并更新配置',
+ 'group': 'dcalc',
+ }, allow_redirects=False) as resp:
+ resp.raise_for_status()
+ payload = await resp.json(content_type=None)
+ if not isinstance(payload, dict) or payload.get('code') != 200:
+ raise ValueError('BARK拒绝推送')
+ await redis.set(key, 'sent', ex=ALERT_RETENTION_SECONDS)
+ sent = True
+ return True
+ except Exception as exc:
+ # ClientResponseError包含完整请求URL，不输出异常正文或堆栈以免暴露设备key。
+ logger.warning('official token BARK notification failed (%s)', type(exc).__name__)
+ return False
+ finally:
+ if not sent:
+ try:
+ await redis.eval(RELEASE_LOCK, 1, key, owner)
+ except Exception:
+ logger.warning('unable to release official token notification lock')
+
+
+async def read_cache(redis) -> list[dict]:
+ value = await redis.get(CACHE_KEY)
+ if value is None:
+ return []
+ try:
+ rows = [normalize_row(item)
+ for item in json.loads(gzip_unzip(value))]
+ if [row['region'] for row in rows] != [f'跨{region}' for region in REGIONS]:
+ return []
+ if any(not 0 <= time.time() - int(row['updataTime']) < RETENTION_SECONDS for row in rows):
+ return []
+ return rows
+ except (ValueError, TypeError, KeyError, OSError):
+ logger.warning('invalid official gold cache')
+ return []
+
+
+def is_fresh(rows: list[dict]) -> bool:
+ return (len(rows) == len(REGIONS)
+ and all(0 <= time.time() - int(row['updataTime']) < FRESH_SECONDS for row in rows))
+
+
+async def refresh_cache(redis) -> bool:
+ if is_fresh(await read_cache(redis)) or await redis.get(COOLDOWN_KEY):
+ return False
+ owner = uuid4().hex
+ if not await redis.set(LOCK_KEY, owner, nx=True, ex=60):
+ return False
+ try:
+ if is_fresh(await read_cache(redis)) or await redis.get(COOLDOWN_KEY):
+ return False
+ try:
+ data = await fetch_gold()
+ except TokenExpired:
+ logger.warning('official gold authentication failed; check HELPER_TOKEN and HELPER_ID')
+ await notify_token_expired(redis)
+ except Exception:
+ logger.exception('fetch official gold failed')
+ else:
+ if await redis.get(LOCK_KEY) not in (owner, owner.encode()):
+ return False
+ await redis.set(CACHE_KEY, gzip_zip(json.dumps(data)), ex=RETENTION_SECONDS)
+ await redis.delete(COOLDOWN_KEY)
+ return True
+ # 网络异常、认证失败均保留上次完整缓存，不写空列表，不改变原更新时间。
+ await redis.set(COOLDOWN_KEY, '1', ex=ERROR_COOLDOWN_SECONDS)
+ return False
+ finally:
+ await redis.eval(RELEASE_LOCK, 1, LOCK_KEY, owner)
+
+
+async def get_data(redis) -> list[dict]:
+ cached = await read_cache(redis)
+ if cached:
+ return cached
+ # 冷启动时允许请求补齐；与定时任务共用刷新锁，不会重复请求上游。
+ await refresh_cache(redis)
+ return await read_cache(redis)
diff --git a/backend/api/utils/thirdpart_gold.py b/backend/api/utils/thirdpart_gold.py
new file mode 100644
index 00000000..ab1e8817
--- /dev/null
+++ b/backend/api/utils/thirdpart_gold.py
@@ -0,0 +1,305 @@
+"""YXDR 首次报价均值；Redis 缓存及跨进程刷新锁，所有 HTTP 使用现有 aiohttp。"""
+
+import asyncio
+import hashlib
+from html.parser import HTMLParser
+import json
+import logging
+import math
+import re
+from statistics import fmean
+import time
+from datetime import datetime, timedelta, timezone
+from urllib.parse import urljoin, urlsplit
+from uuid import uuid4
+
+import aiohttp
+from redis.asyncio import Redis
+from redis.asyncio.retry import Retry
+from redis.backoff import NoBackoff
+
+from api.types.colg import GOLD_REGION_CODES, ThirdPartGoldCache
+from config.main import config
+
+logger = logging.getLogger('dcalc.thirdpart_gold')
+BASE_URL = 'https://www.yxdr.com'
+REGIONS = {
+ 'kua1': '跨1', 'kua2': '跨2', 'kua3a': '跨3A', 'kua3b': '跨3B',
+ 'kua4': '跨4', 'kua5': '跨5', 'kua6': '跨6', 'kua7': '跨7', 'kua8': '跨8',
+}
+HEADERS = {
+ 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
+ 'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
+ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
+ 'Accept-Language': 'zh-CN,zh;q=0.9',
+}
+# 网站公开 DNF 页面已停用/移除的渠道；不请求 VIP 专属渠道。
+DISABLED_PLATFORMS = {1, 12, 21, 50, 51, 52, 53}
+PAGE_SIZE = 200
+FRESH_SECONDS = 30 * 60
+RETENTION_SECONDS = 24 * 60 * 60
+# 按每分钟最多8次页面访问处理，理论间隔7.5秒，取8秒留出边界余量。
+# 渠道在区内并发；上游仍可能限频，遇限频停止本轮并保留已有缓存。
+REGION_INTERVAL = 8
+LOCK_SECONDS = 15 * 60
+ERROR_COOLDOWN_SECONDS = 5 * 60
+REDIS_TIMEOUT_SECONDS = 1
+CACHE_PREFIX = 'openapi:colg:thirdpart_gold:v1'
+LOCK_KEY = f'{CACHE_PREFIX}:lock'
+COOLDOWN_KEY = f'{CACHE_PREFIX}:cooldown'
+RELEASE_LOCK = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+ return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+_refresh_tasks: set[asyncio.Task] = set()
+
+
+def create_redis():
+ # 只约束本接口；共享池没有显式超时，不能让缓存故障拖住 HTTP 请求。
+ # from_url 创建的连接池由客户端持有，aclose 时一并释放。
+ return Redis.from_url(config.REDIS_URL, protocol=2,
+ socket_connect_timeout=REDIS_TIMEOUT_SECONDS,
+ socket_timeout=REDIS_TIMEOUT_SECONDS,
+ retry=Retry(NoBackoff(), 0))
+
+
+async def get_redis():
+ async with create_redis() as redis:
+ yield redis
+
+
+def _refresh_done(task):
+ _refresh_tasks.discard(task)
+ if not task.cancelled():
+ error = task.exception()
+ if error:
+ logger.error('thirdpart gold background task failed',
+ exc_info=(type(error), error, error.__traceback__))
+
+
+async def launch_refresh(token: str):
+ # BackgroundTasks 只负责启动，HTTP/ASGI 请求不会跟随整轮刷新持续数分钟。
+ task = asyncio.create_task(refresh_cache(token), name='thirdpart-gold-refresh')
+ _refresh_tasks.add(task)
+ task.add_done_callback(_refresh_done)
+
+
+async def close_refresh_tasks():
+ tasks = list(_refresh_tasks)
+ for task in tasks:
+ task.cancel()
+ await asyncio.gather(*tasks, return_exceptions=True)
+
+
+class UpstreamLimited(RuntimeError):
+ pass
+
+
+class ConfigScripts(HTMLParser):
+ def __init__(self):
+ super().__init__()
+ self.sources = []
+
+ def handle_starttag(self, tag, attrs):
+ src = dict(attrs).get('src')
+ if tag == 'script' and src and 'CoinSale_' in src:
+ self.sources.append(src)
+
+
+def read_variable(source: str, name: str):
+ match = re.search(r'\b(?:window\.)?' + re.escape(name) + r'\s*=\s*', source)
+ if not match:
+ raise ValueError(f'YXDR 未下发 {name}')
+ return json.JSONDecoder().raw_decode(source, match.end())[0]
+
+
+def make_payload(channel: dict, seed: dict, user: dict) -> dict:
+ platform_id = json.loads(channel['data'])['PfId']
+ value = f"{seed['secret']}_{platform_id}_{channel['sign']}"
+ return {
+ 'data': channel['data'], 'sign': channel['sign'],
+ 'price1': 0, 'price2': 0, 'cross': 0, 'dnfCross': user['dnfCross'],
+ 'time': seed['time'], 'secret': hashlib.md5(value.encode('utf-8')).hexdigest(),
+ }
+
+
+def first_page_average(offers: list[dict]) -> tuple[float, int]:
+ """等权平均每条报价的 1/Price（万金币/元），不是 1/平均单价。"""
+ valid = []
+ for offer in offers:
+ try:
+ price = float(offer['Price'])
+ money = float(offer['Money'])
+ number = float(offer['Number'])
+ except (KeyError, TypeError, ValueError, OverflowError):
+ continue
+ if not all(math.isfinite(value) and value > 0 for value in (price, money, number)):
+ continue
+ if not math.isfinite(1 / price):
+ continue
+ valid.append({**offer, 'Price': price})
+ valid.sort(key=lambda item: (item['Price'], item['PfId'], item.get('BuyUrl') or ''))
+ seen, page = set(), []
+ for offer in valid:
+ # 与页面一致，仅对这些平台合并重复商品；7881 同 URL 可有不同规格。
+ key = (offer['PfId'], offer.get('BuyUrl'))
+ if offer['PfId'] in {1, 40, 41} and key[1]:
+ if key in seen:
+ continue
+ seen.add(key)
+ page.append(offer)
+ if len(page) == PAGE_SIZE:
+ break
+ if not page:
+ raise ValueError('YXDR 未返回有效报价')
+ return round(fmean(1 / item['Price'] for item in page), 4), len(page)
+
+
+async def _read_response(resp) -> str:
+ text = await resp.text()
+ if resp.status == 429 or 'LimitDeny' in text or '访问频率过高' in text:
+ raise UpstreamLimited('YXDR 访问频率受限')
+ resp.raise_for_status()
+ return text
+
+
+async def fetch_region(session: aiohttp.ClientSession, region: str) -> dict:
+ page_url = f'{BASE_URL}/bijiaqi/dnf/youxibi/{region}'
+ async with session.get(page_url) as resp:
+ page = await _read_response(resp)
+ seed = read_variable(page, 'bijiaSecret')
+ user = read_variable(page, 'bijiaUserInfo')
+ scripts = ConfigScripts()
+ scripts.feed(page)
+ if len(scripts.sources) != 1:
+ raise ValueError('YXDR 区服配置脚本异常')
+ config_url = urljoin(page_url, scripts.sources[0])
+ parts = urlsplit(config_url)
+ if parts.scheme != 'https' or parts.netloc != 'www.yxdr.com':
+ raise ValueError('YXDR 区服配置脚本地址异常')
+ async with session.get(config_url, headers={'Referer': page_url}) as resp:
+ config = await _read_response(resp)
+ channels = [channel for channel in read_variable(config, 'bijiaqiChanels')
+ if json.loads(channel['data'])['PfId'] not in DISABLED_PLATFORMS]
+ if not channels:
+ raise ValueError('YXDR 无可用公开渠道')
+
+ async def fetch_channel(channel):
+ platform = json.loads(channel['data'])
+ if platform['CoinUnit'] != '万金币':
+ raise ValueError('YXDR 金币单位发生变化')
+ headers = {'Referer': page_url, 'Origin': BASE_URL,
+ 'X-Requested-With': 'XMLHttpRequest', 'Accept': 'application/json'}
+ async with session.post(f'{BASE_URL}/bijia/coinsale',
+ json=make_payload(channel, seed, user), headers=headers) as resp:
+ result = json.loads(await _read_response(resp))
+ if not result.get('success') or not isinstance(result.get('data'), list):
+ raise ValueError(f"YXDR 渠道 {platform['PfId']}/{platform['CoinNo']} 抓取失败")
+ return [{**offer, 'PfId': platform['PfId']} for offer in result['data']]
+
+ # 同区并发，避免短时签名在顺序请求中失效；任何渠道失败都不覆盖旧均值。
+ results = await asyncio.gather(*(fetch_channel(channel) for channel in channels),
+ return_exceptions=True)
+ # 优先传播限频，即使另一个渠道先失败，也不能继续抓下一个区。
+ for result in results:
+ if isinstance(result, UpstreamLimited):
+ raise result
+ for result in results:
+ if isinstance(result, BaseException):
+ raise result
+ price, count = first_page_average([offer for result in results for offer in result])
+ now = datetime.now(timezone(timedelta(hours=8)))
+ return {'region': REGIONS[region], 'code': GOLD_REGION_CODES[REGIONS[region]],
+ 'goldPrice': price, 'sampleCount': count,
+ 'updataTime': int(now.timestamp()), 'fetchedAt': now.timestamp()}
+
+
+async def read_cache(redis) -> list[dict]:
+ values = await redis.mget([f'{CACHE_PREFIX}:{region}' for region in REGIONS])
+ result = []
+ for region, value in zip(REGIONS, values):
+ if value is None:
+ continue
+ try:
+ item = json.loads(value)
+ age = time.time() - item['fetchedAt']
+ if item['region'] != REGIONS[region] or not 0 <= age < RETENTION_SECONDS:
+ continue
+ # 兼容此前缓存的格式化日期及缺少code的记录，无须重新抓取所有跨区。
+ result.append(ThirdPartGoldCache.model_validate({
+ **item, 'updataTime': int(item['fetchedAt']),
+ 'code': GOLD_REGION_CODES[REGIONS[region]], 'stale': age >= FRESH_SECONDS,
+ }).model_dump())
+ except (KeyError, TypeError, ValueError):
+ logger.warning('invalid thirdpart gold cache region=%s', region)
+ return result
+
+
+async def schedule_refresh(redis, background_tasks, data: list[dict]) -> bool:
+ # 缓存均新鲜时不访问上游；失败缺区时遵守全局冷却期。
+ if len(data) == len(REGIONS) and not any(item['stale'] for item in data):
+ return False
+ if await redis.get(COOLDOWN_KEY):
+ return False
+ token = uuid4().hex
+ if not await redis.set(LOCK_KEY, token, nx=True, ex=LOCK_SECONDS):
+ return False
+ try:
+ if await redis.get(COOLDOWN_KEY):
+ await redis.eval(RELEASE_LOCK, 1, LOCK_KEY, token)
+ return False
+ if background_tasks is None:
+ await launch_refresh(token)
+ else:
+ background_tasks.add_task(launch_refresh, token)
+ except BaseException:
+ await redis.eval(RELEASE_LOCK, 1, LOCK_KEY, token)
+ raise
+ return True
+
+
+async def refresh_cache(token: str):
+ # BackgroundTasks 中不用请求依赖的 Redis 客户端，它可能已被依赖清理器关闭。
+ async with create_redis() as redis:
+ try:
+ if await redis.get(LOCK_KEY) not in (token, token.encode()):
+ return
+ cached = await read_cache(redis)
+ fresh_regions = {item['region'] for item in cached if not item['stale']}
+ pending = [region for region, name in REGIONS.items() if name not in fresh_regions]
+ successful, last_start = 0, None
+ timeout = aiohttp.ClientTimeout(total=15)
+ async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout,
+ connector=aiohttp.TCPConnector(limit=5)) as session:
+ for region in pending:
+ if last_start is not None:
+ await asyncio.sleep(max(0, REGION_INTERVAL - (time.monotonic() - last_start)))
+ # 不在租约到期/已被其他刷新取代后继续请求或写缓存。
+ if await redis.get(LOCK_KEY) not in (token, token.encode()):
+ return
+ last_start = time.monotonic()
+ try:
+ item = await fetch_region(session, region)
+ if await redis.get(LOCK_KEY) not in (token, token.encode()):
+ return
+ await redis.set(f'{CACHE_PREFIX}:{region}', json.dumps(item),
+ ex=RETENTION_SECONDS)
+ successful += 1
+ except UpstreamLimited:
+ logger.warning('thirdpart gold rate limited region=%s', region)
+ break
+ except Exception:
+ logger.exception('fetch thirdpart gold failed region=%s', region)
+ # 新鲜数据由每区 fetchedAt 控制30分钟周期；只对失败设置长冷却。
+ # 下次重试只补缺失/过期区，避免每次重抓前8区导致最后一区一直受限。
+ cooldown = 1 if successful == len(pending) else ERROR_COOLDOWN_SECONDS
+ await redis.set(COOLDOWN_KEY, '1', ex=cooldown)
+ except Exception:
+ logger.exception('refresh thirdpart gold cache failed')
+ finally:
+ try:
+ await redis.eval(RELEASE_LOCK, 1, LOCK_KEY, token)
+ except Exception:
+ logger.exception('release thirdpart gold refresh lock failed')
diff --git a/backend/config/main.py b/backend/config/main.py
index 0ae393a4..ba947137 100644
--- a/backend/config/main.py
+++ b/backend/config/main.py
@@ -14,6 +14,7 @@ class Config(BaseSettings):
COLG_TOKEN: str = ''
HELPER_TOKEN: str = ''
HELPER_ID: str = ''
+ BARK_URL: str | None = None
model_config = SettingsConfigDict(
env_file=(
diff --git a/backend/main.py b/backend/main.py
index 575fea63..976b36b2 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -9,6 +9,8 @@ from fastapi import FastAPI
from api.core import Exceptions, Gzip, Middleware
from api.core.Logging import configure_logging
from api.main import api_router
+from api.utils.thirdpart_gold import close_refresh_tasks
+from api.utils.gold_refresh import start_periodic_refresh, stop_periodic_refresh
from config.main import config
from database.connect import close_connection_pools
@@ -20,9 +22,12 @@ async def lifespan(_: FastAPI):
print('dcalc is starting...')
print('Application starting...')
print('Application startup complete.')
+ await start_periodic_refresh()
try:
yield
finally:
+ await stop_periodic_refresh()
+ await close_refresh_tasks()
await close_connection_pools()
diff --git a/backend/tests/unit/test_colg_official.py b/backend/tests/unit/test_colg_official.py
index 0d7689e5..2b1f0df5 100644
--- a/backend/tests/unit/test_colg_official.py
+++ b/backend/tests/unit/test_colg_official.py
@@ -1,9 +1,15 @@
import asyncio
+import time
+import json
-from api.routes import colg
+import pytest
+
+from api.utils import official_gold as gold
+from api.core.Gzip import gzip_zip
class FakeResponse:
+ status = 200
def __init__(self, region):
self.region = region
@@ -20,8 +26,8 @@ class FakeResponse:
await asyncio.sleep(0)
return {
'data': {
- 'godDealRank': [{'price': '100'}, {'price': '101'}],
- 'godTopDealRank': [{'price': '200'}, {'price': '201'}],
+ 'godDealRank': [{'price': value} for value in ['100.5', '101', '102', '103', '104', '105']],
+ 'godTopDealRank': [{'price': value} for value in ['200.25', '201', '202', '203', '204', '205']],
}
}
@@ -51,15 +57,259 @@ def test_fetch_official_gold_uses_async_http_with_isolated_forms(monkeypatch):
sessions.append(session)
return session
- monkeypatch.setattr(colg.aiohttp, 'ClientSession', create_session)
+ monkeypatch.setattr(gold.aiohttp, 'ClientSession', create_session)
- result = asyncio.run(colg._fetch_official_gold())
+ result = asyncio.run(gold.fetch_gold())
- assert [data['serverNo'] for _, data in calls] == list(colg.OFFICIAL_GOLD_REGIONS)
- assert all(url == colg.OFFICIAL_GOLD_URL for url, _ in calls)
- assert len({id(data) for _, data in calls}) == len(colg.OFFICIAL_GOLD_REGIONS)
- assert [item['region'] for item in result] == [f'跨{region}' for region in colg.OFFICIAL_GOLD_REGIONS]
- assert result[0]['goldDealRank'] == [100, 101]
- assert result[0]['goldTopDealRank'] == [200, 201]
+ assert [data['serverNo'] for _, data in calls] == list(gold.REGIONS)
+ assert all(url == gold.URL for url, _ in calls)
+ assert len({id(data) for _, data in calls}) == len(gold.REGIONS)
+ assert [item['region'] for item in result] == [f'跨{region}' for region in gold.REGIONS]
+ assert [item['code'] for item in result] == ['k1', 'k2', 'k3a', 'k3b', 'k4', 'k5', 'k6', 'k7']
+ assert all(isinstance(item['updataTime'], str) and item['updataTime'].isdigit() for item in result)
+ assert abs(int(result[0]['updataTime']) - time.time()) < 2
+ assert result[0]['goldDealRank'] == [100.5, 101, 102, 103, 104]
+ assert result[0]['goldTopDealRank'] == [200.25, 201, 202, 203, 204]
+ assert result[0]['max'] == round(10000 / 100.5, 4)
+ assert result[0]['min'] == round(10000 / 204, 4)
assert sessions[0].kwargs['timeout'].total == 10
- assert 'serverNo' not in colg.payload_helper
+
+
+def test_official_route_uses_new_cache_contract_and_numeric_rank_arrays(monkeypatch, client):
+ calls = []
+
+ async def cached(redis):
+ calls.append(redis)
+ return [{'updataTime': '1793711716', 'region': '跨3B', 'code': 'k3b',
+ 'goldDealRank': [100.5, 101], 'goldTopDealRank': [200.25, 201],
+ 'max': round(10000 / 100.5, 4), 'min': round(10000 / 201, 4)}]
+
+ monkeypatch.setattr(gold, 'get_data', cached)
+ body = client.get('/api/open/gold/official/').json()
+ assert set(body) == {'code', 'message', 'data'}
+ assert body['code'] == 200
+ assert len(calls) == 1
+ row = body['data'][0]
+ assert set(row) == {'updataTime', 'region', 'code', 'goldDealRank', 'goldTopDealRank', 'max', 'min'}
+ assert isinstance(row['updataTime'], str) and row['updataTime'].isdigit()
+ assert row['code'] == 'k3b'
+ assert row['goldDealRank'] == [100.5, 101]
+ assert row['goldTopDealRank'] == [200.25, 201]
+ assert row['max'] == round(10000 / 100.5, 4)
+ assert row['min'] == round(10000 / 201, 4)
+ for rank in (row['goldDealRank'], row['goldTopDealRank']):
+ assert len(rank) <= 5
+ assert all(isinstance(value, (int, float)) for value in rank)
+
+
+def test_gold_openapi_matches_timestamp_types_and_required_region_codes(client):
+ schemas = client.app.openapi()['components']['schemas']
+ realtime = schemas['RealTimeGold']
+ official = schemas['OfficialGold']
+ assert realtime['properties']['updataTime']['type'] == 'integer'
+ assert official['properties']['updataTime']['type'] == 'string'
+ assert 'code' in realtime['required'] and 'code' in official['required']
+ assert realtime['properties']['goldPrice']['type'] == 'number'
+ for name in ('goldDealRank', 'goldTopDealRank'):
+ assert official['properties'][name]['maxItems'] == 5
+ assert official['properties'][name]['items']['type'] == 'number'
+
+
+class FakeRedis:
+ def __init__(self):
+ self.values = {}
+ self.ttls = {}
+
+ async def get(self, key):
+ return self.values.get(key)
+
+ async def set(self, key, value, nx=False, ex=None):
+ if nx and key in self.values:
+ return False
+ self.values[key] = value
+ self.ttls[key] = ex
+ return True
+
+ async def delete(self, key):
+ return int(self.values.pop(key, None) is not None)
+
+ async def eval(self, script, number, key, owner):
+ if self.values.get(key) == owner:
+ return await self.delete(key)
+ return 0
+
+
+def cached_rows(age=0):
+ return [{'updataTime': str(int(time.time() - age)), 'region': f'跨{region}',
+ 'code': f'k{region.lower()}', 'goldDealRank': [100.5], 'goldTopDealRank': [200.25]}
+ for region in gold.REGIONS]
+
+
+def test_official_converts_point_prices_to_gold_per_100_points():
+ row = {**cached_rows()[0], 'goldDealRank': [125, 150], 'goldTopDealRank': [200, 250]}
+ result = gold.normalize_row(row)
+ assert result['max'] == 80
+ assert result['min'] == 40
+ assert result['goldDealRank'] == [125, 150]
+ assert result['goldTopDealRank'] == [200, 250]
+
+
+def test_conversion_ignores_nonpositive_invalid_prices_and_uses_null_when_empty():
+ row = {**cached_rows()[0], 'goldDealRank': [0, -1, float('nan'), float('inf')],
+ 'goldTopDealRank': []}
+ result = gold.normalize_row(row)
+ assert result['max'] is None and result['min'] is None
+ row['goldTopDealRank'] = [125]
+ result = gold.normalize_row(row)
+ assert result['max'] == result['min'] == 80
+
+
+def test_old_official_cache_gets_max_and_min_without_refetch():
+ redis = FakeRedis()
+ redis.values[gold.CACHE_KEY] = gzip_zip(json.dumps(cached_rows()))
+ rows = asyncio.run(gold.read_cache(redis))
+ assert len(rows) == 8
+ assert rows[0]['max'] == round(10000 / 100.5, 4)
+ assert rows[0]['min'] == round(10000 / 200.25, 4)
+
+
+@pytest.mark.parametrize('payload', [
+ {'returnCode': -30003, 'returnMsg': 'auth verification failed'},
+ {'result': '-30003'}, {'message': 'token expired'}, {'msg': '请重新登录'},
+])
+def test_expired_credentials_are_classified(payload):
+ assert gold.is_auth_failure(payload)
+
+
+@pytest.mark.parametrize('payload', [
+ {'returnCode': 500, 'message': 'server error'}, {'message': 'network timeout'}, {'data': {}},
+])
+def test_regular_upstream_failures_do_not_trigger_expiry_alert(payload):
+ assert not gold.is_auth_failure(payload)
+
+
+def test_fetch_detects_actual_auth_failure_response(monkeypatch):
+ class Response(FakeResponse):
+ async def json(self, **kwargs):
+ return {'result': -30003, 'returnCode': -30003, 'returnMsg': 'auth verification failed'}
+
+ class Session(FakeSession):
+ def post(self, url, data):
+ return Response(data['serverNo'])
+
+ monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
+ with pytest.raises(gold.TokenExpired):
+ asyncio.run(gold.fetch_gold())
+
+
+def test_refresh_updates_cache_without_http_request_and_skips_when_fresh(monkeypatch):
+ redis = FakeRedis()
+ calls = []
+
+ async def fetch():
+ calls.append(1)
+ return cached_rows()
+
+ monkeypatch.setattr(gold, 'fetch_gold', fetch)
+ assert asyncio.run(gold.refresh_cache(redis)) is True
+ assert len(asyncio.run(gold.read_cache(redis))) == 8
+ assert redis.ttls[gold.CACHE_KEY] == gold.RETENTION_SECONDS
+ assert gold.LOCK_KEY not in redis.values
+ assert asyncio.run(gold.refresh_cache(redis)) is False
+ assert calls == [1]
+
+
+def test_failed_refresh_preserves_cache_and_notifies_once_per_fetch(monkeypatch):
+ redis = FakeRedis()
+ old = gzip_zip(json.dumps(cached_rows(age=1900)))
+ redis.values[gold.CACHE_KEY] = old
+ alerts = []
+
+ async def fetch():
+ raise gold.TokenExpired()
+
+ async def notify(redis):
+ alerts.append(1)
+
+ monkeypatch.setattr(gold, 'fetch_gold', fetch)
+ monkeypatch.setattr(gold, 'notify_token_expired', notify)
+ assert asyncio.run(gold.refresh_cache(redis)) is False
+ assert redis.values[gold.CACHE_KEY] == old
+ assert alerts == [1]
+ assert gold.LOCK_KEY not in redis.values
+ assert redis.ttls[gold.COOLDOWN_KEY] == gold.ERROR_COOLDOWN_SECONDS
+ assert asyncio.run(gold.refresh_cache(redis)) is False
+ assert alerts == [1]
+
+
+def test_network_failure_preserves_cache_without_token_alert(monkeypatch):
+ redis = FakeRedis()
+ old = gzip_zip(json.dumps(cached_rows(age=1900)))
+ redis.values[gold.CACHE_KEY] = old
+
+ async def fetch():
+ raise TimeoutError('upstream unavailable')
+
+ async def notify(redis):
+ raise AssertionError('network failures are not credential failures')
+
+ monkeypatch.setattr(gold, 'fetch_gold', fetch)
+ monkeypatch.setattr(gold, 'notify_token_expired', notify)
+ assert asyncio.run(gold.refresh_cache(redis)) is False
+ assert redis.values[gold.CACHE_KEY] == old
+
+
+@pytest.mark.parametrize('url', [None, '', ' '])
+def test_unconfigured_bark_never_creates_http_session(monkeypatch, url):
+ monkeypatch.setattr(gold.config, 'BARK_URL', url)
+
+ def session(**kwargs):
+ raise AssertionError('BARK must be completely skipped when not configured')
+
+ monkeypatch.setattr(gold.aiohttp, 'ClientSession', session)
+ redis = FakeRedis()
+ assert asyncio.run(gold.notify_token_expired(redis)) is False
+ assert not redis.values
+
+
+def test_bark_alert_is_deduplicated_and_does_not_include_credentials(monkeypatch):
+ redis = FakeRedis()
+ posts = []
+ monkeypatch.setattr(gold.config, 'BARK_URL', 'https://bark.example/device-secret')
+ monkeypatch.setattr(gold.config, 'HELPER_TOKEN', 'private-token')
+
+ class Response(FakeResponse):
+ async def json(self, **kwargs):
+ return {'code': 200}
+
+ class Session(FakeSession):
+ def post(self, url, json, **kwargs):
+ posts.append((url, json))
+ return Response('push')
+
+ monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
+ assert asyncio.run(gold.notify_token_expired(redis)) is True
+ assert asyncio.run(gold.notify_token_expired(redis)) is False
+ assert len(posts) == 1
+ assert posts[0][0] == gold.config.BARK_URL
+ assert 'private-token' not in json.dumps(posts[0][1])
+ assert 'device-secret' not in json.dumps(posts[0][1])
+ assert all('private-token' not in key for key in redis.values)
+ assert next(iter(redis.ttls.values())) == gold.ALERT_RETENTION_SECONDS
+ monkeypatch.setattr(gold.config, 'HELPER_TOKEN', 'new-token')
+ assert asyncio.run(gold.notify_token_expired(redis)) is True
+ assert len(posts) == 2
+
+
+def test_failed_bark_push_can_retry_without_leaking_device_url(monkeypatch, caplog):
+ monkeypatch.setattr(gold.config, 'BARK_URL', 'https://bark.example/device-secret')
+ redis = FakeRedis()
+
+ class Session(FakeSession):
+ def post(self, url, **kwargs):
+ raise RuntimeError(url)
+
+ monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
+ assert asyncio.run(gold.notify_token_expired(redis)) is False
+ assert not redis.values
+ assert 'device-secret' not in caplog.text
diff --git a/backend/tests/unit/test_config.py b/backend/tests/unit/test_config.py
index b7a771e3..8177fb0c 100644
--- a/backend/tests/unit/test_config.py
+++ b/backend/tests/unit/test_config.py
@@ -3,3 +3,9 @@ from config.main import Config
def test_config_ignores_extra_environment_fields():
assert Config.model_config.get('extra') == 'ignore'
+
+
+def test_bark_has_no_default_destination(monkeypatch):
+ monkeypatch.delenv('BARK_URL', raising=False)
+ assert Config(_env_file=None).BARK_URL is None
+ assert Config(_env_file=None, BARK_URL='https://bark.example/my-device').BARK_URL == 'https://bark.example/my-device'
diff --git a/backend/tests/unit/test_gold_refresh.py b/backend/tests/unit/test_gold_refresh.py
new file mode 100644
index 00000000..56d85f1e
--- /dev/null
+++ b/backend/tests/unit/test_gold_refresh.py
@@ -0,0 +1,99 @@
+import asyncio
+
+from api.utils import gold_refresh, thirdpart_gold
+
+
+def test_both_periodic_jobs_run_without_requests_and_stop_cleanly(monkeypatch):
+ async def run():
+ calls = [0, 0]
+ ready = [asyncio.Event(), asyncio.Event()]
+
+ async def realtime():
+ calls[0] += 1
+ if calls[0] >= 2:
+ ready[0].set()
+
+ async def official():
+ calls[1] += 1
+ if calls[1] >= 2:
+ ready[1].set()
+
+ monkeypatch.setattr(gold_refresh, 'refresh_realtime', realtime)
+ monkeypatch.setattr(gold_refresh, 'refresh_official', official)
+ monkeypatch.setattr(gold_refresh, 'CHECK_INTERVAL_SECONDS', .01)
+ try:
+ await gold_refresh.start_periodic_refresh()
+ await gold_refresh.start_periodic_refresh()
+ assert len(gold_refresh._periodic_tasks) == 2
+ await asyncio.wait_for(asyncio.gather(*(event.wait() for event in ready)), 1)
+ finally:
+ await gold_refresh.stop_periodic_refresh()
+ assert not gold_refresh._periodic_tasks
+ snapshot = list(calls)
+ await asyncio.sleep(.02)
+ assert calls == snapshot
+
+ asyncio.run(run())
+
+
+def test_periodic_job_survives_transient_failure(monkeypatch):
+ async def run():
+ calls = []
+ ready = asyncio.Event()
+
+ async def job():
+ calls.append(1)
+ if len(calls) == 1:
+ raise ConnectionError('Redis unavailable')
+ ready.set()
+
+ monkeypatch.setattr(gold_refresh, 'CHECK_INTERVAL_SECONDS', .01)
+ task = asyncio.create_task(gold_refresh._periodic(job))
+ try:
+ await asyncio.wait_for(ready.wait(), 1)
+ assert len(calls) >= 2
+ finally:
+ task.cancel()
+ await asyncio.gather(task, return_exceptions=True)
+
+ asyncio.run(run())
+
+
+def test_realtime_can_launch_without_request_background_tasks(monkeypatch):
+ class Redis:
+ async def get(self, key):
+ return None
+
+ async def set(self, *args, **kwargs):
+ return True
+
+ launched = []
+
+ async def launch(token):
+ launched.append(token)
+
+ monkeypatch.setattr(thirdpart_gold, 'launch_refresh', launch)
+ assert asyncio.run(thirdpart_gold.schedule_refresh(Redis(), None, [])) is True
+ assert len(launched) == 1
+
+
+def test_application_lifecycle_starts_and_stops_refresh_jobs(monkeypatch):
+ import main
+ calls = []
+
+ def callback(name):
+ async def call():
+ calls.append(name)
+ return call
+
+ monkeypatch.setattr(main, 'start_periodic_refresh', callback('start'))
+ monkeypatch.setattr(main, 'stop_periodic_refresh', callback('stop'))
+ monkeypatch.setattr(main, 'close_refresh_tasks', callback('refresh_closed'))
+ monkeypatch.setattr(main, 'close_connection_pools', callback('pools_closed'))
+
+ async def run():
+ async with main.lifespan(main.application):
+ assert calls == ['start']
+
+ asyncio.run(run())
+ assert calls == ['start', 'stop', 'refresh_closed', 'pools_closed']
diff --git a/backend/tests/unit/test_thirdpart_gold.py b/backend/tests/unit/test_thirdpart_gold.py
new file mode 100644
index 00000000..60fd1c28
--- /dev/null
+++ b/backend/tests/unit/test_thirdpart_gold.py
@@ -0,0 +1,418 @@
+import asyncio
+import hashlib
+import json
+import time
+
+import pytest
+from fastapi import BackgroundTasks
+
+from api.routes import colg
+from api.types.colg import GOLD_REGION_CODES
+from api.utils import thirdpart_gold as gold
+
+
+def offer(price, url='listing', platform=17, money=1):
+ return {'Price': price, 'Money': money, 'Number': 1, 'PfId': platform, 'BuyUrl': url}
+
+
+def cached_item(region='跨1', age=0):
+ fetched_at = time.time() - age
+ return {'region': region, 'goldPrice': 100.0, 'sampleCount': 10,
+ 'code': GOLD_REGION_CODES[region], 'updataTime': int(fetched_at),
+ 'fetchedAt': fetched_at}
+
+
+class FakeRedis:
+ def __init__(self):
+ self.values = {}
+ self.writes = []
+
+ async def __aenter__(self):
+ return self
+
+ async def __aexit__(self, *args):
+ return False
+
+ async def mget(self, keys):
+ return [self.values.get(key) for key in keys]
+
+ async def get(self, key):
+ return self.values.get(key)
+
+ async def set(self, key, value, nx=False, ex=None):
+ if nx and key in self.values:
+ return False
+ self.values[key] = value
+ self.writes.append((key, value, ex))
+ return True
+
+ async def eval(self, script, number, key, token):
+ if self.values.get(key) == token:
+ del self.values[key]
+ return 1
+ return 0
+
+
+def test_average_is_mean_of_ratios_without_minimum_money_filter():
+ # 100 and 50 万/元 => 75; inverse mean price would incorrectly give 66.67.
+ assert gold.first_page_average([offer(.01), offer(.02)]) == (75.0, 2)
+ invalid = [offer(0), offer(-1), offer(float('nan')), offer(float('inf')),
+ offer(.01, money=-1), offer(None), offer(.01, money=float('nan'))]
+ assert gold.first_page_average(invalid + [offer(.01, money=.5)]) == (100.0, 1)
+ with pytest.raises(ValueError, match='有效报价'):
+ gold.first_page_average(invalid)
+
+
+def test_page_deduplicates_before_cutoff_and_preserves_7881_variants():
+ rows = [offer(.01, 'duplicate', 40), offer(.01, 'duplicate', 40)]
+ rows += [offer(.02, str(i), 17) for i in range(199)]
+ rows += [offer(.5, 'outside', 17)]
+ assert gold.first_page_average(rows) == (50.25, 200)
+ assert gold.first_page_average([offer(.01, platform=44), offer(.02, platform=44)]) == (75.0, 2)
+
+
+def test_request_signature_matches_observed_browser_request():
+ channel = {'data': '{"PfId":17,"CoinNo":"yxb"}',
+ 'sign': 'e33b1f23fdbe4677e5e6f8bc7bdcbf78'}
+ seed = {'time': '639292586508765510', 'secret': 'a30ad49b1b2532c403de6319ec4f1c1a'}
+ payload = gold.make_payload(channel, seed, {'dnfCross': 1})
+ assert payload['secret'] == 'b7ac9fd4e4f15735dcbe1411b627bb4e'
+ assert payload['data'] == channel['data']
+ assert payload['time'] == seed['time']
+ assert payload['price1'] == payload['price2'] == 0
+
+
+class FakeResponse:
+ status = 200
+
+ def __init__(self, text):
+ self.body = text
+
+ async def __aenter__(self):
+ return self
+
+ async def __aexit__(self, *args):
+ return False
+
+ async def text(self):
+ return self.body
+
+ def raise_for_status(self):
+ pass
+
+
+def test_region_fetches_its_own_seed_config_and_all_initial_channels():
+ seed = {'time': 'fresh-time', 'secret': 'fresh-seed'}
+ channel = {'data': json.dumps({'PfId': 17, 'CoinNo': 'yxb', 'CoinUnit': '万金币'}),
+ 'sign': 'config-sign'}
+ page = ('window.bijiaSecret=' + json.dumps(seed) + ';'
+ 'window.bijiaUserInfo={"dnfCross":1};'
+ '')
+ posts, gets = [], []
+
+ class Session:
+ def get(self, url, **kwargs):
+ gets.append(url)
+ return FakeResponse(page if '/bijiaqi/' in url else
+ 'window.bijiaqiChanels=' + json.dumps([channel]) + ';')
+
+ def post(self, url, json, **kwargs):
+ posts.append((url, json))
+ return FakeResponse('{"success":1,"data":[{"Price":0.01,"Money":1,"Number":1}]}')
+
+ result = asyncio.run(gold.fetch_region(Session(), 'kua8'))
+ assert result['region'] == '跨8'
+ assert result['code'] == 'k8'
+ assert isinstance(result['updataTime'], int)
+ assert result['goldPrice'] == 100
+ assert result['sampleCount'] == 1
+ assert gets == [gold.BASE_URL + '/bijiaqi/dnf/youxibi/kua8',
+ gold.BASE_URL + '/cate/1838/CoinSale_yxb_254_0.js?ver=935']
+ assert posts[0][1]['secret'] == hashlib.md5(b'fresh-seed_17_config-sign').hexdigest()
+ assert posts[0][1]['time'] == 'fresh-time'
+ assert 'page' not in posts[0][1]
+
+
+def test_rate_limited_html_does_not_become_empty_quotes():
+ with pytest.raises(gold.UpstreamLimited):
+ asyncio.run(gold._read_response(FakeResponse('LimitDeny')))
+
+
+def test_cache_omits_missing_regions_marks_old_values_and_retains_order():
+ redis = FakeRedis()
+ redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps(cached_item(age=1900))
+ redis.values[f'{gold.CACHE_PREFIX}:kua2'] = json.dumps(cached_item('跨2', age=90000))
+ redis.values[f'{gold.CACHE_PREFIX}:kua8'] = json.dumps(cached_item('跨8'))
+ rows = asyncio.run(gold.read_cache(redis))
+ assert [row['region'] for row in rows] == ['跨1', '跨8']
+ assert [row['stale'] for row in rows] == [True, False]
+
+
+def test_invalid_cached_average_is_not_served():
+ redis = FakeRedis()
+ redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps({**cached_item(), 'goldPrice': 0})
+ redis.values[f'{gold.CACHE_PREFIX}:kua8'] = json.dumps({**cached_item('跨8'), 'sampleCount': 0})
+ assert asyncio.run(gold.read_cache(redis)) == []
+
+
+def test_legacy_cache_gets_epoch_and_region_code_without_refetching():
+ redis = FakeRedis()
+ legacy = cached_item('跨3A')
+ legacy.pop('code')
+ legacy['updataTime'] = '2026-11-03 21:13:17'
+ redis.values[f'{gold.CACHE_PREFIX}:kua3a'] = json.dumps(legacy)
+ rows = asyncio.run(gold.read_cache(redis))
+ assert len(rows) == 1
+ assert rows[0]['code'] == 'k3a'
+ assert rows[0]['updataTime'] == int(legacy['fetchedAt'])
+ assert not redis.writes
+
+
+def test_cold_concurrent_requests_schedule_only_one_refresh():
+ redis = FakeRedis()
+ tasks = BackgroundTasks()
+
+ async def run():
+ return await asyncio.gather(*(gold.schedule_refresh(redis, tasks, []) for _ in range(10)))
+
+ assert sum(asyncio.run(run())) == 1
+ assert len(tasks.tasks) == 1
+ assert tasks.tasks[0].func is gold.launch_refresh
+ assert next(ttl for key, _, ttl in redis.writes if key == gold.LOCK_KEY) == gold.LOCK_SECONDS
+
+
+def test_fresh_cache_and_failure_cooldown_do_not_schedule_upstream():
+ redis = FakeRedis()
+ tasks = BackgroundTasks()
+ rows = [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]
+ assert asyncio.run(gold.schedule_refresh(redis, tasks, rows)) is False
+ redis.values[gold.COOLDOWN_KEY] = '1'
+ assert asyncio.run(gold.schedule_refresh(redis, tasks, [])) is False
+ assert not tasks.tasks
+ assert gold.LOCK_KEY not in redis.values
+
+
+@pytest.mark.parametrize('limited', [False, True])
+def test_refresh_preserves_failed_region_and_releases_its_lock(monkeypatch, limited):
+ redis = FakeRedis()
+ redis.values[gold.LOCK_KEY] = 'owner'
+ old = json.dumps(cached_item(age=1900))
+ redis.values[f'{gold.CACHE_PREFIX}:kua1'] = old
+ monkeypatch.setattr(gold, 'create_redis', lambda: redis)
+ monkeypatch.setattr(gold, 'REGIONS', {'kua1': '跨1', 'kua8': '跨8'})
+ monkeypatch.setattr(gold, 'REGION_INTERVAL', 0)
+ calls = []
+
+ async def fetch(session, region):
+ calls.append(region)
+ if region == 'kua1':
+ raise gold.UpstreamLimited() if limited else ValueError('channel failed')
+ return cached_item('跨8')
+
+ monkeypatch.setattr(gold, 'fetch_region', fetch)
+ asyncio.run(gold.refresh_cache('owner'))
+ assert redis.values[f'{gold.CACHE_PREFIX}:kua1'] == old
+ assert gold.LOCK_KEY not in redis.values
+ assert redis.values[gold.COOLDOWN_KEY] == '1'
+ if limited:
+ assert calls == ['kua1']
+ else:
+ assert calls == ['kua1', 'kua8']
+ assert f'{gold.CACHE_PREFIX}:kua8' in redis.values
+
+
+def test_refresh_cannot_delete_another_owners_lock(monkeypatch):
+ redis = FakeRedis()
+ redis.values[gold.LOCK_KEY] = 'replacement'
+ monkeypatch.setattr(gold, 'create_redis', lambda: redis)
+ asyncio.run(gold.refresh_cache('expired-owner'))
+ assert redis.values[gold.LOCK_KEY] == 'replacement'
+
+
+def test_route_serializes_existing_contract_and_hides_internal_timestamp(monkeypatch, client):
+ async def read(redis):
+ return [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]
+
+ async def schedule(*args):
+ return False
+
+ monkeypatch.setattr(gold, 'read_cache', read)
+ monkeypatch.setattr(gold, 'schedule_refresh', schedule)
+ body = client.get('/api/open/gold/thirdpart/').json()
+ assert body['code'] == 200
+ assert len(body['data']) == 9
+ for row in body['data']:
+ assert set(row) == {'updataTime', 'region', 'code', 'goldPrice'}
+ schema = client.app.openapi()
+ assert schema['paths']['/api/open/gold/thirdpart/']['get']['operationId'] == 'thirdpartGold'
+ assert 'ThirdPartGoldCache' not in schema['components']['schemas']
+
+
+def test_realtime_uses_thirdpart_cache_and_returns_only_defined_fields(monkeypatch, client):
+ async def read(redis):
+ return [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]
+
+ schedules = []
+
+ async def schedule(redis, tasks, data):
+ schedules.append(data)
+ return False
+
+ def old_colg_request(*args, **kwargs):
+ raise AssertionError('realtime must no longer request COLG')
+
+ monkeypatch.setattr(gold, 'read_cache', read)
+ monkeypatch.setattr(gold, 'schedule_refresh', schedule)
+ monkeypatch.setattr(colg.requests, 'post', old_colg_request)
+ body = client.get('/api/open/gold/realtime/').json()
+ assert set(body) == {'code', 'message', 'data'}
+ assert body['code'] == 200
+ assert len(body['data']) == 9
+ assert len(schedules) == 1
+ for name, row in zip(gold.REGIONS.values(), body['data']):
+ assert set(row) == {'updataTime', 'region', 'code', 'goldPrice'}
+ assert row['region'] == name
+ assert row['code'] == GOLD_REGION_CODES[name]
+ assert isinstance(row['updataTime'], int)
+ assert isinstance(row['goldPrice'], (int, float))
+
+
+def test_realtime_initialization_still_returns_contract_fields(monkeypatch, client):
+ async def read(redis):
+ return [{**cached_item('跨2'), 'stale': False}]
+
+ async def schedule(*args):
+ return False
+
+ monkeypatch.setattr(gold, 'read_cache', read)
+ monkeypatch.setattr(gold, 'schedule_refresh', schedule)
+ body = client.get('/api/open/gold/realtime/').json()
+ assert body['code'] == 202
+ assert len(body['data']) == 1
+ assert set(body['data'][0]) == {'updataTime', 'region', 'code', 'goldPrice'}
+ assert body['data'][0]['code'] == 'k2'
+
+
+def test_route_cold_start_and_redis_failure_do_not_block_or_scrape(monkeypatch):
+ redis = FakeRedis()
+ tasks = BackgroundTasks()
+ result = asyncio.run(colg.get_thirdpart_gold(redis, tasks))
+ assert result.code == 202 and result.data == []
+ assert len(tasks.tasks) == 1
+
+ async def unavailable(redis):
+ raise ConnectionError('Redis unavailable')
+
+ monkeypatch.setattr(gold, 'read_cache', unavailable)
+ tasks = BackgroundTasks()
+ result = asyncio.run(colg.get_thirdpart_gold(redis, tasks))
+ assert result.code == 503 and result.data == []
+ assert not tasks.tasks
+
+
+def test_route_partial_cache_reports_initialization_progress(monkeypatch):
+ async def read(redis):
+ return [{**cached_item(), 'stale': False}]
+
+ async def schedule(*args):
+ return False
+
+ monkeypatch.setattr(gold, 'read_cache', read)
+ monkeypatch.setattr(gold, 'schedule_refresh', schedule)
+ result = asyncio.run(colg.get_thirdpart_gold(None, BackgroundTasks()))
+ assert result.code == 202
+ assert '1/9' in result.message
+ assert len(result.data) == 1
+
+
+def test_background_launcher_finishes_before_refresh_completes(monkeypatch):
+ async def run():
+ started, finish = asyncio.Event(), asyncio.Event()
+
+ async def refresh(token):
+ started.set()
+ await finish.wait()
+
+ monkeypatch.setattr(gold, 'refresh_cache', refresh)
+ tasks = BackgroundTasks()
+ tasks.add_task(gold.launch_refresh, 'owner')
+ try:
+ await asyncio.wait_for(tasks(), .5)
+ await asyncio.wait_for(started.wait(), .5)
+ assert len(gold._refresh_tasks) == 1
+ assert not next(iter(gold._refresh_tasks)).done()
+ finally:
+ finish.set()
+ await gold.close_refresh_tasks()
+ assert not gold._refresh_tasks
+
+ asyncio.run(run())
+
+
+def test_shutdown_cancels_refresh_and_releases_redis_lock(monkeypatch):
+ redis = FakeRedis()
+ redis.values[gold.LOCK_KEY] = 'owner'
+ monkeypatch.setattr(gold, 'create_redis', lambda: redis)
+
+ async def run():
+ started = asyncio.Event()
+
+ async def fetch(session, region):
+ started.set()
+ await asyncio.Event().wait()
+
+ monkeypatch.setattr(gold, 'fetch_region', fetch)
+ try:
+ await gold.launch_refresh('owner')
+ await asyncio.wait_for(started.wait(), .5)
+ finally:
+ await gold.close_refresh_tasks()
+ assert gold.LOCK_KEY not in redis.values
+ assert not gold._refresh_tasks
+
+ asyncio.run(run())
+
+
+def test_retry_only_fetches_missing_regions_instead_of_repeating_fresh_pages(monkeypatch):
+ redis = FakeRedis()
+ redis.values[gold.LOCK_KEY] = 'owner'
+ for region, name in gold.REGIONS.items():
+ if region != 'kua8':
+ redis.values[f'{gold.CACHE_PREFIX}:{region}'] = json.dumps(cached_item(name))
+ monkeypatch.setattr(gold, 'create_redis', lambda: redis)
+ calls = []
+
+ async def fetch(session, region):
+ calls.append(region)
+ return cached_item(gold.REGIONS[region])
+
+ monkeypatch.setattr(gold, 'fetch_region', fetch)
+ asyncio.run(gold.refresh_cache('owner'))
+ assert calls == ['kua8']
+ assert len(asyncio.run(gold.read_cache(redis))) == 9
+
+
+def test_unresponsive_redis_returns_prompt_error_instead_of_hanging(monkeypatch):
+ async def run():
+ connections = []
+
+ async def blackhole(reader, writer):
+ connections.append(writer)
+ await reader.read()
+
+ server = await asyncio.start_server(blackhole, '127.0.0.1', 0)
+ port = server.sockets[0].getsockname()[1]
+ monkeypatch.setattr(gold.config, 'REDIS_URL', f'redis://127.0.0.1:{port}/0')
+ try:
+ async with gold.create_redis() as redis:
+ result = await asyncio.wait_for(colg.get_thirdpart_gold(redis, BackgroundTasks()), 2.5)
+ assert result.code == 503
+ assert result.data == []
+ finally:
+ server.close()
+ for writer in connections:
+ writer.close()
+ await writer.wait_closed()
+ await server.wait_closed()
+
+ asyncio.run(run())
diff --git a/scripts/yxdr_coinsale.py b/scripts/yxdr_coinsale.py
new file mode 100644
index 00000000..c53fad6f
--- /dev/null
+++ b/scripts/yxdr_coinsale.py
@@ -0,0 +1,311 @@
+#!/usr/bin/env python3
+"""Fetch the initial YXDR DNF coin offers; Python 3 standard library only.
+
+Examples (run from the repository root):
+ python3 scripts/yxdr_coinsale.py
+ python3 scripts/yxdr_coinsale.py --regions kua1 --price1 100
+ python3 scripts/yxdr_coinsale.py --regions kua3a kua3b --limit 200
+
+The site has nine region entries: kua1, kua2, kua3a, kua3b, kua4..kua8.
+For EACH region, fetch fresh bijiaSecret from its HTML, then load the referenced
+CoinSale_*.js. Each bijiaqiChanels entry supplies an opaque data STRING and sign.
+POST secret = md5(bijiaSecret.secret + '_' + PfId + '_' + sign), with the
+unchanged data/sign and the page's time. No page/pageSize parameter is present
+in the site's coinsale request. We save every initial response and additionally
+export the cheapest --limit filtered offers (default 200). This cutoff is ours:
+the UI also has platform-preservation, deduplication and random tie-breaking.
+
+Do not reuse a captured secret/time across runs or reconstruct the data string:
+its original representation is signed. Public unauthenticated use only; respect
+the site's access limits. Default region-start interval is 60 seconds. A limit
+response stops the entire run, preserving results already written. No retries.
+"""
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+from copy import copy
+import csv
+import hashlib
+import json
+import math
+from datetime import datetime, timezone
+from html.parser import HTMLParser
+from http.cookiejar import CookieJar
+from pathlib import Path
+import re
+import sys
+import time
+from urllib.error import HTTPError, URLError
+from urllib.parse import urljoin, urlsplit
+from urllib.request import HTTPCookieProcessor, Request, build_opener
+
+
+BASE = "https://www.yxdr.com"
+REGIONS = {
+ "kua1": "跨一区", "kua2": "跨二区", "kua3a": "跨三A区",
+ "kua3b": "跨三B区", "kua4": "跨四区", "kua5": "跨五区",
+ "kua6": "跨六区", "kua7": "跨七区", "kua8": "跨八区",
+}
+HEADERS = {
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/131.0.0.0 Safari/537.36",
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+ "Accept-Language": "zh-CN,zh;q=0.9",
+}
+# Disabled/removed in the public DNF coinsale frontend observed 2026-11-02.
+DISABLED_PLATFORMS = {1, 12, 21, 50, 51, 52, 53}
+
+
+class RateLimited(RuntimeError):
+ pass
+
+
+class Scripts(HTMLParser):
+ def __init__(self):
+ super().__init__()
+ self.sources = []
+
+ def handle_starttag(self, tag, attrs):
+ if tag == "script":
+ src = dict(attrs).get("src")
+ if src:
+ self.sources.append(src)
+
+
+def read_variable(source, name):
+ """Decode a JSON assignment without evaluating arbitrary site JavaScript."""
+ match = re.search(r"\b(?:window\.)?" + re.escape(name) + r"\s*=\s*", source)
+ if not match:
+ raise ValueError(f"找不到 {name}；页面可能受限、发生跳转或结构已变化")
+ return json.JSONDecoder().raw_decode(source, match.end())[0]
+
+
+class Client:
+ def __init__(self, timeout):
+ self.cookies = CookieJar()
+ self.opener = build_opener(HTTPCookieProcessor(self.cookies))
+ self.timeout = timeout
+
+ def fork(self):
+ client = Client(self.timeout)
+ for cookie in self.cookies:
+ client.cookies.set_cookie(copy(cookie))
+ return client
+
+ def request(self, url, payload=None, referer=None):
+ parts = urlsplit(url)
+ if parts.scheme != "https" or parts.netloc != "www.yxdr.com":
+ raise ValueError(f"拒绝将请求发送到站外地址：{url}")
+ headers = dict(HEADERS)
+ body = None
+ if referer:
+ headers["Referer"] = referer
+ if payload is not None:
+ headers.update({"Content-Type": "application/json; charset=UTF-8",
+ "Accept": "application/json, text/javascript, */*; q=0.01",
+ "Origin": BASE, "X-Requested-With": "XMLHttpRequest"})
+ body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+ try:
+ with self.opener.open(Request(url, data=body, headers=headers),
+ timeout=self.timeout) as response:
+ source = response.read().decode("utf-8-sig")
+ except HTTPError as exc:
+ source = exc.read().decode("utf-8", errors="replace")
+ if exc.code == 429 or "LimitDeny" in source or "访问频率过高" in source:
+ raise RateLimited("站点限制访问频率，已停止；请稍后再运行") from exc
+ raise RuntimeError(f"HTTP {exc.code}: {url}") from exc
+ if "LimitDeny" in source or "访问频率过高" in source:
+ raise RateLimited("站点限制访问频率，已停止；请稍后再运行")
+ return source
+
+
+def make_payload(channel, seed, user, price1, price2):
+ platform = json.loads(channel["data"])
+ secret_input = f"{seed['secret']}_{platform['PfId']}_{channel['sign']}"
+ return {
+ "data": channel["data"], # Keep the signed STRING byte-for-byte.
+ "sign": channel["sign"],
+ "price1": price1, "price2": price2,
+ "cross": 0, "dnfCross": user["dnfCross"],
+ "time": seed["time"],
+ "secret": hashlib.md5(secret_input.encode("utf-8")).hexdigest(),
+ }
+
+
+def select_offers(rows, price1, price2, limit):
+ selected = []
+ for row in rows:
+ money, number, price = row["Money"], row["Number"], row["Price"]
+ if money != -1 and (money * number < 5 or price == 0):
+ continue
+ if price1 and money * number < price1:
+ continue
+ if price2 and "ling" not in (row.get("TradeType") or "") and money > price2:
+ continue
+ if money >= 1_000_000:
+ continue
+ selected.append(dict(row))
+ selected.sort(key=lambda row: (row["Price"], row["PfId"], row.get("BuyUrl") or ""))
+ # The frontend merges repeated listing URLs for these platforms.
+ result, seen = [], set()
+ for row in selected:
+ key = (row["PfId"], row.get("BuyUrl"))
+ if row["PfId"] in {1, 40, 41} and key[1]:
+ if key in seen:
+ continue
+ seen.add(key)
+ result.append(row)
+ return result[:limit]
+
+
+def fetch_region(client, region, args):
+ # Get the canonical page; filter values are explicitly sent to coinsale.
+ # The supplied query URL can redirect to the canonical page on a fresh visit.
+ page_url = f"{BASE}/bijiaqi/dnf/youxibi/{region}"
+ page = client.request(page_url)
+ seed = read_variable(page, "bijiaSecret")
+ user = read_variable(page, "bijiaUserInfo")
+ scripts = Scripts()
+ scripts.feed(page)
+ config_src = next((src for src in scripts.sources if "CoinSale_" in src), None)
+ if not config_src:
+ raise ValueError("页面没有下发 CoinSale 区服配置")
+ config_url = urljoin(page_url, config_src)
+ config = client.request(config_url, referer=page_url)
+ channels = read_variable(config, "bijiaqiChanels")
+ report = {
+ "region": region, "region_name": REGIONS[region], "source": page_url,
+ "config_url": config_url, "fetched_at": datetime.now(timezone.utc).isoformat(),
+ "price1": args.price1, "price2": args.price2, "limit": args.limit,
+ "channels": [], "items": [],
+ }
+ rows, requests = [], []
+ for channel in channels:
+ platform = json.loads(channel["data"])
+ pf_id = platform["PfId"]
+ entry = {"platform_id": pf_id, "platform_name": platform["PfName"],
+ "coin_no": platform["CoinNo"], "server": platform["GsName"]}
+ report["channels"].append(entry)
+ if pf_id in DISABLED_PLATFORMS:
+ entry["status"] = "skipped_by_public_frontend"
+ continue
+ payload = make_payload(channel, seed, user, args.price1, args.price2)
+ requests.append((client.fork(), platform, payload, entry))
+
+ def fetch_channel(task):
+ session, platform, payload, entry = task
+ pf_id = platform["PfId"]
+ channel_rows = []
+ try:
+ response = json.loads(session.request(f"{BASE}/bijia/coinsale", payload, page_url))
+ entry["response"] = response # Preserve failed/empty responses, too.
+ if response.get("success") and isinstance(response.get("data"), list):
+ entry["status"] = "success"
+ entry["count"] = len(response["data"])
+ for item in response["data"]:
+ channel_rows.append({**item, "PfId": pf_id, "PfName": platform["PfName"],
+ "CoinNo": platform["CoinNo"], "Region": region,
+ "RegionName": REGIONS[region],
+ "CoinUnit": platform["CoinUnit"],
+ "SourceServer": platform["GsName"]})
+ else:
+ entry["status"] = "failed"
+ entry["error"] = response.get("debug") or "接口返回 success=0 或 data 格式异常"
+ except RateLimited:
+ raise
+ except (RuntimeError, URLError, TimeoutError, ValueError) as exc:
+ entry.update(status="failed", error=str(exc))
+ print(f" {platform['PfName']}/{platform['CoinNo']}: "
+ f"{entry['status']} {entry.get('count', 0)} 条"
+ f" {entry.get('error', '')}", flush=True)
+ return channel_rows
+
+ # The webpage dispatches its channel requests together. Send them while the
+ # freshly issued signing material is current, with an isolated cookie jar
+ # per worker; keep result ordering deterministic.
+ with ThreadPoolExecutor(max_workers=5) as pool:
+ for channel_rows in pool.map(fetch_channel, requests):
+ rows.extend(channel_rows)
+ report["raw_offer_count"] = len(rows)
+ report["items"] = select_offers(rows, args.price1, args.price2, args.limit)
+ active = [entry for entry in report["channels"]
+ if entry["status"] != "skipped_by_public_frontend"]
+ successful = sum(entry["status"] == "success" for entry in active)
+ report["status"] = ("success" if active and successful == len(active)
+ else "partial" if successful else "failed")
+ return report
+
+
+def save_region(directory, report):
+ name = report["region"]
+ (directory / f"{name}.json").write_text(
+ json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+ fields = ["Region", "RegionName", "PfId", "PfName", "CoinNo", "SourceServer",
+ "Price", "Amount", "CoinUnit", "Money", "Number", "TradeType", "Seller",
+ "State", "BuyUrl", "ListUrl"]
+ with (directory / f"{name}.csv").open("w", encoding="utf-8-sig", newline="") as handle:
+ writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
+ writer.writeheader()
+ writer.writerows(report["items"])
+
+
+def nonnegative(value):
+ number = float(value)
+ if not math.isfinite(number) or number < 0:
+ raise argparse.ArgumentTypeError("必须是非负有限数值")
+ return number
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__,
+ formatter_class=argparse.RawDescriptionHelpFormatter)
+ parser.add_argument("--regions", nargs="+", choices=REGIONS, default=list(REGIONS))
+ parser.add_argument("--price1", type=nonnegative, default=0, help="最低总价，元")
+ parser.add_argument("--price2", type=nonnegative, default=0, help="最高总价，0 表示不限")
+ parser.add_argument("--limit", type=int, default=200)
+ parser.add_argument("--interval", type=nonnegative, default=60,
+ help="相邻跨区开始请求的最短间隔秒数，默认60")
+ parser.add_argument("--timeout", type=nonnegative, default=30)
+ parser.add_argument("--output", type=Path,
+ default=Path("tmp/yxdr") / datetime.now().strftime("%Y%m%d-%H%M%S"))
+ args = parser.parse_args()
+ if args.limit < 1 or args.timeout <= 0:
+ parser.error("limit 和 timeout 必须大于0")
+ if args.price2 and args.price1 > args.price2:
+ parser.error("price1 不能大于 price2")
+ args.output.mkdir(parents=True, exist_ok=True)
+ client = Client(args.timeout)
+ summary, last_start, failed = [], None, False
+ for region in dict.fromkeys(args.regions):
+ if last_start is not None:
+ remaining = args.interval - (time.monotonic() - last_start)
+ if remaining > 0:
+ print(f"等待 {remaining:.1f} 秒后获取 {REGIONS[region]}", flush=True)
+ time.sleep(remaining)
+ last_start = time.monotonic()
+ print(f"获取 {REGIONS[region]} ({region})", flush=True)
+ stop = False
+ try:
+ report = fetch_region(client, region, args)
+ save_region(args.output, report)
+ status = report["status"]
+ summary.append({"region": region, "status": status,
+ "raw_count": report["raw_offer_count"],
+ "exported_count": len(report["items"])})
+ failed |= status != "success"
+ except (RateLimited, RuntimeError, URLError, TimeoutError, ValueError) as exc:
+ failed = True
+ summary.append({"region": region, "status": "failed", "error": str(exc)})
+ print(f" 失败：{exc}", file=sys.stderr, flush=True)
+ stop = isinstance(exc, RateLimited)
+ (args.output / "summary.json").write_text(
+ json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+ if stop:
+ break
+ print(f"输出目录：{args.output.resolve()}", flush=True)
+ return 1 if failed else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())