fix: handle Telegram urllib3 header formatter compatibility (#6074)

This commit is contained in:
qqcomeup
2026-07-07 19:58:49 +08:00
committed by GitHub
parent 09bb32f681
commit 0fbf05d72f
3 changed files with 157 additions and 16 deletions
+24
View File
@@ -0,0 +1,24 @@
def ensure_urllib3_header_param_compat() -> None:
"""
pyTelegramBotAPI imports urllib3.fields.format_header_param at import time.
Some urllib3-future builds only expose newer formatter names.
RFC 2231 formatting is kept as the last fallback because it encodes
non-ASCII values differently from urllib3's old default.
"""
try:
from urllib3 import fields
except ImportError:
return
if hasattr(fields, "format_header_param"):
return
for fallback_name in (
"format_header_param_html5",
"format_multipart_header_param",
"format_header_param_rfc2231",
):
fallback = getattr(fields, fallback_name, None)
if fallback is not None:
fields.format_header_param = fallback
return
+21 -16
View File
@@ -8,36 +8,41 @@ from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Union
from urllib.parse import urljoin, quote
from telebot import TeleBot, apihelper
from telebot.types import (
from app.modules.telegram.compat import ensure_urllib3_header_param_compat
# Must run before importing pyTelegramBotAPI.
ensure_urllib3_header_param_compat()
from telebot import TeleBot, apihelper # noqa: E402
from telebot.types import ( # noqa: E402
BotCommand,
InlineKeyboardMarkup,
InlineKeyboardButton,
InputMediaPhoto,
)
try:
from telebot.types import ForceReply
from telebot.types import ForceReply # noqa: E402
except ImportError:
ForceReply = None
from telegramify_markdown import standardize, telegramify # noqa
from telegramify_markdown import standardize, telegramify # noqa: E402
try:
from telegramify_markdown import entities_to_markdownv2 # noqa
from telegramify_markdown import entities_to_markdownv2 # noqa: E402
except ImportError:
entities_to_markdownv2 = None
try:
from telegramify_markdown.content import ContentTypes, File, Photo, Text
from telegramify_markdown.content import ContentTypes, File, Photo, Text # noqa: E402
except ImportError:
from telegramify_markdown.type import ContentTypes, File, Photo, Text
from telegramify_markdown.type import ContentTypes, File, Photo, Text # noqa: E402
from app.core.config import settings
from app.core.context import MediaInfo, Context
from app.core.metainfo import MetaInfo
from app.helper.image import ImageHelper
from app.helper.thread import ThreadHelper
from app.log import logger
from app.utils.common import retry
from app.utils.http import RequestUtils
from app.utils.string import StringUtils
from app.core.config import settings # noqa: E402
from app.core.context import MediaInfo, Context # noqa: E402
from app.core.metainfo import MetaInfo # noqa: E402
from app.helper.image import ImageHelper # noqa: E402
from app.helper.thread import ThreadHelper # noqa: E402
from app.log import logger # noqa: E402
from app.utils.common import retry # noqa: E402
from app.utils.http import RequestUtils # noqa: E402
from app.utils.string import StringUtils # noqa: E402
TELEGRAM_PARSE_MODE_MARKDOWN = "MarkdownV2"
+112
View File
@@ -0,0 +1,112 @@
import importlib
import sys
from types import SimpleNamespace
import pytest
def test_ensure_urllib3_header_param_compat_adds_best_available_alias(monkeypatch):
"""urllib3 future 缺少旧别名时,应补齐最接近的新入口。"""
from app.modules.telegram.compat import ensure_urllib3_header_param_compat
def multipart_formatter(name, value):
return f"multipart:{name}={value}"
def rfc2231_formatter(name, value):
return f"rfc2231:{name}={value}"
fake_fields = SimpleNamespace(
format_multipart_header_param=multipart_formatter,
format_header_param_rfc2231=rfc2231_formatter,
)
fake_urllib3 = SimpleNamespace(fields=fake_fields)
monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
ensure_urllib3_header_param_compat()
assert fake_fields.format_header_param is multipart_formatter
def test_ensure_urllib3_header_param_compat_prefers_html5_formatter(monkeypatch):
"""旧版本 urllib3 的 html5 formatter 存在时,应优先保持原别名语义。"""
from app.modules.telegram.compat import ensure_urllib3_header_param_compat
def html5_formatter(name, value):
return f"html5:{name}={value}"
def multipart_formatter(name, value):
return f"multipart:{name}={value}"
fake_fields = SimpleNamespace(
format_header_param_html5=html5_formatter,
format_multipart_header_param=multipart_formatter,
)
fake_urllib3 = SimpleNamespace(fields=fake_fields)
monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
ensure_urllib3_header_param_compat()
assert fake_fields.format_header_param is html5_formatter
def test_ensure_urllib3_header_param_compat_keeps_existing_formatter(monkeypatch):
"""已有 format_header_param 时不应覆盖,避免改变 urllib3 正常行为。"""
from app.modules.telegram.compat import ensure_urllib3_header_param_compat
def existing_formatter(name, value):
return f"existing:{name}={value}"
def rfc2231_formatter(name, value):
return f"fallback:{name}={value}"
fake_fields = SimpleNamespace(
format_header_param=existing_formatter,
format_header_param_rfc2231=rfc2231_formatter,
)
fake_urllib3 = SimpleNamespace(fields=fake_fields)
monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
ensure_urllib3_header_param_compat()
assert fake_fields.format_header_param is existing_formatter
def test_ensure_urllib3_header_param_compat_noops_without_fallback(monkeypatch):
"""没有任何可用 fallback 时保持 no-op,让原始导入错误暴露。"""
from app.modules.telegram.compat import ensure_urllib3_header_param_compat
fake_fields = SimpleNamespace()
fake_urllib3 = SimpleNamespace(fields=fake_fields)
monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
ensure_urllib3_header_param_compat()
assert not hasattr(fake_fields, "format_header_param")
def test_telegram_module_imports_when_urllib3_header_alias_missing(monkeypatch):
"""导入 Telegram 模块前旧别名缺失时,应先补齐再导入 telebot。"""
from urllib3 import fields
if not any(
hasattr(fields, name)
for name in (
"format_header_param_html5",
"format_multipart_header_param",
"format_header_param_rfc2231",
)
):
pytest.skip("urllib3 has no compatible header formatter fallback")
monkeypatch.delattr(fields, "format_header_param", raising=False)
for module_name in list(sys.modules):
if (
module_name == "app.modules.telegram.telegram"
or module_name.startswith("telebot")
):
monkeypatch.delitem(sys.modules, module_name, raising=False)
telegram_module = importlib.import_module("app.modules.telegram.telegram")
assert telegram_module.Telegram
assert hasattr(fields, "format_header_param")