import re import asyncio from core.builtins import Bot from core.component import module from core.utils.http import get_url from config import Config mod_dl = module( bind_prefix='mod_dl', desc='下载 CurseForge 和 Mdorinth 上的 Mod。', developers=['HornCopper', 'OasisAkari', 'z0z0r4'], recommend_modules=['mcmod'], alias='moddl') x_api_key = Config("curseforge_api_key") def cn_chk(string: str): for word in string: if u'\u4e00' <= word <= u'\u9fff': return True return False @mod_dl.handle(' [] {通过模组名获取模组下载链接}') async def main(msg: Bot.MessageSession): mod_name = msg.parsed_msg[''] ver = msg.parsed_msg.get('', None) if ver: match_ver = re.match(r'^\d+\.\d+\.\d+$|^\d+\.\d+$|\d+w\d+[abcd]', ver) if match_ver is None: mod_name += ' ' + ver ver = False if cn_chk(mod_name): return {'msg': '暂不支持中文搜索。', 'success': False} async def search_modrinth(name: str, ver: str): url = f'https://api.modrinth.com/v2/search?query={name}&limit=10' if ver: url += f'&facets=[["versions:{ver}"],["project_type:mod"]]' else: url += f'&facets=[["project_type:mod"]]' resp = await get_url(url, 200, fmt="json", timeout=5, attempt=3) if resp is not None: results = [] if len(resp["hits"]) == 0: return None for project in resp['hits']: results.append(("modrinth", project["title"], project["project_id"], project["versions"])) return results async def search_curseforge(name: str, ver: str): headers = { 'Accept': 'application/json', 'x-api-key': x_api_key } url = f'https://api.curseforge.com/v1/mods/search?gameId=432&searchFilter={name}&sortField=2&sortOrder=desc&pageSize=10&classId=6' if ver: url += f'&gameVersion={ver}' resp = await get_url(url, 200, fmt="json", timeout=5, attempt=3, headers=headers) if resp is not None: if resp["pagination"]["resultCount"] == 0: return None results = [] for mod in resp["data"]: results.append(("curseforge", mod["name"], mod["id"], None)) return results async def get_modrinth_project_version(project_id: str, ver: str): url = f'https://api.modrinth.com/v2/project/{project_id}/version?game_versions=["{ver}"]&featured=true' resp = (await get_url(url, 200, fmt="json", timeout=5, attempt=3))[0] if resp is not None: return resp async def get_curseforge_mod_version_index(modid: str): headers = { 'Accept': 'application/json', 'x-api-key': x_api_key } url = f'https://api.curseforge.com/v1/mods/{modid}' resp = await get_url(url, 200, fmt="json", timeout=5, attempt=3, headers=headers) if resp is not None: return resp["data"]['latestFilesIndexes'] async def get_curseforge_mod_file(modid: str, ver: str): headers = { 'Accept': 'application/json', 'x-api-key': x_api_key } url = f'https://api.curseforge.com/v1/mods/{modid}/files?gameVersion={ver}' resp = await get_url(url, 200, fmt="json", timeout=5, attempt=3, headers=headers) if resp is not None: return resp["data"][0] # 搜索 Mod result = await asyncio.gather(*(search_modrinth(mod_name, ver), search_curseforge(mod_name, ver))) cache_result = [] if result[0] is None and result[1] is None: await msg.finish("搜索失败，无结果。") else: # 合并搜索结果 reply_text, count = [], 0 if result[0] is None: reply_text.append("Modrinth 结果：无") reply_text.append("Modrinth 结果：") for mod in result[0]: count += 1 reply_text.append(f"{count}. {mod[1]}") cache_result.append(mod) if result[1] is None: reply_text.append("CurseForge 结果：无") else: reply_text.append("CurseForge 结果：") for mod in result[1]: count += 1 reply_text.append(f"{count}. {mod[1]}") cache_result.append(mod) reply = await msg.waitReply('\n'.join(reply_text) + '\n请回复编号来选择mod。') replied = reply.asDisplay(text_only=True) # 查找 Mod if replied.isdigit(): replied = int(replied) if replied > len(cache_result): return await msg.finish('编号超出范围。') else: mod_info = cache_result[replied - 1] else: return await msg.finish('无效的编号，必须为纯数字。') if mod_info[0] == "modrinth": # modrinth mod if ver is None: reply2 = await msg.waitReply('此mod拥有如下版本：\n' + '\n'.join(mod_info[3]) + '\n请回复版本号来选择版本。') replied2 = reply2.asDisplay(text_only=True) if replied2 in mod_info[3]: version_info = await get_modrinth_project_version(mod_info[2], replied2) if version_info is not None: await msg.finish(f'{" ".join(version_info["loaders"])}\n下载链接：{version_info["files"][0]["url"]}\n文件名：{version_info["files"][0]["filename"]}') elif ver not in mod_info[3]: await msg.finish("未找到指定版本。") elif ver in mod_info[3]: version_info = await get_modrinth_project_version(mod_info[2], ver) if version_info is not None: await msg.finish(f'{" ".join(version_info["loaders"])}\n下载链接：{version_info["files"][0]["url"]}\n文件名：{version_info["files"][0]["filename"]}') else: # curseforge mod version_index = await get_curseforge_mod_version_index(mod_info[2]) if version_index is not None: if ver is None: reply_text = [] for version in version_index: reply_text.append(version["gameVersion"]) reply2 = await msg.waitReply('此mod拥有如下版本：\n' + '\n'.join(reply_text) + '\n请回复版本号来选择版本。') ver = reply2.asDisplay(text_only=True) elif ver not in version_index: await msg.finish("未找到指定版本。") if ver in version_index: file_info = await get_curseforge_mod_file(mod_info[2], ver) if file_info is not None: await msg.finish(f'{" ".join(file_info["gameVersions"])}\n下载链接：{file_info["downloadUrl"]}\n文件名：{file_info["fileName"]}') else: await msg.finish("未找到指定版本。")