"""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 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 api.utils.gold_time import refresh_slot 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 SAMPLE_SLICE = slice(3, 10) 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:v2' 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]: """匹配公开页面 i.item1 的第4～10条，平均显示单价（元/万金币）。""" 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) for value in (price, money, number)) or price < 0: continue # 页面默认展示条件，不额外设置price1/price2价格过滤。 if money != -1 and (money * number < 5 or price == 0): continue if money >= 1_000_000 or (offer.get('Seller') or '').startswith('智莹科技'): 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 selected = page[SAMPLE_SLICE] if not selected: raise ValueError('YXDR 未返回有效报价') # 页面 Math.floorDigits(price, 5) 先toFixed(6)再删掉末位。 prices = [float(f"{item['Price']:.6f}"[:-1]) for item in selected] return round(fmean(prices), 2), len(prices) 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, *, started_at: float | None = None) -> dict: if started_at is None: started_at = time.time() 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]) return {'region': REGIONS[region], 'code': GOLD_REGION_CODES[REGIONS[region]], 'goldPrice': price, 'sampleCount': count, 'updataTime': refresh_slot(started_at), 'fetchedAt': started_at} 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': refresh_slot(item['fetchedAt']), 'code': GOLD_REGION_CODES[REGIONS[region]], 'stale': refresh_slot(item['fetchedAt']) < refresh_slot(), }).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 started_at = time.time() 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, started_at=started_at) 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')