Merge pull request #20 from lgc2333/fix-telegram-local

Feat: feat & fix for `image_fetch` function

src/nonebot_plugin_alconna/uniseg/tools.py

@@ -1,18 +1,20 @@
 import random
+from pathlib import Path
 from base64 import b64decode
 from typing import TYPE_CHECKING, List, Type, Union, Literal, Optional, overload
 
 from yarl import URL
 from nonebot import get_bots
 from nonebot.typing import T_State
 from nonebot import get_bot as _get_bot
+from nonebot.exception import ActionFailed
 from nonebot.internal.driver.model import Request
 from nonebot.internal.adapter import Bot, Event, Adapter
 
 from .segment import Image
 
 
-async def image_fetch(event: Event, bot: Bot, state: T_State, img: Image):
+async def image_fetch(event: Event, bot: Bot, state: T_State, img: Image, **kwargs):
     adapter_name = bot.adapter.get_name()
     if adapter_name == "RedProtocol":
         origin = img.origin
@@ -26,7 +28,7 @@ async def image_fetch(event: Event, bot: Bot, state: T_State, img: Image):
         return await origin.download(bot)
 
     if img.url:  # mirai2, qqguild, kook, villa, feishu, minecraft, ding
-        req = Request("GET", img.url)
+        req = Request("GET", img.url, **kwargs)
         resp = await bot.adapter.request(req)
         return resp.content
     if not img.id:
@@ -37,7 +39,7 @@ async def image_fetch(event: Event, bot: Bot, state: T_State, img: Image):
 
             assert isinstance(bot, Bot)
         url = (await bot.get_image(file=img.id))["data"]["url"]
-        req = Request("GET", url)
+        req = Request("GET", url, **kwargs)
         resp = await bot.adapter.request(req)
         return resp.content
     if adapter_name == "OneBot V12":
@@ -49,16 +51,21 @@ async def image_fetch(event: Event, bot: Bot, state: T_State, img: Image):
         return b64decode(resp) if isinstance(resp, str) else bytes(resp)
     if adapter_name == "mirai2":
         url = f"https://gchat.qpic.cn/gchatpic_new/0/0-0-" f"{img.id.replace('-', '').upper()}/0"
-        req = Request("GET", url)
+        req = Request("GET", url, **kwargs)
         resp = await bot.adapter.request(req)
         return resp.content
     if adapter_name == "Telegram":
         if TYPE_CHECKING:
             from nonebot.adapters.telegram.bot import Bot
 
             assert isinstance(bot, Bot)
-        url = URL(bot.bot_config.api_server) / "file" / f"bot{bot.bot_config.token}" / img.id
-        req = Request("GET", url)
+        res = await bot.get_file(file_id=img.id)
+        if not res.file_path:
+            raise ActionFailed("Telegram", "get file failed")
+        if (p := Path(res.file_path)).exists():  # telegram api local mode
+            return p.read_bytes()
+        url = URL(bot.bot_config.api_server) / "file" / f"bot{bot.bot_config.token}" / res.file_path
+        req = Request("GET", url, **kwargs)
         resp = await bot.adapter.request(req)
         return resp.content
     if adapter_name == "ntchat":