diff --git a/plugins/genshin/farming/_spider.py b/plugins/genshin/farming/_spider.py index 7edc5672..d595b01c 100644 --- a/plugins/genshin/farming/_spider.py +++ b/plugins/genshin/farming/_spider.py @@ -20,16 +20,16 @@ P = ParamSpec("P") def get_material_serial_name(names: Iterable[str]) -> str: + """获取材料的系列名，找出相同字符即可""" counter = None for name in names: - if counter is None: - counter = Counter(name) - continue - counter = counter & Counter(name) + counter = (counter or Counter(name)) & Counter(name) return "".join(counter.keys()).replace("的", "").replace("之", "") class Spider(ABC): + """每日素材表爬虫的基类""" + _lock = RLock() _client: AsyncClient | None = None @@ -47,25 +47,29 @@ class Spider(ABC): @abstractmethod async def __call__(self) -> dict[int, FarmingData]: - pass + """爬虫的具体实现的方法""" @classmethod @final async def execute(cls, assets: AssetsService) -> dict[int, FarmingData]: + """根据爬虫的优先级来爬取每日素材表（优先级高的先爬，失败了再换优先级低的爬）""" result = None for spider in sorted( map(lambda x: x(assets), cls.__subclasses__()), key=lambda x: (x.priority, x.__class__.__name__) ): - if (result := await spider()) is not None: + if result := await spider(): return result - if result is None: + if result is None: # 所有爬虫都爬取失败 logger.error("每日素材刷新失败，请稍后重试") class Ambr(Spider): + """爬取 Ambr""" + farming_url = "https://api.ambr.top/v2/chs/dailyDungeon" async def _request(self, url: str) -> dict | None: + """对指定 url 发起 get 请求，并返回 json 中的 data 字段数据，可自动重试""" response = None for attempts in range(RETRY_TIMES): try: @@ -80,16 +84,17 @@ class Ambr(Spider): if response is not None: return response.json()["data"] - async def _parse_item_data(self, material_json_data: dict): + async def _parse_item_data(self, material_json_data: dict) -> list[AvatarData | WeaponData]: + """解析角色或武器数据""" items = [] for key in ["avatar", "weapon"]: cls = AvatarData if key == "avatar" else WeaponData if chunk_data := material_json_data["additions"]["requiredBy"].get(key): - for data in filter(lambda x: x["rank"] > 3, chunk_data): + for data in filter(lambda x: x["rank"] > 3, chunk_data): # 跳过 3 星及其以下的武器 item_id = data["id"] if isinstance(item_id, str): - continue # 旅行者 + continue # 跳过旅行者，旅行者的 id 类型为 str item_icon = await getattr(self.assets, key)(item_id).icon() items.append( cls( @@ -102,6 +107,7 @@ class Ambr(Spider): return items async def _parse_weekday_data(self, weekday_data: dict) -> list[AreaData]: + """解析某一天的数据""" area_data_list = [] for domain in weekday_data.values(): area_name = AREAS[int(domain["city"]) - 1] @@ -127,6 +133,7 @@ class Ambr(Spider): return area_data_list async def __call__(self) -> dict[int, FarmingData]: + """从 Ambr 上爬取每日素材表的具体实现方法""" week_map = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] if (full_farming_json_data := await self._request(self.farming_url)) is None: logger.error("从 Ambr 上爬取每日素材失败，请稍后重试") diff --git a/plugins/genshin/farming/plugin.py b/plugins/genshin/farming/plugin.py index 780f9a1c..571a29e2 100644 --- a/plugins/genshin/farming/plugin.py +++ b/plugins/genshin/farming/plugin.py @@ -15,7 +15,7 @@ from core.dependence.assets import AssetsService from gram_core.plugin import Plugin, handler from gram_core.services.template.models import FileType, RenderGroupResult from gram_core.services.template.services import TemplateService -from plugins.genshin.farming._const import AREAS +from plugins.genshin.farming._const import AREAS, WEEK_MAP from plugins.genshin.farming._model import AreaData, AvatarData, FullFarmingData, RenderData, UserOwned, WeaponData from plugins.genshin.farming._spider import Spider from plugins.tools.genshin import CharacterDetails, CookiesNotFoundError, GenshinHelper, PlayerNotFoundError @@ -24,19 +24,18 @@ from utils.log import logger from utils.uid import mask_number if TYPE_CHECKING: + from simnet.models.genshin.chronicle.characters import Character from telegram import Message, Update from telegram.ext import ContextTypes R = TypeVar("R") P = ParamSpec("P") -_RETRY_TIMES = 5 -_WEEK_MAP = ["一", "二", "三", "四", "五", "六", "日"] - DATA_FILE_PATH = DATA_DIR.joinpath("daily_material.json").resolve() def sort_item(item: AvatarData | WeaponData) -> tuple: + """对角色/武器进行排序： 拥有 > 星级 > 等级 > 命座/精炼""" rarity = item.rarity level = item.level if isinstance(item, AvatarData): @@ -66,6 +65,7 @@ class DailyFarming(Plugin): self.character_details = character_details async def _refresh_farming_data(self) -> bool: + """刷新并保存每日素材表""" async with self.lock: if (result := await Spider.execute(self.assets_service)) is not None: self.full_farming_data.__root__ = result @@ -109,9 +109,10 @@ class DailyFarming(Plugin): await aiofiles.os.remove(DATA_FILE_PATH) await refresh_task() - async def _get_character_skill(self, client, character) -> list[int]: - if getattr(client, "damaged", False): - return [] + async def _get_character_skill(self, client: "GenshinClient", character: "Character") -> list[int]: + """获取角色的天赋等级""" + if getattr(client, "damaged", False): # 检查 client 是否损坏 + return [] # 若损坏则直接退出 try: detail = await self.character_details.get_character_details(client, character) return [t.level for t in detail.talents if t.type in ["attack", "skill", "burst"]] @@ -132,7 +133,7 @@ class DailyFarming(Plugin): logger.debug("获取账号数据成功: UID=%s", client.player_id) characters = await client.get_genshin_characters(client.player_id) - for character in filter(lambda x: x.name != "旅行者", characters): + for character in filter(lambda x: x.name != "旅行者", characters): # 跳过旅行者 character_id = str(character.id) character_assets = self.assets_service.avatar(character_id) character_icon = await character_assets.icon(False) @@ -179,29 +180,29 @@ class DailyFarming(Plugin): # 有上述异常的， client 会返回 None return None, user_data - async def _get_render_data(self, user_id, weekday, title, time_text): - # 尝试获取用户已绑定的原神账号信息 - client, user_owned = await self._get_user_items(user_id) + async def _get_render_data(self, user_id: int, weekday: int, title: str, time_text: str) -> RenderData: + """获取模板渲染所需的数据""" + client, user_owned = await self._get_user_items(user_id) # 尝试获取用户已绑定的原神账号信息 today_farming = self.full_farming_data.weekday(weekday) area_avatars: list[AreaData] = [] area_weapons: list[AreaData] = [] for area_data in today_farming.areas: items = [] - new_area_data = deepcopy(area_data) - for avatar in area_data.avatars: + + for avatar in area_data.avatars: # 将原有角色信息替换为用户已拥有的角色信息 items.append(user_owned.avatars.get(str(avatar.id), avatar)) new_area_data.avatars = list(sorted(items, key=sort_item, reverse=True)) - for weapon in area_data.weapons: + for weapon in area_data.weapons: # 武器同上 if weapons := user_owned.weapons.get(str(weapon.id), []): items.extend(weapons) else: items.append(weapon) new_area_data.weapons = list(sorted(items, key=sort_item, reverse=True)) - [area_weapons, area_avatars][bool(area_data.avatars)].append(new_area_data) + [area_weapons, area_avatars][bool(area_data.avatars)].append(new_area_data) # 角色与武器分开存放 return RenderData( title=title, @@ -223,7 +224,7 @@ class DailyFarming(Plugin): weekday, title, time_text, full = _parse_time(args) - self.log_user(update, logger.info, "每日素材命令请求 || 参数 weekday=%s full=%s", _WEEK_MAP[weekday - 1], full) + self.log_user(update, logger.info, "每日素材命令请求 || 参数 weekday=%s full=%s", WEEK_MAP[weekday - 1], full) if weekday == 7: the_day = "今天" if title == "今日" else "这天" @@ -280,6 +281,7 @@ class DailyFarming(Plugin): @handler.command("refresh_farming_data", admin=True, block=False) async def refresh_farming_data(self, update: "Update", _): + """管理员用的、用于刷新每日素材表的指令""" user = update.effective_user message = update.effective_message @@ -290,7 +292,7 @@ class DailyFarming(Plugin): return notice = await message.reply_text("派蒙正在重新摘抄每日素材表，请稍等~", parse_mode=ParseMode.HTML) - async with self.lock: # 锁住第一把锁 + async with self.lock: await self._refresh_farming_data() await notice.edit_text( "每日素材表" @@ -300,17 +302,18 @@ class DailyFarming(Plugin): def _parse_time(args: list[str]) -> tuple[int, str, str, bool]: + """用于解析`daily_farming`指令的参数""" now = datetime.now() try: weekday = (_ := int(args[0])) - (_ > 0) weekday = (weekday % 7 + 7) % 7 - time_text = title = f"星期{_WEEK_MAP[weekday]}" + time_text = title = f"星期{WEEK_MAP[weekday]}" except (ValueError, IndexError): title = "今日" weekday = now.weekday() - (1 if now.hour < 4 else 0) weekday = 6 if weekday < 0 else weekday - time_text = f"星期{_WEEK_MAP[weekday]}" + time_text = f"星期{WEEK_MAP[weekday]}" full = bool(args and args[-1] == "full")