RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
399 lines
17 KiB
Python
399 lines
17 KiB
Python
from urllib.parse import unquote, urlsplit
|
||
# tg_channel.py — Telegram 通道适配（utility / chat 两个入口共用）
|
||
# 职责: 把统一 content（str 或 OneBot 段数组）转成 TG API 调用；
|
||
# TG id 命名空间换算（业务状态按通道隔离）。
|
||
# 注意: 本模块是平台适配层，不含任何业务逻辑。
|
||
import asyncio
|
||
import base64
|
||
import contextlib
|
||
import os
|
||
import re
|
||
import urllib.request
|
||
import tempfile
|
||
|
||
from aiogram.types import FSInputFile
|
||
|
||
from channel import Channel, admin_list
|
||
|
||
# ---------------- TG id 命名空间 ----------------
|
||
# 业务状态（聊天历史/邮箱/live关注/游戏）以"裸 id"做 key。
|
||
# QQ 侧沿用原始 QQ 号；TG 侧加偏移，避免两端 id 撞号串状态。
|
||
TG_USER_OFFSET = 10 ** 10 # TG 私聊用户: ns = raw + offset
|
||
TG_GROUP_OFFSET = 10 ** 12 # TG 群: ns = offset + abs(raw_chat_id)
|
||
|
||
TG_CHUNK = 4000 # TG 单条上限 4096，留余量
|
||
|
||
|
||
class TGChannel(Channel):
|
||
name = "tg"
|
||
|
||
def __init__(self, bot, tg_username: str = "", admin_ids=None):
|
||
self.bot = bot
|
||
self.tg_username = tg_username
|
||
if admin_ids is not None:
|
||
self._raw_admin = set(int(x) for x in admin_ids)
|
||
else:
|
||
self._raw_admin = set(admin_list("tg"))
|
||
|
||
# ---- id 换算 ----
|
||
@staticmethod
|
||
def ns_user(raw_uid: int) -> int:
|
||
return TG_USER_OFFSET + int(raw_uid)
|
||
|
||
@staticmethod
|
||
def ns_group(raw_chat_id: int) -> int:
|
||
return TG_GROUP_OFFSET + abs(int(raw_chat_id))
|
||
|
||
def _to_raw_target(self, target_id) -> int:
|
||
tid = int(target_id)
|
||
if tid >= TG_GROUP_OFFSET:
|
||
return -(tid - TG_GROUP_OFFSET)
|
||
if tid >= TG_USER_OFFSET:
|
||
return tid - TG_USER_OFFSET
|
||
return tid # 原样（监控循环直接传裸 id 的场景）
|
||
|
||
def typing(self, target_id):
|
||
"""TG「正在输入」: sendChatAction("typing")，服务端状态约 5s 过期，
|
||
上下文存续期间每 4s 续一次。"""
|
||
chat_id = self._to_raw_target(target_id)
|
||
bot = self.bot
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def _typing():
|
||
stop = asyncio.Event()
|
||
|
||
async def beat():
|
||
while True:
|
||
try:
|
||
await bot.send_chat_action(chat_id=chat_id, action="typing")
|
||
except Exception as e:
|
||
print(f"[tg] typing 发送失败: {e}")
|
||
try:
|
||
await asyncio.wait_for(stop.wait(), timeout=4)
|
||
return
|
||
except asyncio.TimeoutError:
|
||
continue
|
||
|
||
task = asyncio.create_task(beat())
|
||
try:
|
||
yield
|
||
finally:
|
||
stop.set()
|
||
await asyncio.gather(task, return_exceptions=True)
|
||
return _typing()
|
||
|
||
def is_admin(self, user_id) -> bool:
|
||
try:
|
||
uid = int(user_id)
|
||
except (TypeError, ValueError):
|
||
return False
|
||
if uid >= TG_USER_OFFSET:
|
||
uid -= TG_USER_OFFSET
|
||
return uid in self._raw_admin or super().is_admin(user_id)
|
||
|
||
# ---- 发送 ----
|
||
@staticmethod
|
||
def _translate_oss_path(file_uri: str) -> str:
|
||
"""把远端 OSS 挂载或旧盘符 URI 映射到本机 OSS；普通本地 URI 保持原路径。"""
|
||
uri = urlsplit(file_uri.replace("\\", "/"))
|
||
if uri.scheme != "file" or uri.query or uri.fragment:
|
||
raise ValueError("需要有效的 file URI，文件名中的特殊字符须进行 URL 编码")
|
||
if re.fullmatch(r"[A-Za-z]:", uri.netloc):
|
||
p = uri.netloc + unquote(uri.path)
|
||
elif uri.netloc.lower() in ("", "localhost"):
|
||
p = unquote(uri.path)
|
||
else:
|
||
raise ValueError("不支持远程文件主机")
|
||
p = p.replace("\\", "/")
|
||
if "\x00" in p:
|
||
raise ValueError("文件路径包含空字符")
|
||
if re.match(r"^/[A-Za-z]:/", p):
|
||
p = p[1:]
|
||
relative = None
|
||
if p == "/mnt/A" or p.startswith("/mnt/A/"):
|
||
relative = p[len("/mnt/A"):].lstrip("/")
|
||
elif p.lower() == "c:/oss" or p.lower().startswith("c:/oss/"):
|
||
relative = p[len("c:/oss"):].lstrip("/")
|
||
elif re.match(r"^[A-Za-z]:", p):
|
||
raise ValueError("旧盘符路径不属于 OSS 挂载")
|
||
if relative is not None:
|
||
root = os.environ.get("OSS_MOUNT_PATH", "/mnt/oss")
|
||
if not os.path.isabs(root):
|
||
raise ValueError("OSS_MOUNT_PATH 必须为绝对路径")
|
||
root = os.path.realpath(root)
|
||
local = os.path.realpath(os.path.join(root, relative))
|
||
if os.path.commonpath((root, local)) != root:
|
||
raise ValueError("OSS 文件路径越出了挂载目录")
|
||
return local
|
||
if not os.path.isabs(p):
|
||
raise ValueError("本地文件路径必须为绝对路径")
|
||
return os.path.normpath(p)
|
||
|
||
async def _send_text(self, chat_id, text: str):
|
||
for i in range(0, len(text), TG_CHUNK):
|
||
chunk = text[i:i + TG_CHUNK]
|
||
await _retry_send(
|
||
lambda c=chunk: self.bot.send_message(chat_id=chat_id, text=c,
|
||
parse_mode=None,
|
||
disable_web_page_preview=True),
|
||
"发文本")
|
||
|
||
|
||
async def _send_file(self, chat_id, file: str, manual: str = None):
|
||
"""file 段：file:// 或本地路径发 document；manual 为失败提示里的手动访问链接。"""
|
||
if not file:
|
||
return False
|
||
local = self._translate_oss_path(file) if file.startswith("file://") else file
|
||
if not os.path.exists(local):
|
||
print(f"[tg] 文件本地不存在: {local}", flush=True)
|
||
await _send_notice(self.bot, chat_id,
|
||
f"文件发送失败（本地文件缺失）。请手动访问 {manual or TG_MANUAL_URL}")
|
||
return False
|
||
result = await _send_file_robust(self.bot, chat_id, local, manual)
|
||
return result is not None
|
||
|
||
async def _send_sticker(self, chat_id, file: str):
|
||
"""Send Agent sticker segments through Telegram's sticker API."""
|
||
if not file:
|
||
return False
|
||
local = self._translate_oss_path(file) if file.startswith("file://") else file
|
||
if not os.path.exists(local):
|
||
print(f"[tg] 贴纸本地不存在: {local}", flush=True)
|
||
return False
|
||
result = await _send_upload_with_policy(
|
||
self.bot, chat_id,
|
||
lambda: self.bot.send_sticker(
|
||
chat_id=chat_id, sticker=FSInputFile(local),
|
||
request_timeout=TG_UPLOAD_TIMEOUT,
|
||
),
|
||
"发贴纸", _manual_access_url(local),
|
||
)
|
||
return result is not None
|
||
|
||
async def _send_image(self, chat_id, file: str):
|
||
if file.startswith("file://"):
|
||
local = self._translate_oss_path(file)
|
||
if os.path.exists(local):
|
||
result = await _send_image_robust(self.bot, chat_id, local)
|
||
return result is not None
|
||
print(f"[tg] OSS 图片本地不存在: {local}")
|
||
return False
|
||
if file.startswith("http://") or file.startswith("https://"):
|
||
manual = _manual_access_url(file)
|
||
size = await asyncio.to_thread(_probe_url_size, file)
|
||
if size > TG_DOC_LIMIT:
|
||
print(f"[tg] URL 文件过大({size} 字节)，跳过发送", flush=True)
|
||
await _send_notice(self.bot, chat_id, f"文件过大。请手动访问 {manual}")
|
||
return False
|
||
if size > TG_PHOTO_LIMIT:
|
||
result = await _send_upload_with_policy(
|
||
self.bot, chat_id,
|
||
lambda: self.bot.send_document(chat_id=chat_id, document=file,
|
||
request_timeout=TG_UPLOAD_TIMEOUT),
|
||
"按URL发文件", manual)
|
||
return result is not None
|
||
result = await _send_upload_with_policy(
|
||
self.bot, chat_id,
|
||
lambda: self.bot.send_photo(chat_id=chat_id, photo=file,
|
||
request_timeout=TG_UPLOAD_TIMEOUT),
|
||
"按URL发图片", manual)
|
||
return result is not None
|
||
# base64（可能带 data: 前缀）
|
||
b64 = file.split(",", 1)[-1] if file.startswith("data:") else file
|
||
data = base64.b64decode(b64)
|
||
suffix = ".jpg"
|
||
if file.startswith("data:"):
|
||
mime = file.split(";", 1)[0].split(":", 1)[-1]
|
||
suffix = {"image/png": ".png", "image/gif": ".gif", "image/webp": ".webp"}.get(mime, ".jpg")
|
||
fd, path = tempfile.mkstemp(suffix=suffix)
|
||
try:
|
||
with os.fdopen(fd, "wb") as f:
|
||
f.write(data)
|
||
result = await _send_image_robust(self.bot, chat_id, path)
|
||
return result is not None
|
||
finally:
|
||
try:
|
||
os.unlink(path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
async def send(self, msg_type, target_id, content):
|
||
chat_id = self._to_raw_target(target_id)
|
||
if isinstance(content, str):
|
||
if not content:
|
||
return True
|
||
await self._send_text(chat_id, content)
|
||
return True
|
||
# 段数组（OneBot 段格式）
|
||
text_buf = []
|
||
all_ok = True
|
||
for seg in content:
|
||
stype = seg.get("type")
|
||
sdata = seg.get("data", {})
|
||
if stype == "text":
|
||
text_buf.append(sdata.get("text", ""))
|
||
elif stype == "image":
|
||
if text_buf:
|
||
await self._send_text(chat_id, "".join(text_buf))
|
||
text_buf = []
|
||
if sdata.get("sub_type") == 1:
|
||
ok = await self._send_sticker(chat_id, sdata.get("file", ""))
|
||
else:
|
||
ok = await self._send_image(chat_id, sdata.get("file", ""))
|
||
all_ok = ok and all_ok
|
||
elif stype == "file":
|
||
if text_buf:
|
||
await self._send_text(chat_id, "".join(text_buf))
|
||
text_buf = []
|
||
ok = await self._send_file(chat_id, sdata.get("file", ""),
|
||
sdata.get("manual"))
|
||
all_ok = ok and all_ok
|
||
if text_buf:
|
||
await self._send_text(chat_id, "".join(text_buf))
|
||
return all_ok
|
||
|
||
|
||
TG_SEND_RETRIES = 3 # 发送失败重试次数（仅网络类瞬时错误）
|
||
|
||
|
||
def _is_retryable(e: Exception) -> bool:
|
||
"""只有瞬时错误才重试；BadRequest/Forbidden 等永久错误直接抛出"""
|
||
name = type(e).__name__
|
||
for key in ("Network", "Timeout", "Server", "Disconnected",
|
||
"Connector", "ClientError", "RetryAfter"):
|
||
if key in name:
|
||
return True
|
||
return False
|
||
|
||
|
||
TG_IMAGE_RETRIES = 5 # 图片/文件发送重试次数（仅 TG 侧；QQ 侧发送政策见 qq_channel/napcat 链路）
|
||
TG_UPLOAD_TIMEOUT = 300 # 上传类请求总超时（官方默认 60s 对大图太紧）
|
||
TG_PHOTO_LIMIT = 10 * 1024 * 1024 # send_photo 官方上限 10MB
|
||
TG_DOC_LIMIT = 50 * 1024 * 1024 # send_document 官方上限 50MB
|
||
TG_MANUAL_URL = "https://openapi.rrrosmontis.icu/"
|
||
|
||
|
||
def _manual_access_url(source) -> str:
|
||
"""从图片来源推导精确的手动访问链接（本地文件名/URL 中的数字 pid）。"""
|
||
try:
|
||
if isinstance(source, str) and source.startswith("http"):
|
||
m = re.search(r"[?&]id=(\d+)", source)
|
||
if m:
|
||
return f"{TG_MANUAL_URL}pixiv/get_image?id={m.group(1)}"
|
||
else:
|
||
stem = os.path.splitext(os.path.basename(str(source)))[0]
|
||
if stem.isdigit():
|
||
return f"{TG_MANUAL_URL}pixiv/get_image?id={stem}"
|
||
except Exception:
|
||
pass
|
||
return TG_MANUAL_URL
|
||
|
||
|
||
def _probe_url_size(url: str) -> int:
|
||
"""HEAD 后端镜像拿 Content-Length；失败返回 0（未知）。"""
|
||
try:
|
||
req = urllib.request.Request(url, method="HEAD",
|
||
headers={"User-Agent": "Mozilla/5.0"})
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
return int(resp.headers.get("Content-Length") or 0)
|
||
except Exception as e:
|
||
print(f"[tg] HEAD 探测大小失败: {type(e).__name__}: {str(e)[:80]}", flush=True)
|
||
return 0
|
||
|
||
|
||
def _looks_timeout(e: Exception) -> bool:
|
||
"""超时类错误识别（aiohttp/aiogram 的 timeout 命名不统一，按名字+文本兜）"""
|
||
text = type(e).__name__ + " " + str(e)
|
||
return "imeout" in text
|
||
|
||
|
||
async def _send_notice(bot, chat_id, text: str):
|
||
try:
|
||
await _retry_send(lambda: bot.send_message(chat_id=chat_id, text=text, parse_mode=None),
|
||
"发送失败提示")
|
||
except Exception as ex:
|
||
print(f"[tg] 失败提示也发不出: {type(ex).__name__}: {str(ex)[:80]}", flush=True)
|
||
|
||
|
||
async def _send_upload_with_policy(bot, chat_id, factory, desc: str, manual_url: str = None):
|
||
"""统一上传策略：5 次重试、300s 超时；放弃后按错误类型给用户一句可操作的提示。"""
|
||
last = None
|
||
timeout_seen = False
|
||
for attempt in range(TG_IMAGE_RETRIES):
|
||
try:
|
||
return await factory()
|
||
except Exception as e:
|
||
last = e
|
||
if _looks_timeout(e):
|
||
timeout_seen = True
|
||
if not _is_retryable(e):
|
||
break
|
||
if attempt < TG_IMAGE_RETRIES - 1:
|
||
delay = min(2 ** attempt * 2, 15)
|
||
print(f"[tg] {desc}失败(第{attempt + 1}/{TG_IMAGE_RETRIES}次): "
|
||
f"{type(e).__name__}: {str(e)[:100]}，{delay}s 后重试", flush=True)
|
||
await asyncio.sleep(delay)
|
||
reason = "上传超时" if (timeout_seen or _looks_timeout(last)) else "多次尝试发送失败"
|
||
notice = f"{reason}。请手动访问 {manual_url or TG_MANUAL_URL}"
|
||
print(f"[tg] {desc}放弃: {type(last).__name__}: {str(last)[:80]}；已通知用户", flush=True)
|
||
await _send_notice(bot, chat_id, notice)
|
||
return None
|
||
|
||
|
||
async def _send_image_robust(bot, chat_id, local: str):
|
||
"""本地图片按大小分流：>10MB 直接走文件通道，>50MB 不发送只提示。"""
|
||
try:
|
||
size = os.path.getsize(local)
|
||
except OSError:
|
||
size = 0
|
||
manual = _manual_access_url(local)
|
||
if size > TG_DOC_LIMIT:
|
||
print(f"[tg] 文件过大({size} 字节)，跳过发送", flush=True)
|
||
await _send_notice(bot, chat_id, f"文件过大。请手动访问 {manual}")
|
||
return None
|
||
if size > TG_PHOTO_LIMIT:
|
||
return await _send_upload_with_policy(
|
||
bot, chat_id,
|
||
lambda: bot.send_document(chat_id=chat_id, document=FSInputFile(local),
|
||
request_timeout=TG_UPLOAD_TIMEOUT),
|
||
"发文件", manual)
|
||
return await _send_upload_with_policy(
|
||
bot, chat_id,
|
||
lambda: bot.send_photo(chat_id=chat_id, photo=FSInputFile(local),
|
||
request_timeout=TG_UPLOAD_TIMEOUT),
|
||
"发图片", manual)
|
||
|
||
|
||
async def _send_file_robust(bot, chat_id, local: str, manual_url: str = None):
|
||
"""发送任意文件（压缩包等）：>50MB 拒发，其余一律走 send_document。"""
|
||
try:
|
||
size = os.path.getsize(local)
|
||
except OSError:
|
||
size = 0
|
||
if size > TG_DOC_LIMIT:
|
||
print(f"[tg] 文件过大({size} 字节)，跳过发送", flush=True)
|
||
await _send_notice(bot, chat_id, f"文件过大。请手动访问 {manual_url or TG_MANUAL_URL}")
|
||
return None
|
||
return await _send_upload_with_policy(
|
||
bot, chat_id,
|
||
lambda: bot.send_document(chat_id=chat_id, document=FSInputFile(local),
|
||
request_timeout=TG_UPLOAD_TIMEOUT),
|
||
"发文件", manual_url or TG_MANUAL_URL)
|
||
|
||
|
||
async def _retry_send(factory, desc: str):
|
||
last = None
|
||
for attempt in range(TG_SEND_RETRIES):
|
||
try:
|
||
return await factory()
|
||
except Exception as e:
|
||
last = e
|
||
if not _is_retryable(e):
|
||
raise
|
||
if attempt < TG_SEND_RETRIES - 1:
|
||
print(f"[tg] {desc}失败(第{attempt + 1}/{TG_SEND_RETRIES}次): "
|
||
f"{type(e).__name__}: {str(e)[:120]}，{2 ** attempt}s 后重试")
|
||
await asyncio.sleep(2 ** attempt)
|
||
raise last
|