codfrm/MoviePilot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add torrents message

Browse Source
This commit is contained in:
jxxghp 2023-06-06 12:53:48 +08:00
parent 05564c8ffe
commit 9047e0252a
4 changed files with 64 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
app/modules/telegram/__init__.py
View File

@ -102,10 +102,10 @@ class TelegramModule(_ModuleBase):
def post_torrents_message(self, title: str, items: List[TorrentInfo],
userid: Union[str, int] = None) -> Optional[bool]:
"""
TODO 发送种子信息选择列表
发送种子信息选择列表
:param title: 标题
:param items: 消息列表
:param userid: 用户ID
:return: 成功或失败
"""
pass
return self.telegram.send_torrents_msg(title=title, torrents=items, userid=userid)

31
app/modules/telegram/telegram.py
View File

@ -2,7 +2,7 @@ from threading import Event, Thread
from typing import Optional, List
from urllib.parse import urlencode
from app.core import settings, MediaInfo
from app.core import settings, MediaInfo, TorrentInfo
from app.log import logger
from app.utils.http import RequestUtils
from app.utils.singleton import Singleton
@ -111,6 +111,35 @@ class Telegram(metaclass=Singleton):
logger.error(f"发送消息失败：{msg_e}")
return False
def send_torrents_msg(self, torrents: List[TorrentInfo], userid: str = "", title: str = "") -> Optional[bool]:
"""
发送列表消息
"""
if not self._telegram_token or not self._telegram_chat_id:
return None
try:
index, caption = 1, "*%s*" % title
for torrent in torrents:
link = torrent.page_url
title = torrent.title
free = torrent.get_volume_factor_string()
seeder = f"{torrent.seeders}↑"
description = torrent.description
caption = f"{caption}\n{index}. [{title}]({link}) {free} {seeder}\n{description}"
index += 1
if userid:
chat_id = userid
else:
chat_id = self._telegram_chat_id
return self.__send_request(chat_id=chat_id, caption=caption)
except Exception as msg_e:
logger.error(f"发送消息失败：{msg_e}")
return False
def __send_request(self, chat_id="", image="", caption="") -> bool:
"""
向Telegram发送报文

4
app/modules/wechat/__init__.py
View File

@ -135,10 +135,10 @@ class WechatModule(_ModuleBase):
def post_torrents_message(self, title: str, items: List[TorrentInfo],
userid: Union[str, int] = None) -> Optional[bool]:
"""
TODO 发送种子信息选择列表
发送种子信息选择列表
:param title: 标题
:param items: 消息列表
:param userid: 用户ID
:return: 成功或失败
"""
pass
return self.wechat.send_torrents_msg(title=title, torrents=items, userid=userid)

35
app/modules/wechat/wechat.py
View File

@ -3,7 +3,7 @@ import threading
from datetime import datetime
from typing import Optional, List
from app.core import settings, MediaInfo
from app.core import settings, MediaInfo, TorrentInfo
from app.log import logger
from app.utils.http import RequestUtils
from app.utils.singleton import Singleton
@ -131,7 +131,7 @@ class WeChat(metaclass=Singleton):
}
return self.__post_request(message_url, req_json)
def send_msg(self, title: str, text: str = "", image: str = "", userid: str = None):
def send_msg(self, title: str, text: str = "", image: str = "", userid: str = None) -> Optional[bool]:
"""
微信消息发送入口，支持文本、图片、链接跳转、指定发送对象
:param title: 消息标题
@ -145,11 +145,11 @@ class WeChat(metaclass=Singleton):
return None
if image:
ret_code, ret_msg = self.__send_image_message(title, text, image, userid)
ret_code = self.__send_image_message(title, text, image, userid)
else:
ret_code, ret_msg = self.__send_message(title, text, userid)
ret_code = self.__send_message(title, text, userid)
return ret_code, ret_msg
return ret_code
def send_medias_msg(self, medias: List[MediaInfo], userid: str = "") -> Optional[bool]:
"""
@ -187,6 +187,31 @@ class WeChat(metaclass=Singleton):
}
return self.__post_request(message_url, req_json)
def send_torrents_msg(self, torrents: List[TorrentInfo], userid: str = "", title: str = "") -> Optional[bool]:
"""
发送列表消息
"""
if not self.__get_access_token():
logger.error("获取微信access_token失败，请检查参数配置")
return None
try:
index, caption = 1, "*%s*" % title
for torrent in torrents:
link = torrent.page_url
title = torrent.title
free = torrent.get_volume_factor_string()
seeder = f"{torrent.seeders}↑"
description = torrent.description
caption = f"{caption}\n{index}. [{title}]({link}) {free} {seeder}\n{description}"
index += 1
return self.__send_message(title, caption, userid)
except Exception as msg_e:
logger.error(f"发送消息失败：{msg_e}")
return False
def __post_request(self, message_url: str, req_json: dict) -> bool:
"""
向微信发送请求