mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 03:48:03 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
438 lines
14 KiB
Python
438 lines
14 KiB
Python
"""
|
||
提供消息历史、通知回调和浏览器 Web Push 订阅管理接口。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
from typing import Annotated, Any, List, Optional, Protocol, Union
|
||
|
||
from fastapi import Depends, Request
|
||
from starlette.responses import PlainTextResponse
|
||
|
||
from app.adapters.external.wechat import WXBizMsgCrypt
|
||
from app.adapters.web.security.access import verify_apitoken, verify_token
|
||
from app.api.context import get_background_task_registry, resolve_background_task_registry
|
||
from app.api.dependencies.agent import get_message_query_service
|
||
from app.api.dependencies.auth import get_current_active_superuser
|
||
from app.api.principal import ApiPrincipal
|
||
from app.api.response import ResponseAPIRouter
|
||
from app.application.configuration import (
|
||
get_api_runtime_config_snapshot,
|
||
get_configured_system_config,
|
||
)
|
||
from app.application.messaging.message import MessageQueryService
|
||
from app.application.notification import get_notification_configs
|
||
from app.chain.message import MessageChain
|
||
from app.runtime.log import logger
|
||
from app.runtime.tasks import TaskRegistry
|
||
from app.runtime.webpush import webpush_registry
|
||
from app.schemas.message import MessageClearBefore as _SchemaMessageClearBefore
|
||
from app.schemas.message import MessageClearData as _SchemaMessageClearData
|
||
from app.schemas.message import MessageClearScope as _SchemaMessageClearScope
|
||
from app.schemas.message import MessageHistoryItem as _SchemaMessageHistoryItem
|
||
from app.schemas.message import Subscription as _SchemaSubscription
|
||
from app.schemas.message import SubscriptionMessage as _SchemaSubscriptionMessage
|
||
from app.schemas.message import WebMessageItem as _SchemaWebMessageItem
|
||
from app.schemas.response import Response as _SchemaResponse
|
||
from app.schemas.token import TokenPayload as _SchemaTokenPayload
|
||
from app.schemas.types import NotificationChannel, SystemConfigKey
|
||
|
||
router = ResponseAPIRouter()
|
||
|
||
# WNS 未指定有效期时按一天发送，单位为秒。
|
||
_WNS_DEFAULT_TTL = 86400
|
||
|
||
|
||
class WebPushError(Protocol):
|
||
"""
|
||
Web Push 订阅状态判断所需的最小异常协议。
|
||
|
||
:ivar response: 推送服务的 HTTP 响应；失败前未收到响应时为 None
|
||
"""
|
||
|
||
response: Any
|
||
|
||
|
||
def is_webpush_subscription_gone(error: WebPushError) -> bool:
|
||
"""
|
||
判断 Web Push 订阅是否已在浏览器或推送服务侧失效。
|
||
"""
|
||
response: Any = getattr(error, "response", None)
|
||
status_code = getattr(response, "status_code", None) or getattr(
|
||
response,
|
||
"status",
|
||
None,
|
||
)
|
||
return status_code in {404, 410}
|
||
|
||
|
||
def is_wns_endpoint(endpoint: str | None) -> bool:
|
||
"""
|
||
判断是否为 Microsoft WNS 推送端点。
|
||
"""
|
||
return bool(endpoint and "notify.windows.com" in endpoint)
|
||
|
||
|
||
def webpush_options_for_endpoint(endpoint: str | None) -> dict[str, Any]:
|
||
"""
|
||
返回指定推送端点需要的 Web Push 附加参数（ttl 与请求头）。
|
||
"""
|
||
if not is_wns_endpoint(endpoint):
|
||
return {}
|
||
return {
|
||
"ttl": _WNS_DEFAULT_TTL,
|
||
"headers": {"X-WNS-Cache-Policy": "cache"},
|
||
}
|
||
|
||
|
||
def _normalize_notification_clear_timestamp(value: Any) -> int:
|
||
"""
|
||
规范化通知清理时间戳。
|
||
"""
|
||
try:
|
||
normalized_value = int(value or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
return normalized_value if normalized_value > 0 else 0
|
||
|
||
|
||
def _get_notification_clear_before() -> _SchemaMessageClearBefore:
|
||
"""
|
||
读取通知中心清理时间配置。
|
||
"""
|
||
value = get_configured_system_config().get(SystemConfigKey.NotificationClearBefore)
|
||
if isinstance(value, dict):
|
||
return _SchemaMessageClearBefore(
|
||
all=_normalize_notification_clear_timestamp(value.get("all")),
|
||
system=_normalize_notification_clear_timestamp(value.get("system")),
|
||
media=_normalize_notification_clear_timestamp(value.get("media")),
|
||
)
|
||
return _SchemaMessageClearBefore(
|
||
all=_normalize_notification_clear_timestamp(value),
|
||
)
|
||
|
||
|
||
def _format_notification_clear_time(value: int) -> Optional[str]:
|
||
"""
|
||
将清理时间戳转换为消息表使用的时间字符串。
|
||
"""
|
||
if not value:
|
||
return None
|
||
timestamp = value / 1000 if value > 10000000000 else value
|
||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
|
||
|
||
|
||
def start_message_chain(body: Any, form: Any, args: Any) -> None:
|
||
"""
|
||
启动链式任务
|
||
"""
|
||
MessageChain().process(body=body, form=form, args=args)
|
||
|
||
|
||
@router.post("/", summary="接收用户消息", response_model=_SchemaResponse[None])
|
||
async def user_message(
|
||
task_registry: Annotated[TaskRegistry, Depends(get_background_task_registry)],
|
||
request: Request,
|
||
_: _SchemaTokenPayload = Depends(verify_apitoken),
|
||
) -> _SchemaResponse[None]:
|
||
"""
|
||
用户消息响应；推荐通过 X-API-KEY 请求头传递 API_TOKEN，查询参数 token 仅保留兼容。
|
||
"""
|
||
body = await request.body()
|
||
form = await request.form()
|
||
args = request.query_params
|
||
source = args.get("source")
|
||
content_type = request.headers.get("content-type") or ""
|
||
body_text = body.decode("utf-8", errors="replace")
|
||
image_markers = [
|
||
marker
|
||
for marker in (
|
||
'"photo"',
|
||
'"document"',
|
||
'"files"',
|
||
'"attachments"',
|
||
'"url_private"',
|
||
'"image/"',
|
||
'"image_url"',
|
||
)
|
||
if marker in body_text
|
||
]
|
||
logger.info(
|
||
"消息入口收到请求: source=%s, content_type=%s, body_bytes=%s, form_keys=%s, image_markers=%s",
|
||
source,
|
||
content_type,
|
||
len(body),
|
||
list(form.keys()) if form else [],
|
||
image_markers,
|
||
)
|
||
resolve_background_task_registry(task_registry).create_sync(
|
||
start_message_chain, body, form, args, owner="api.message.user"
|
||
)
|
||
return _SchemaResponse(success=True)
|
||
|
||
|
||
@router.post("/web", summary="接收WEB消息", response_model=_SchemaResponse[None])
|
||
async def web_message(
|
||
request: Request,
|
||
text: Optional[str] = None,
|
||
current_user: ApiPrincipal = Depends(get_current_active_superuser),
|
||
) -> _SchemaResponse[None]:
|
||
"""
|
||
WEB消息响应
|
||
"""
|
||
images = None
|
||
content_type = request.headers.get("content-type") or ""
|
||
if "application/json" in content_type:
|
||
try:
|
||
payload = await request.json()
|
||
except Exception:
|
||
payload = None
|
||
if isinstance(payload, dict):
|
||
text = payload.get("text", text)
|
||
image = payload.get("image")
|
||
images = payload.get("images")
|
||
if image:
|
||
if isinstance(images, list):
|
||
images = [*images, image]
|
||
else:
|
||
images = [image]
|
||
elif isinstance(images, str):
|
||
images = [images]
|
||
|
||
MessageChain().handle_message(
|
||
channel=NotificationChannel.Web,
|
||
source=current_user.name,
|
||
userid=current_user.name,
|
||
username=current_user.name,
|
||
text=text or "",
|
||
images=images,
|
||
)
|
||
return _SchemaResponse(success=True)
|
||
|
||
|
||
@router.get("/web", summary="获取WEB消息", response_model=List[_SchemaWebMessageItem])
|
||
async def get_web_message(
|
||
_: _SchemaTokenPayload = Depends(verify_token),
|
||
service: MessageQueryService = Depends(get_message_query_service),
|
||
page: Optional[int] = 1,
|
||
count: Optional[int] = 20,
|
||
) -> List[dict[str, Any]]:
|
||
"""
|
||
获取WEB消息列表
|
||
"""
|
||
return await service.list_web(
|
||
page=page if page is not None else 1,
|
||
count=count if count is not None else 20,
|
||
)
|
||
|
||
|
||
@router.get("/notification", summary="获取通知消息", response_model=List[_SchemaMessageHistoryItem])
|
||
async def get_notification_message(
|
||
_: _SchemaTokenPayload = Depends(verify_token),
|
||
service: MessageQueryService = Depends(get_message_query_service),
|
||
page: Optional[int] = 1,
|
||
count: Optional[int] = 20,
|
||
) -> List[_SchemaMessageHistoryItem]:
|
||
"""
|
||
获取系统发送的通知消息列表。
|
||
"""
|
||
clear_before = _get_notification_clear_before()
|
||
messages = await service.list_notifications(
|
||
page=page if page is not None else 1,
|
||
count=count if count is not None else 20,
|
||
all_clear_before=_format_notification_clear_time(clear_before.all),
|
||
system_clear_before=_format_notification_clear_time(clear_before.system),
|
||
media_clear_before=_format_notification_clear_time(clear_before.media),
|
||
)
|
||
return [_SchemaMessageHistoryItem(**message) for message in messages]
|
||
|
||
|
||
@router.delete(
|
||
"/notification",
|
||
summary="清理通知消息",
|
||
response_model=_SchemaResponse[_SchemaMessageClearData],
|
||
)
|
||
async def clear_notification_message(
|
||
scope: _SchemaMessageClearScope = _SchemaMessageClearScope.All,
|
||
_: _SchemaTokenPayload = Depends(verify_token),
|
||
) -> _SchemaResponse[dict[str, Any]]:
|
||
"""
|
||
记录通知中心清理时间，后续通知历史查询会在服务端过滤。
|
||
"""
|
||
clear_before = _get_notification_clear_before()
|
||
value = clear_before.model_dump()
|
||
value[scope.value] = int(time.time() * 1000)
|
||
await get_configured_system_config().async_set(SystemConfigKey.NotificationClearBefore, value)
|
||
return _SchemaResponse(success=True, data={"clear_before": value})
|
||
|
||
|
||
def wechat_verify(
|
||
echostr: str,
|
||
msg_signature: str,
|
||
timestamp: Union[str, int],
|
||
nonce: str,
|
||
source: Optional[str] = None,
|
||
) -> Any:
|
||
"""
|
||
微信验证响应
|
||
"""
|
||
# 获取服务配置
|
||
client_configs = get_notification_configs(include_disabled=True)
|
||
if not client_configs:
|
||
return "未找到对应的消息配置"
|
||
client_config = next(
|
||
(
|
||
config
|
||
for config in client_configs
|
||
if config.type == "wechat"
|
||
and config.enabled
|
||
and (config.config or {}).get("WECHAT_MODE", "app") != "bot"
|
||
and (not source or config.name == source)
|
||
),
|
||
None,
|
||
)
|
||
if not client_config:
|
||
return "未找到对应的消息配置"
|
||
config = client_config.config or {}
|
||
try:
|
||
wxcpt = WXBizMsgCrypt(
|
||
sToken=config.get("WECHAT_TOKEN"),
|
||
sEncodingAESKey=config.get("WECHAT_ENCODING_AESKEY"),
|
||
sReceiveId=config.get("WECHAT_CORPID"),
|
||
)
|
||
ret, sEchoStr = wxcpt.VerifyURL(
|
||
sMsgSignature=msg_signature,
|
||
sTimeStamp=timestamp,
|
||
sNonce=nonce,
|
||
sEchoStr=echostr,
|
||
)
|
||
if ret == 0:
|
||
# 验证URL成功，将sEchoStr返回给企业号
|
||
return PlainTextResponse(sEchoStr)
|
||
return "微信验证失败"
|
||
except Exception as err:
|
||
logger.error(f"微信请求验证失败: {str(err)}", exc_info=True)
|
||
return "消息验证失败，请稍后重试"
|
||
|
||
|
||
def vocechat_verify() -> Any:
|
||
"""
|
||
VoceChat验证响应
|
||
"""
|
||
return {"status": "OK"}
|
||
|
||
|
||
@router.get(
|
||
"/",
|
||
summary="回调请求验证",
|
||
response_model=None,
|
||
responses={
|
||
200: {
|
||
"description": "消息平台原生验证响应",
|
||
"content": {
|
||
"text/plain": {"schema": {"type": "string"}},
|
||
"application/json": {
|
||
"schema": {
|
||
"type": "object",
|
||
"properties": {"status": {"type": "string"}},
|
||
}
|
||
},
|
||
},
|
||
}
|
||
},
|
||
)
|
||
def incoming_verify(
|
||
token: Optional[str] = None,
|
||
echostr: Optional[str] = None,
|
||
msg_signature: Optional[str] = None,
|
||
timestamp: Optional[Union[str, int]] = None,
|
||
nonce: Optional[str] = None,
|
||
source: Optional[str] = None,
|
||
_: _SchemaTokenPayload = Depends(verify_apitoken),
|
||
) -> Any:
|
||
"""
|
||
微信/VoceChat等验证响应
|
||
"""
|
||
logger.info(
|
||
f"收到验证请求: echostr={echostr}, "
|
||
f"msg_signature={msg_signature}, timestamp={timestamp}, nonce={nonce}"
|
||
)
|
||
if echostr and msg_signature and timestamp and nonce:
|
||
return wechat_verify(echostr, msg_signature, timestamp, nonce, source)
|
||
return vocechat_verify()
|
||
|
||
|
||
@router.post(
|
||
"/webpush/subscribe",
|
||
summary="客户端webpush通知订阅",
|
||
response_model=_SchemaResponse[None],
|
||
)
|
||
async def subscribe(
|
||
subscription: _SchemaSubscription, _: _SchemaTokenPayload = Depends(verify_token)
|
||
) -> _SchemaResponse[None]:
|
||
"""
|
||
客户端webpush通知订阅
|
||
"""
|
||
subinfo = subscription.model_dump()
|
||
webpush_registry.upsert(subinfo)
|
||
logger.debug(f"通知订阅成功: {subinfo}")
|
||
return _SchemaResponse(success=True)
|
||
|
||
|
||
@router.post(
|
||
"/webpush/send", summary="发送webpush通知", response_model=_SchemaResponse[None]
|
||
)
|
||
def send_notification(
|
||
payload: _SchemaSubscriptionMessage,
|
||
_: ApiPrincipal = Depends(get_current_active_superuser),
|
||
) -> _SchemaResponse[None]:
|
||
"""
|
||
发送webpush通知；会推送到全部已订阅浏览器，因此仅限超级管理员调用
|
||
"""
|
||
from app.adapters.network.webpush import WebPushDeliveryError, send_webpush
|
||
|
||
subscriptions = webpush_registry.list()
|
||
if not subscriptions:
|
||
return _SchemaResponse(success=True, message="没有可发送的浏览器通知")
|
||
|
||
success_count = 0
|
||
failure_count = 0
|
||
for sub in subscriptions:
|
||
try:
|
||
send_webpush(
|
||
subscription=sub,
|
||
data=json.dumps(payload.model_dump()),
|
||
vapid_private_key=get_api_runtime_config_snapshot().vapid_private_key,
|
||
vapid_subject=get_api_runtime_config_snapshot().vapid_subject,
|
||
**webpush_options_for_endpoint(sub.get("endpoint")),
|
||
)
|
||
success_count += 1
|
||
except WebPushDeliveryError as err:
|
||
logger.error(f"WebPush发送失败: {str(err)}")
|
||
if is_webpush_subscription_gone(err) and webpush_registry.remove(sub):
|
||
logger.info(f"已移除失效WebPush订阅: {sub.get('endpoint')}")
|
||
failure_count += 1
|
||
except Exception as err:
|
||
logger.error(f"WebPush发送失败: {str(err)}", exc_info=True)
|
||
failure_count += 1
|
||
|
||
if not failure_count:
|
||
return _SchemaResponse(
|
||
success=True,
|
||
message=f"消息已发送到 {success_count} 个设备",
|
||
)
|
||
if success_count:
|
||
return _SchemaResponse(
|
||
success=True,
|
||
message=(
|
||
f"消息已发送到 {success_count} 个设备，"
|
||
f"{failure_count} 个设备发送失败"
|
||
),
|
||
)
|
||
return _SchemaResponse(
|
||
success=False,
|
||
message="消息发送失败，请检查浏览器通知权限后重试",
|
||
)
|