mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-03 22:04:57 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""
|
||
通过群晖 Chat Webhook 发送消息并查询机器人可见用户。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from threading import Lock
|
||
from typing import List, Optional, cast
|
||
from urllib.parse import quote
|
||
|
||
from app.adapters.network.http import RequestUtils
|
||
from app.domain.context import Context, MediaInfo
|
||
from app.domain.metainfo import MetaInfo
|
||
from app.foundation import size as size_tools
|
||
from app.foundation import url as url_tools
|
||
from app.runtime.log import logger
|
||
|
||
lock = Lock()
|
||
|
||
|
||
class SynologyChat:
|
||
"""
|
||
管理群晖 Chat 机器人凭据并发送文本和资源列表。
|
||
"""
|
||
|
||
def __init__(self, SYNOLOGYCHAT_WEBHOOK: Optional[str] = None, SYNOLOGYCHAT_TOKEN: Optional[str] = None, **kwargs: object) -> None:
|
||
"""
|
||
保存群晖机器人凭据，配置完整时创建 HTTP 客户端。
|
||
|
||
缺少任一凭据时保留禁用状态，不执行网络请求。
|
||
|
||
:param SYNOLOGYCHAT_WEBHOOK: 发送消息的 Webhook 地址
|
||
:param SYNOLOGYCHAT_TOKEN: 机器人认证令牌，供入站校验和用户查询共用
|
||
:param kwargs: 兼容通知配置中的其他字段，当前忽略
|
||
"""
|
||
self._webhook_url = ""
|
||
self._token: Optional[str] = None
|
||
self._domain: Optional[str] = None
|
||
if not SYNOLOGYCHAT_WEBHOOK or not SYNOLOGYCHAT_TOKEN:
|
||
logger.error("SynologyChat配置不完整！")
|
||
return
|
||
self._req = RequestUtils(content_type="application/x-www-form-urlencoded")
|
||
self._webhook_url = SYNOLOGYCHAT_WEBHOOK
|
||
self._token = SYNOLOGYCHAT_TOKEN
|
||
if self._webhook_url:
|
||
self._domain = url_tools.base_url(self._webhook_url)
|
||
|
||
def check_token(self, token: Optional[str]) -> bool:
|
||
"""
|
||
仅接受已配置机器人令牌，缺失配置和缺失入站令牌均不能通过认证。
|
||
"""
|
||
return self._token is not None and token == self._token
|
||
|
||
def get_state(self) -> bool:
|
||
"""
|
||
通过机器人用户查询检查服务是否可用。
|
||
|
||
:return: 配置完整且查询到至少一个未删除用户时返回 True
|
||
"""
|
||
if not self._webhook_url or not self._token:
|
||
return False
|
||
ret = self.__get_bot_users()
|
||
if ret:
|
||
return True
|
||
return False
|
||
|
||
def send_msg(self, title: Optional[str], text: Optional[str] = None, image: Optional[str] = None,
|
||
userid: Optional[str] = None, link: Optional[str] = None) -> Optional[bool]:
|
||
"""
|
||
向指定用户或机器人全部可见用户发送消息。
|
||
|
||
:param title: 消息标题，可与正文择一提供
|
||
:param text: 消息正文
|
||
:param image: 可选图片下载地址
|
||
:param userid: 指定数字用户标识；为空时查询机器人全部可见用户
|
||
:param link: 可选详情链接
|
||
:return: 发送成功返回 True；配置、内容、用户查询或发送失败时返回 False
|
||
"""
|
||
if not title and not text:
|
||
logger.error("标题和内容不能同时为空")
|
||
return False
|
||
if not self._webhook_url or not self._token:
|
||
return False
|
||
try:
|
||
# 拼装消息内容
|
||
titles = str(title).split('\n')
|
||
if len(titles) > 1:
|
||
title = titles[0]
|
||
if not text:
|
||
text = "\n".join(titles[1:])
|
||
else:
|
||
text = "%s\n%s" % ("\n".join(titles[1:]), text)
|
||
|
||
if text:
|
||
caption = "*%s*\n%s" % (title, text.replace("\n\n", "\n"))
|
||
else:
|
||
# 正文为空时，上方检查已保证标题存在。
|
||
caption = cast(str, title)
|
||
|
||
if link:
|
||
caption = f"{caption}\n[查看详情]({link})"
|
||
|
||
payload_data: dict[str, object] = {'text': quote(caption)}
|
||
if image:
|
||
payload_data['file_url'] = quote(image)
|
||
if userid:
|
||
payload_data['user_ids'] = [int(userid)]
|
||
else:
|
||
userids = self.__get_bot_users()
|
||
if not userids:
|
||
logger.error("SynologyChat机器人没有对任何用户可见")
|
||
return False
|
||
payload_data['user_ids'] = userids
|
||
|
||
return self.__send_request(payload_data)
|
||
|
||
except Exception as msg_e:
|
||
logger.error(f"SynologyChat发送消息错误：{str(msg_e)}")
|
||
return False
|
||
|
||
def send_medias_msg(self, medias: List[MediaInfo], userid: Optional[str] = None, title: Optional[str] = None) -> Optional[bool]:
|
||
"""
|
||
将媒体条目组成带详情链接的列表消息。
|
||
|
||
:param medias: 待展示媒体列表，不接受空列表
|
||
:param userid: 数字用户标识；为空时发送给机器人全部可见用户
|
||
:param title: 列表标题，必须提供非空值
|
||
:return: 发送成功返回 True；输入、配置或发送失败返回 False
|
||
"""
|
||
if not medias:
|
||
return False
|
||
if not self._webhook_url or not self._token:
|
||
return False
|
||
try:
|
||
if not title or not isinstance(medias, list):
|
||
return False
|
||
image: Optional[str]
|
||
index, image, caption = 1, "", "*%s*" % title
|
||
for media in medias:
|
||
if not image:
|
||
image = media.get_message_image()
|
||
if media.vote_average:
|
||
caption = "%s\n%s. <%s|%s>\n_%s，%s_" % (caption,
|
||
index,
|
||
media.detail_link,
|
||
media.title_year,
|
||
f"类型：{media.type.value if media.type else ''}",
|
||
f"评分：{media.vote_average}")
|
||
else:
|
||
caption = "%s\n%s. <%s|%s>\n_%s_" % (caption,
|
||
index,
|
||
media.detail_link,
|
||
media.title_year,
|
||
f"类型：{media.type.value if media.type else ''}")
|
||
index += 1
|
||
|
||
userids: List[object]
|
||
if userid:
|
||
userids = [int(userid)]
|
||
else:
|
||
userids = self.__get_bot_users()
|
||
payload_data: dict[str, object] = {
|
||
"text": quote(caption),
|
||
"user_ids": userids
|
||
}
|
||
return self.__send_request(payload_data)
|
||
|
||
except Exception as msg_e:
|
||
logger.error(f"SynologyChat发送消息错误：{str(msg_e)}")
|
||
return False
|
||
|
||
def send_torrents_msg(self, torrents: List[Context],
|
||
userid: Optional[str] = None, title: Optional[str] = None, link: Optional[str] = None) -> Optional[bool]:
|
||
"""
|
||
将种子上下文组成带站点和资源信息的列表消息。
|
||
|
||
:param torrents: 待展示种子列表，不接受空列表
|
||
:param userid: 数字用户标识；为空时发送给机器人全部可见用户
|
||
:param title: 列表标题
|
||
:param link: 兼容调用参数；当前循环会覆盖为最后一个种子的详情地址
|
||
:return: 发送成功返回 True；配置缺失返回 None；空列表或发送失败返回 False
|
||
"""
|
||
if not self._webhook_url or not self._token:
|
||
return None
|
||
|
||
if not torrents:
|
||
return False
|
||
|
||
try:
|
||
index, caption = 1, "*%s*" % title
|
||
for context in torrents:
|
||
torrent = context.torrent_info
|
||
if torrent is None:
|
||
continue
|
||
site_name = torrent.site_name or ""
|
||
meta = MetaInfo(torrent.title or "", torrent.description)
|
||
link = torrent.page_url
|
||
title = f"{meta.season_episode} " \
|
||
f"{meta.resource_term} " \
|
||
f"{meta.video_term} " \
|
||
f"{meta.release_group}"
|
||
title = re.sub(r"\s+", " ", title).strip()
|
||
free = torrent.volume_factor
|
||
seeder = f"{torrent.seeders}↑"
|
||
description = torrent.description
|
||
caption = f"{caption}\n{index}.【{site_name}】<{link}|{title}> " \
|
||
f"{size_tools.format_compact_size(torrent.size)} {free} {seeder}\n" \
|
||
f"_{description}_"
|
||
index += 1
|
||
|
||
if link:
|
||
caption = f"{caption}\n[查看详情]({link})"
|
||
|
||
userids: List[object]
|
||
if userid:
|
||
userids = [int(userid)]
|
||
else:
|
||
userids = self.__get_bot_users()
|
||
|
||
payload_data: dict[str, object] = {
|
||
"text": quote(caption),
|
||
"user_ids": userids
|
||
}
|
||
return self.__send_request(payload_data)
|
||
except Exception as msg_e:
|
||
logger.error(f"SynologyChat发送消息错误：{str(msg_e)}")
|
||
return False
|
||
|
||
def __get_bot_users(self) -> List[object]:
|
||
"""
|
||
查询机器人可见且未删除的用户标识。
|
||
|
||
:return: 用户标识列表；缺少配置或请求失败返回空列表
|
||
"""
|
||
if not self._domain or not self._token:
|
||
return []
|
||
req_url = f"{self._domain}" \
|
||
f"/webapi/entry.cgi?api=SYNO.Chat.External&method=user_list&version=2&token=" \
|
||
f"{self._token}"
|
||
ret = self._req.get_res(url=req_url)
|
||
if ret and ret.status_code == 200:
|
||
users = ret.json().get("data", {}).get("users", []) or []
|
||
return [user.get("user_id") for user in users if user.get("deleted", True) is False]
|
||
else:
|
||
return []
|
||
|
||
def __send_request(self, payload_data: dict[str, object]) -> bool:
|
||
"""
|
||
提交群晖消息载荷，同时检查 HTTP 和业务错误码。
|
||
|
||
:param payload_data: 已编码的消息正文、接收人和可选文件地址
|
||
:return: HTTP 200 且非空响应没有业务错误码时返回 True，否则返回 False
|
||
"""
|
||
payload = f"payload={json.dumps(payload_data)}"
|
||
ret = self._req.post_res(url=self._webhook_url, data=payload)
|
||
if ret and ret.status_code == 200:
|
||
result = ret.json()
|
||
if result:
|
||
errno = result.get('error', {}).get('code')
|
||
errmsg = result.get('error', {}).get('errors')
|
||
if not errno:
|
||
return True
|
||
logger.error(f"SynologyChat返回错误：{errno}-{errmsg}")
|
||
return False
|
||
else:
|
||
logger.error(f"SynologyChat返回：{ret.text}")
|
||
return False
|
||
elif ret is not None:
|
||
logger.error(f"SynologyChat请求失败，错误码：{ret.status_code}，错误原因：{ret.reason}")
|
||
return False
|
||
else:
|
||
logger.error("SynologyChat请求失败，未获取到返回信息")
|
||
return False
|