"""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')