akari-bot/modules/mod_dl/__init__.py

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('<mod_name> [<version>] {通过模组名获取模组下载链接}')
async def main(msg: Bot.MessageSession):
    mod_name = msg.parsed_msg['<mod_name>']
    ver = msg.parsed_msg.get('<version>', 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("未找到指定版本。")