mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 20:05:42 +08:00
e0642eafa7
* chore(types): clear AniList endpoint mypy debt * chore(types): clear Douban endpoint mypy debt * chore(types): clear recommendation endpoint mypy debt * chore(types): clear subscription notification mypy debt * chore(types): clear SDK string compatibility mypy debt * chore(types): ratchet v3 host mypy across remaining files * chore(types): reduce v3 host mypy debt and retire ledger * chore(types): reconcile host snapshot with upstream HTTP pool * test(types): cover host paths exposed by mypy cleanup
413 lines
12 KiB
Python
413 lines
12 KiB
Python
"""
|
||
渠道交互的通用状态存储、消息端口与按钮展示工具。
|
||
"""
|
||
|
||
import math
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
from threading import Lock
|
||
from typing import (
|
||
Any,
|
||
Dict,
|
||
Generic,
|
||
List,
|
||
Optional,
|
||
Protocol,
|
||
Sequence,
|
||
Tuple,
|
||
TypeVar,
|
||
Union,
|
||
)
|
||
|
||
from app.schemas.message import Message
|
||
from app.schemas.notification import ChannelCapabilityManager
|
||
from app.schemas.types import NotificationChannel
|
||
|
||
|
||
class _UserInteractionEntry(Protocol):
|
||
"""
|
||
声明按用户维护的交互状态所需的最小字段。
|
||
|
||
:ivar request_id: 当前交互请求标识，用于校验按钮是否仍属于活动会话
|
||
:ivar user_id: 所属用户标识，用于隔离不同用户的交互状态
|
||
:ivar created_at: 会话创建的本地时间，用于计算过期时间
|
||
"""
|
||
|
||
request_id: str
|
||
user_id: str
|
||
created_at: datetime
|
||
|
||
|
||
InteractionEntryT = TypeVar("InteractionEntryT", bound=_UserInteractionEntry)
|
||
|
||
|
||
class _ExpiringUserInteractionStore(Generic[InteractionEntryT]):
|
||
"""
|
||
提供单用户单会话的过期状态存储和双向索引。
|
||
"""
|
||
|
||
_ttl = timedelta(hours=24)
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
初始化按请求 ID、用户 ID 查询的线程安全状态表。
|
||
"""
|
||
self._by_id: Dict[str, InteractionEntryT] = {}
|
||
self._by_user: Dict[str, str] = {}
|
||
self._lock = Lock()
|
||
|
||
def _cleanup_locked(self) -> None:
|
||
"""
|
||
在持锁状态下移除超过统一保留时间的交互状态。
|
||
"""
|
||
expire_before = datetime.now() - self._ttl
|
||
expired = [
|
||
request_id
|
||
for request_id, request in self._by_id.items()
|
||
if request.created_at < expire_before
|
||
]
|
||
for request_id in expired:
|
||
request = self._by_id.pop(request_id, None)
|
||
if request:
|
||
self._by_user.pop(str(request.user_id), None)
|
||
|
||
def _replace_locked(self, request: InteractionEntryT) -> None:
|
||
"""
|
||
在持锁状态下替换用户旧状态并写入新的双向索引。
|
||
"""
|
||
user_key = str(request.user_id)
|
||
old_request_id = self._by_user.get(user_key)
|
||
if old_request_id:
|
||
self._by_id.pop(old_request_id, None)
|
||
self._by_id[request.request_id] = request
|
||
self._by_user[user_key] = request.request_id
|
||
|
||
def get_by_user(self, user_id: Optional[Union[str, int]]) -> Optional[InteractionEntryT]:
|
||
"""
|
||
按用户返回当前仍有效的交互状态。
|
||
"""
|
||
with self._lock:
|
||
self._cleanup_locked()
|
||
request_id = self._by_user.get(str(user_id))
|
||
if not request_id:
|
||
return None
|
||
return self._by_id.get(request_id)
|
||
|
||
def get_by_id(
|
||
self, request_id: str, user_id: Union[str, int]
|
||
) -> Optional[InteractionEntryT]:
|
||
"""
|
||
按请求 ID 查询状态，并校验请求属于当前用户。
|
||
"""
|
||
with self._lock:
|
||
self._cleanup_locked()
|
||
request = self._by_id.get(request_id)
|
||
if not request or str(request.user_id) != str(user_id):
|
||
return None
|
||
return request
|
||
|
||
def remove(self, request_id: str) -> None:
|
||
"""
|
||
删除指定交互状态及其用户索引。
|
||
"""
|
||
with self._lock:
|
||
request = self._by_id.pop(request_id, None)
|
||
if request:
|
||
self._by_user.pop(str(request.user_id), None)
|
||
|
||
def clear(self) -> None:
|
||
"""
|
||
清空全部交互状态，供生命周期结束和测试场景使用。
|
||
"""
|
||
with self._lock:
|
||
self._by_id.clear()
|
||
self._by_user.clear()
|
||
|
||
|
||
@dataclass
|
||
class PendingSlashInteraction:
|
||
"""
|
||
通用 slash 命令交互上下文。
|
||
|
||
:param request_id: 当前交互请求标识，用于校验按钮是否仍属于活动会话
|
||
:param user_id: 所属用户标识，用于隔离不同用户的交互状态
|
||
:param channel: 回复使用的通知渠道，None 表示未指定
|
||
:param source: 渠道实例标识，None 表示未指定实例
|
||
:param username: 用户显示名称，未知时为 None
|
||
:param command: 当前处理的斜杠命令名称
|
||
:param page: 当前列表页索引，从 0 开始
|
||
:param awaiting_input: 当前等待的文本输入用途，None 表示未等待补充输入
|
||
:param created_at: 会话创建的本地时间，用于计算过期时间
|
||
"""
|
||
|
||
request_id: str
|
||
user_id: str
|
||
channel: Optional[NotificationChannel]
|
||
source: Optional[str]
|
||
username: Optional[str]
|
||
command: str
|
||
page: int = 0
|
||
awaiting_input: Optional[str] = None
|
||
created_at: datetime = field(default_factory=datetime.now)
|
||
|
||
|
||
class SlashInteractionManager(
|
||
_ExpiringUserInteractionStore[PendingSlashInteraction]
|
||
):
|
||
"""
|
||
管理单个 slash 命令的交互会话。
|
||
"""
|
||
|
||
def create_or_replace(
|
||
self,
|
||
user_id: Optional[Union[str, int]],
|
||
command: str,
|
||
channel: Optional[NotificationChannel],
|
||
source: Optional[str],
|
||
username: Optional[str],
|
||
) -> PendingSlashInteraction:
|
||
"""
|
||
为用户创建 slash 会话并替换其上一条会话。
|
||
"""
|
||
with self._lock:
|
||
self._cleanup_locked()
|
||
user_key = str(user_id)
|
||
request = PendingSlashInteraction(
|
||
request_id=uuid.uuid4().hex[:12],
|
||
user_id=user_key,
|
||
command=command,
|
||
channel=channel,
|
||
source=source,
|
||
username=username,
|
||
)
|
||
self._replace_locked(request)
|
||
return request
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class InteractionContext:
|
||
"""
|
||
描述一次与渠道无关的用户交互上下文。
|
||
|
||
:param channel: 接收本次交互的通知渠道
|
||
:param source: 渠道实例标识，None 表示未指定实例
|
||
:param user_id: 所属用户标识，用于隔离不同用户的交互状态
|
||
:param username: 用户显示名称，未知时为 None
|
||
:param original_message_id: 渠道原始消息标识，未知时为 None
|
||
:param original_chat_id: 渠道原始聊天标识，未知时为 None
|
||
:param is_channel_admin: 渠道管理员身份判断；None 表示尚未判定
|
||
"""
|
||
|
||
channel: NotificationChannel
|
||
source: Optional[str]
|
||
user_id: Union[str, int]
|
||
username: Optional[str]
|
||
original_message_id: Optional[Union[str, int]] = None
|
||
original_chat_id: Optional[str] = None
|
||
is_channel_admin: Optional[bool] = None
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class InteractionDispatch:
|
||
"""
|
||
描述交互路由是否命中以及是否延迟结束处理状态。
|
||
|
||
:param handled: 是否已由本次路由消费交互
|
||
:param defer_processing_finish: 是否由后续异步流程结束渠道处理状态，默认立即结束
|
||
"""
|
||
|
||
handled: bool
|
||
defer_processing_finish: bool = False
|
||
|
||
|
||
class MessageGateway(Protocol):
|
||
"""
|
||
声明交互控制器使用的消息发送和编辑能力。
|
||
"""
|
||
|
||
def post_message(self, message: Message) -> None:
|
||
"""
|
||
发送一条交互消息。
|
||
|
||
:param message: 包含渠道、接收用户和展示内容的消息
|
||
"""
|
||
...
|
||
|
||
def edit_message(
|
||
self,
|
||
channel: NotificationChannel,
|
||
source: Optional[str],
|
||
message_id: Union[str, int],
|
||
chat_id: Union[str, int],
|
||
text: str,
|
||
title: Optional[str] = None,
|
||
buttons: Optional[List[List[Dict[str, Any]]]] = None,
|
||
metadata: Optional[Dict[str, Any]] = None,
|
||
) -> bool:
|
||
"""
|
||
编辑一条已发送的交互消息。
|
||
|
||
:param channel: 原消息所属通知渠道
|
||
:param source: 原消息所属渠道实例标识
|
||
:param message_id: 渠道分配的原消息标识
|
||
:param chat_id: 原消息所在的聊天标识
|
||
:param text: 替换后的消息正文
|
||
:param title: 替换后的标题，None 表示不提供独立标题
|
||
:param buttons: 按行组织的新按钮，None 表示不提供按钮
|
||
:param metadata: 渠道特定的附加编辑参数，None 表示无附加参数
|
||
:return: 渠道确认编辑成功时为 True；失败或不支持编辑时为 False
|
||
"""
|
||
...
|
||
|
||
|
||
def supports_interaction_buttons(channel: Optional[NotificationChannel]) -> bool:
|
||
"""
|
||
渠道同时支持按钮和回调时，优先使用按钮交互。
|
||
"""
|
||
return bool(
|
||
channel
|
||
and ChannelCapabilityManager.supports_buttons(channel)
|
||
and ChannelCapabilityManager.supports_callbacks(channel)
|
||
)
|
||
|
||
|
||
def supports_markdown(channel: Optional[NotificationChannel]) -> bool:
|
||
"""
|
||
仅在支持 Markdown 的渠道上输出 Markdown 内容。
|
||
"""
|
||
return bool(channel and ChannelCapabilityManager.supports_markdown(channel))
|
||
|
||
|
||
def page_items(
|
||
items: Sequence[Any],
|
||
page: int,
|
||
page_size: int,
|
||
) -> Tuple[List[Any], int, int]:
|
||
"""
|
||
对列表做分页并规范化页码。
|
||
"""
|
||
total = len(items)
|
||
if total == 0:
|
||
return [], 0, 1
|
||
total_pages = max(1, math.ceil(total / max(1, page_size)))
|
||
page = min(max(0, page), total_pages - 1)
|
||
start = page * page_size
|
||
end = start + page_size
|
||
return list(items[start:end]), page, total_pages
|
||
|
||
|
||
def build_navigation_buttons(
|
||
prefix: str,
|
||
request: _UserInteractionEntry,
|
||
page: int,
|
||
total_pages: int,
|
||
) -> List[List[dict[str, str]]]:
|
||
"""
|
||
构造标准上一页/下一页按钮。
|
||
"""
|
||
buttons = []
|
||
nav_row = []
|
||
if page > 0:
|
||
nav_row.append(
|
||
{
|
||
"text": "⬅️ 上一页",
|
||
"callback_data": f"{prefix}:{request.request_id}:page-prev",
|
||
}
|
||
)
|
||
if page < total_pages - 1:
|
||
nav_row.append(
|
||
{
|
||
"text": "下一页 ➡️",
|
||
"callback_data": f"{prefix}:{request.request_id}:page-next",
|
||
}
|
||
)
|
||
if nav_row:
|
||
buttons.append(nav_row)
|
||
return buttons
|
||
|
||
|
||
def update_or_post_message(
|
||
chain: MessageGateway,
|
||
channel: Optional[NotificationChannel],
|
||
source: Optional[str],
|
||
userid: Optional[Union[str, int]],
|
||
username: Optional[str],
|
||
title: str,
|
||
text: str,
|
||
buttons: Optional[List[List[dict[str, str]]]] = None,
|
||
original_message_id: Optional[Union[str, int]] = None,
|
||
original_chat_id: Optional[str] = None,
|
||
) -> None:
|
||
"""
|
||
优先编辑原消息，失败时回退为发送新消息。
|
||
"""
|
||
if (
|
||
original_message_id
|
||
and original_chat_id
|
||
and channel is not None
|
||
and ChannelCapabilityManager.supports_editing(channel)
|
||
):
|
||
edit_kwargs = {}
|
||
if channel == NotificationChannel.WebAgent:
|
||
edit_kwargs["metadata"] = {"userid": userid}
|
||
edited = chain.edit_message(
|
||
channel=channel,
|
||
source=source,
|
||
message_id=original_message_id,
|
||
chat_id=original_chat_id,
|
||
title=title,
|
||
text=text,
|
||
buttons=buttons,
|
||
**edit_kwargs,
|
||
)
|
||
if edited:
|
||
return
|
||
|
||
chain.post_message(
|
||
Message(
|
||
channel=channel,
|
||
source=source,
|
||
userid=userid,
|
||
username=username,
|
||
title=title,
|
||
text=text,
|
||
buttons=buttons,
|
||
# 编辑失败回退发新消息时保留原消息上下文，
|
||
# 保证飞书等渠道能回复到原会话（如群聊），而不是发给私聊。
|
||
original_message_id=original_message_id,
|
||
original_chat_id=original_chat_id,
|
||
save_history=False,
|
||
)
|
||
)
|
||
|
||
|
||
def escape_markdown_table_cell(value: object) -> str:
|
||
"""
|
||
最小化转义 Markdown 表格中的特殊字符。
|
||
"""
|
||
text = str(value or "").replace("\n", "<br>")
|
||
return text.replace("|", "\\|")
|
||
|
||
|
||
def format_markdown_table(
|
||
headers: Sequence[str],
|
||
rows: Sequence[Sequence[object]],
|
||
) -> str:
|
||
"""
|
||
生成 Markdown 表格文本。
|
||
"""
|
||
header_line = (
|
||
"| "
|
||
+ " | ".join(escape_markdown_table_cell(item) for item in headers)
|
||
+ " |"
|
||
)
|
||
separator_line = "| " + " | ".join("---" for _ in headers) + " |"
|
||
data_lines = [
|
||
"| "
|
||
+ " | ".join(escape_markdown_table_cell(item) for item in row)
|
||
+ " |"
|
||
for row in rows
|
||
]
|
||
return "\n".join([header_line, separator_line, *data_lines])
|