# -*- coding:utf-8 -*- # @Software :PyCharm # @Project :LOL-DeepWinPredictor # @Path :/Data_CrawlProcess # @FileName :LPL.py # @Time :2025/4/21 23:36 # @Author :Viper373 # @GitHub :https://github.com/Viper373 # @Home :https://viper3.top # @Blog :https://blog.viper3.top import asyncio import os import re from functools import partial from typing import Dict, List import aiohttp import orjson import requests from pymongo import ASCENDING, errors from Data_CrawlProcess import env from Data_CrawlProcess.Process import Process from tool_utils.log_utils import RichLogger from tool_utils.mongo_utils import MongoUtils from tool_utils.progress_utils import RichProgressUtils process = Process() rich_logger = RichLogger() mongo_utils = MongoUtils() class LPL: def __init__(self, rich_progress=None): """ LPL类初始化方法。 :param rich_progress: RichProgressUtils实例（用于全局进度条） :return: None """ self.url = "https://lpl.qq.com/web201612/data/LOL_MATCH2_MATCH_HOMEPAGE_BMATCH_LIST_{}.js" self.match_url = "https://open.tjstats.com/match-auth-app/open/v1/compound/matchDetail?matchId={}" self.seasons_url = "https://lol.qq.com/act/AutoCMS/publish/LOLWeb/EventdataTab/EventdataTab.js" self.headers = { 'sec-ch-ua': '"Microsoft Edge";v="125", "Chromium";v="125", "Not.A/Brand";v="24"', 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Referer': 'https://lpl.qq.com/web202301/schedule.html', 'X-Requested-With': 'XMLHttpRequest', 'sec-ch-ua-mobile': '?0', 'User-Agent': env.UA, # UA头 'Authorization': env.AUTHORIZATION, # 身份验证 'sec-ch-ua-platform': '"Windows"', } self.seasonsIds_headers = { 'accept': '*/*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6,ja;q=0.5,ko;q=0.4,fr;q=0.3', 'cache-control': 'no-cache', 'pragma': 'no-cache', 'priority': 'u=2', 'referer': 'https://lpl.qq.com/web202301/schedule.html', 'sec-ch-ua': '"Microsoft Edge";v="135", "Not-A.Brand";v="8", "Chromium";v="135"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'script', 'sec-fetch-mode': 'no-cors', 'sec-fetch-site': 'same-origin', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36 Edg/135.0.0.0', } self.cookies = { 'tj_rp_did': '60013b4b18ce11efa938fe48ef6d951e', } self.proxies = env.PROXIES # 代理 self.rich_progress = rich_progress async def auto_seasonIds(self, col_name: str, rich_progress) -> None: """ 自动从LPL官网获取赛季数据，合并env.py中的SEASONS并写入MongoDB。 :param col_name: MongoDB集合名称 :param rich_progress: RichProgressUtils实例 :return: None """ rich_progress = rich_progress or RichProgressUtils() collection = mongo_utils.use_collection(col_name) try: loop = asyncio.get_event_loop() request_func = partial( requests.get, url=self.seasons_url, headers=self.seasonsIds_headers, proxies=self.proxies ) response = await loop.run_in_executor(None, request_func, *tuple()) response.encoding = 'utf-8' match = re.search(r'return\s+(\[.*?\])[\s;]*}\)', response.text) if not match: rich_logger.error("[LPL] auto_seasonIds: 未找到JSON数据") return json_str = match.group(1) data = orjson.loads(json_str) seasons = [] for season in data: for game in season.get('domestic', []): seasons.append({ "name": game.get("gameName"), "id": game.get("iGameId"), "type": game.get("sGameType"), "url": game.get("url") }) for game in season.get('abroad', []): seasons.append({ "name": game.get("gameName"), "id": game.get("iGameId"), "type": game.get("sGameType"), "url": game.get("url") }) rich_logger.info(f"[LPL] auto_seasonIds: 官网获取到 {len(seasons)} 个赛季数据") except Exception as e: rich_logger.error(f"[LPL] auto_seasonIds: 获取官网赛季数据失败: {e}") return def get_existing_seasons(): if not os.path.exists(env.ENV_SEASONS): return [] with open(env.ENV_SEASONS, 'r', encoding='utf-8') as f: content = f.read() match = re.search(r'SEASONS\s*=\s*(\[.*?\])', content, re.DOTALL) if not match: return [] try: return orjson.loads(match.group(1)) except: return [] existing_seasons = get_existing_seasons() all_ids = set() merged_seasons = [] for s in seasons + existing_seasons: if s['id'] not in all_ids: merged_seasons.append(s) all_ids.add(s['id']) rich_logger.info(f"[LPL] auto_seasonIds: 合并后共有 {len(merged_seasons)} 个赛季数据（已自动去重）") content = "SEASONS = " + orjson.dumps(merged_seasons, option=orjson.OPT_INDENT_2).decode('utf-8') + "\n" if os.path.exists(env.ENV_SEASONS): with open(env.ENV_SEASONS, 'r', encoding='utf-8') as f: existing_content = f.read() if 'SEASONS = [' in existing_content: new_content = re.sub(r'SEASONS = \[.*?\]', content, existing_content, flags=re.DOTALL) else: new_content = existing_content + '\n' + content else: new_content = content with open(env.ENV_SEASONS, 'w', encoding='utf-8') as f: f.write(new_content) rich_logger.info(f"[LPL] auto_seasonIds: 赛季数据已成功更新到env.py文件！") # 4. 写入MongoDB collection.create_index([("id", ASCENDING)], unique=True) queue = asyncio.Queue(maxsize=20) fetch_task_id = rich_progress.add_task("[LPL] auto_seasonIds生产", total=len(merged_seasons)) store_task_id = rich_progress.add_task("[LPL] auto_seasonIds入库", total=len(merged_seasons)) async def producer(): for season in merged_seasons: await queue.put(season) rich_progress.advance(fetch_task_id) await queue.put(None) async def consumer(): count = 0 while True: item = await queue.get() if item is None: break try: collection.insert_one(item) except errors.DuplicateKeyError: pass count += 1 rich_progress.advance(store_task_id) rich_progress.update(store_task_id, completed=len(merged_seasons)) await asyncio.gather(producer(), consumer()) rich_logger.info(f"[LPL] auto_seasonIds: 赛季数据已全部入库，共{len(merged_seasons)}条") @staticmethod async def get_seasonIds(col_name: str, seasons: Dict[str, str], rich_progress) -> None: """ 异步生产者-消费者：获取赛季ID并写入数据库。 :param col_name: MongoDB集合名称 :param seasons: 赛季ID映射字典{str: str} :param rich_progress: RichProgressUtils实例 :return: None """ rich_progress = rich_progress or RichProgressUtils() collection = mongo_utils.use_collection(col_name) collection.create_index([("season_id", ASCENDING)], unique=True) queue = asyncio.Queue(maxsize=20) fetch_task_id = rich_progress.add_task("[LPL] seasonIDs生产", total=len(seasons)) store_task_id = rich_progress.add_task("[LPL] seasonIDs入库", total=len(seasons)) async def producer(): for season_name, season_id in seasons.items(): await queue.put({"season_name": season_name, "season_id": season_id}) rich_progress.advance(fetch_task_id) await queue.put(None) async def consumer(): count = 0 while True: item = await queue.get() if item is None: break try: collection.insert_one(item) except errors.DuplicateKeyError: pass count += 1 rich_progress.advance(store_task_id) rich_progress.update(store_task_id, completed=len(seasons)) await asyncio.gather(producer(), consumer()) rich_logger.info(f"爬取完成丨共计[{len(seasons)}]LPL_season") async def get_bMatchIds(self, col_name: str, seasons: Dict[str, str], rich_progress) -> None: """ 异步生产者-消费者：获取bMatchId并写入数据库。 :param col_name: MongoDB集合名称 :param seasons: 赛季ID映射字典{str: str} :param rich_progress: RichProgressUtils实例 :return: None """ rich_progress = rich_progress or self.rich_progress collection = mongo_utils.use_collection(col_name) collection.create_index([("bMatchId", ASCENDING)], unique=True) queue = asyncio.Queue(maxsize=50) all_records = [] # 先收集所有bMatchId async def fetch_all_records(): async def fetch_season_data(_url: str): try: loop = asyncio.get_event_loop() request_func = partial( requests.get, url=_url, headers=self.headers, cookies=self.cookies, proxies=self.proxies ) response = await loop.run_in_executor(None, request_func) if response.status_code != 200: rich_logger.error(f"获取bMatchId失败，状态码: {response.status_code}") return seasons_data = orjson.loads(response.text) if seasons_data.get('status') == "0": msg = seasons_data.get('msg') for match in msg: record = { 'GameName': match.get('GameName'), 'bMatchName': match.get('bMatchName'), 'MatchDate': match.get('MatchDate'), 'bMatchId': match.get('bMatchId') } all_records.append(record) except Exception as _error: rich_logger.error(f"爬取[LPL]bMatchId错误: {_error}") urls = [self.url.format(season_id) for season_id in seasons.values()] await asyncio.gather(*(fetch_season_data(url) for url in urls)) await fetch_all_records() fetch_task_id = rich_progress.add_task("[LPL] bMatchId生产", total=len(all_records)) store_task_id = rich_progress.add_task("[LPL] bMatchId入库", total=len(all_records)) async def producer_queue(): for record in all_records: await queue.put(record) rich_progress.advance(fetch_task_id) await queue.put(None) async def consumer(): count = 0 while True: item = await queue.get() if item is None: break try: collection.insert_one(item) except errors.DuplicateKeyError: pass count += 1 rich_progress.advance(store_task_id) rich_progress.update(store_task_id, completed=len(all_records)) await asyncio.gather(producer_queue(), consumer()) rich_logger.info(f"爬取完成丨LPL_bMatchId已全部入库") async def get_match_data(self, bmatch_ids: List[str], col_name: str, rich_progress=None) -> None: """ 生产者-消费者解耦：高并发爬取match_data并结构化入库，进度独立。 :param bmatch_ids: 比赛ID列表 :param col_name: MongoDB集合名称 :param rich_progress: RichProgressUtils实例 :return: None """ collection = mongo_utils.use_collection(col_name) session_timeout = aiohttp.ClientTimeout(total=30) fetch_task_id = rich_progress.add_task("[LPL] match_data爬取", total=len(bmatch_ids)) process_task_id = rich_progress.add_task("[LPL] process_data入库", total=len(bmatch_ids)) hero_win_rates = process.read_win_rate() process_queue = asyncio.Queue(maxsize=500) total_count = len(bmatch_ids) async def fetcher(): async with aiohttp.ClientSession(timeout=session_timeout) as session: sem = asyncio.Semaphore(50) async def fetch_one(bid): url = self.match_url.format(bid) async with sem: try: async with session.get(url, headers=self.headers, cookies=self.cookies) as resp: data = await resp.json() if data.get('success') and data.get('data'): match = data['data'] await process_queue.put(match) except Exception as e: rich_logger.error(f"[LPL] match_data爬取失败: {bid} {e}") self.rich_progress.advance(fetch_task_id) await asyncio.gather(*(fetch_one(bid) for bid in bmatch_ids)) await process_queue.put(None) async def processor(): count = 0 while True: match = await process_queue.get() if match is None: break try: processed_data = {} teamAId, teamAName, teamBId, teamBName = match['teamAId'], match['teamAName'], match['teamBId'], match['teamBName'] matchWin = 1 if match["matchWin"] == teamAId else 0 processed_data.update({ "teamAId": teamAId, "teamAName": teamAName, "teamBId": teamBId, "teamBName": teamBName, "matchWin": matchWin, }) matchInfos = match["matchInfos"] for bo in matchInfos: teamInfos = bo["teamInfos"] team_po = "A" for team in teamInfos: playerInfos = team["playerInfos"] count_ = 1 for player in playerInfos: playerLocation = "ADC" if player["playerLocation"] == "BOT" else player["playerLocation"] heroId, heroTitle, heroName = player["heroId"], player["heroTitle"], player["heroName"] key = f"{heroId}{playerLocation[:3].upper()}" heroWinRate = hero_win_rates.get(key, 0.50) processed_data.update({ f"{team_po}{count_}playerLocation": playerLocation, f"{team_po}{count_}heroId": heroId, f"{team_po}{count_}heroName": f"{heroTitle}-{heroName}", f"{team_po}{count_}heroWinRate": heroWinRate, }) count_ += 1 team_po = "B" if isinstance(processed_data, dict): collection.insert_one(processed_data) else: rich_logger.error(f"[LPL] processed_data类型错误: {type(processed_data)} | bMatchId: {match.get('bMatchId', 'unknown')}") except Exception as e: match_id = match.get('bMatchId', 'unknown') if isinstance(match, dict) else 'unknown' snippet = orjson.dumps(match)[:200].decode('utf-8') if isinstance(match, dict) else str(match)[:200] rich_logger.error(f"[LPL] process_data单条处理失败: {str(e)} | bMatchId: {match_id} | 数据片段: {snippet} ...") count += 1 rich_progress.advance(process_task_id) rich_progress.update(process_task_id, completed=total_count) await asyncio.gather(fetcher(), processor()) rich_logger.info(f"[LPL] match_data已全部入库") return None async def main(self, lpl_db: str, col_season: str, col_bmatch: str, col_match: str, seasons: dict) -> None: """ LPL主流程，自动串联所有流程：抓取seasonIds、bMatchIds、match_data。 :param lpl_db: LPL专用MongoDB数据库名称 :param col_season: 赛季ID集合名 :param col_bmatch: bMatchId集合名 :param col_match: 结构化比赛数据集合名 :param seasons: 赛季ID映射字典{str: str} :return: None """ mongo_utils.use_db(lpl_db) await self.get_seasonIds(col_season, seasons, self.rich_progress) await self.get_bMatchIds(col_bmatch, seasons, self.rich_progress) bmatch_ids = [item['bMatchId'] for item in mongo_utils.use_collection(col_bmatch).find({}, {'bMatchId': 1, '_id': 0})] await self.get_match_data(bmatch_ids, col_match, self.rich_progress) rich_logger.info("[LPL] main流程执行完毕")