diff --git a/Dockerfile b/Dockerfile
index 1996a876..2b2d5299 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -3,7 +3,9 @@
FROM ghcr.io/astral-sh/uv:0.13-python3.14-alpine AS builder
WORKDIR /app
-ENV UV_LINK_MODE=copy
+# 依赖与应用源码都预编译为 .pyc,容器冷启动不再逐个编译
+ENV UV_LINK_MODE=copy \
+ UV_COMPILE_BYTECODE=1
# Install dependencies (cached layer)
COPY backend/pyproject.toml backend/uv.lock ./
@@ -11,6 +13,7 @@ RUN uv sync --frozen --no-dev
# Copy application source
COPY backend/src ./src
+RUN .venv/bin/python -m compileall -q src
FROM python:3.14-alpine AS runtime
diff --git a/backend/src/main.py b/backend/src/main.py
index 1d56c447..6d3976e0 100644
--- a/backend/src/main.py
+++ b/backend/src/main.py
@@ -1,19 +1,25 @@
+import asyncio
import logging
import os
+from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
+import anyio
import uvicorn
-from fastapi import APIRouter, Depends, FastAPI, Response
+from anyio.abc import TaskGroup, TaskStatus
+from fastapi import APIRouter, Depends, FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from starlette.routing import Route
+from starlette.types import ASGIApp, Receive, Scope, Send
from module.api import v1
from module.api.health import router as health_router
from module.conf import VERSION, settings, setup_logger
from module.core import AppContext
-from module.mcp import create_mcp_app, mcp
+from module.mcp.security import McpAccessMiddleware
setup_logger(reset=True)
logger = logging.getLogger(__name__)
@@ -46,14 +52,64 @@ async def lifespan(app: FastAPI):
# supervised background task and returns immediately (CONTRACT #5),
# so this does not block server startup.
await ctx.start_tasks()
- # 子应用不会运行自己的 lifespan,MCP 会话管理器在这里启动
- async with app.state.mcp_sessions.run():
+ # 子应用不会运行自己的 lifespan,MCP 会话管理器在这里提供的任务组里运行
+ async with app.state.mcp.running():
yield
# Shutdown
await ctx.stop()
await ctx.plugins.stop()
+class _LazyMcpApp:
+ """首个通过访问控制的 /mcp 请求才导入 mcp 库、构建子应用并启动会话管理器。
+
+ 多数用户不用 MCP,启动时不加载可省约 20 MB 内存。会话管理器的 run() 内含
+ anyio 任务组,必须在同一个任务里进出,所以它跑在 lifespan 持有的任务组里,
+ 关停时随任务组取消。
+ """
+
+ def __init__(self, ctx: AppContext) -> None:
+ self._ctx = ctx
+ self._tasks: TaskGroup | None = None
+ self._app: ASGIApp | None = None
+ # 并发的首批请求只构建一次
+ self._lock = asyncio.Lock()
+
+ @asynccontextmanager
+ async def running(self) -> AsyncIterator[None]:
+ async with anyio.create_task_group() as tasks:
+ self._tasks = tasks
+ yield
+ tasks.cancel_scope.cancel()
+ # 每次 create_mcp_app 都新建会话管理器,再次启动时重新构建
+ self._tasks = self._app = None
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ if self._app is None:
+ async with self._lock:
+ if self._app is None:
+ self._app = await self._build()
+ await self._app(scope, receive, send)
+
+ async def _build(self) -> ASGIApp:
+ if self._tasks is None:
+ raise RuntimeError("MCP 只能在应用 lifespan 运行期间使用")
+ from module.mcp import create_mcp_app, mcp
+
+ app = create_mcp_app(self._ctx)
+ manager = mcp.session_manager
+
+ async def serve(
+ *, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED
+ ) -> None:
+ async with manager.run():
+ task_status.started()
+ await anyio.sleep_forever()
+
+ await self._tasks.start(serve)
+ return app
+
+
def create_app() -> FastAPI:
ctx = AppContext.build(settings)
app = FastAPI(lifespan=lifespan)
@@ -67,6 +123,9 @@ def create_app() -> FastAPI:
allow_headers=["*"],
)
+ # SSE(text/event-stream)由中间件默认排除,不会被压缩缓冲
+ app.add_middleware(GZipMiddleware, minimum_size=1000)
+
# mount routers
app.include_router(v1, prefix="/api")
@@ -75,10 +134,10 @@ def create_app() -> FastAPI:
app.include_router(health_router)
# MCP 服务(Streamable HTTP):用路由而不是 mount,端点才是 /mcp 本身
- # (mount 只匹配 /mcp/...,/mcp 会被 307 重定向到 /mcp/)
- app.router.routes.append(Route("/mcp", endpoint=create_mcp_app(ctx)))
- # 每次 create_mcp_app 都新建会话管理器,记下本应用的这一个供 lifespan 运行
- app.state.mcp_sessions = mcp.session_manager
+ # (mount 只匹配 /mcp/...,/mcp 会被 307 重定向到 /mcp/)。访问控制在
+ # 延迟构建之前,被拒的请求不会加载 mcp 库
+ app.state.mcp = _LazyMcpApp(ctx)
+ app.router.routes.append(Route("/mcp", endpoint=McpAccessMiddleware(app.state.mcp)))
return app
@@ -99,14 +158,24 @@ def posters(path: str):
# 前端插件组件与宿主同源(第 3.8 节),只允许加载本站脚本:远程脚本与内联脚本
# 都被拒绝。只加在前端静态文件上,/docs 的 Swagger 页面依赖内联脚本与 CDN。
-def _spa_csp(response: Response) -> None:
+def _spa_headers(request: Request, response: Response) -> None:
response.headers["Content-Security-Policy"] = "script-src 'self'"
+ # 入口 HTML(含回落到 index.html 的前端路由深链接,即末段无扩展名的路径)
+ # 必须每次校验,否则会引用已不存在的旧哈希文件;assets/ 下的文件名带内容
+ # 哈希,可永久缓存。fonts/、images/ 等不带哈希,不设置。只看路径段,
+ # 带 root_path 前缀时同样适用。
+ path = request.url.path
+ last = path.rsplit("/", 1)[-1]
+ if "." not in last or last.endswith(".html"):
+ response.headers["Cache-Control"] = "no-cache"
+ elif "/assets/" in path:
+ response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
def mount_webui(app: FastAPI, dist: Path) -> None:
# 前端路由优先级最低,只在没有路径操作或挂载匹配时生效;浏览器导航
# (Accept: text/html)到不存在的路径时回落到 index.html,其它请求返回 404
- web = APIRouter(dependencies=[Depends(_spa_csp)])
+ web = APIRouter(dependencies=[Depends(_spa_headers)])
web.frontend("/", directory=dist)
app.include_router(web)
diff --git a/backend/src/module/api/events.py b/backend/src/module/api/events.py
index ab18216b..dc6d5e79 100644
--- a/backend/src/module/api/events.py
+++ b/backend/src/module/api/events.py
@@ -3,7 +3,13 @@
单个连接按 tick 节拍分别推送 status/downloader/log/update,节拍与原轮询间隔
保持一致,前端订阅失败时回退到原有轮询逻辑。这几类是「状态快照」而非事件,
-仍按节拍采样。
+仍按节拍采样,但只推变化:
+
+- downloader:所有连接共用一份 5s 缓存查询下载器;连接建立时推一帧,之后
+ 仅在负载与本连接上次推送的不同时才推。
+- log:连接建立时、以及日志被轮转/截断后推整段尾部(``log``,日志为空时推
+ ``log_clear``);其余只推上次之后新增的完整行(``log_append``,总以换行
+ 结尾),没有新增不推。
notification(通知中心未读数)由事件驱动:每个连接在插件事件总线上订阅
``inbox.changed``(通知中心写入/已读/删除时发布),收到后立即查库推送,
@@ -17,6 +23,8 @@ import asyncio
import dataclasses
import json
import logging
+import os
+import time
from collections import deque
from collections.abc import AsyncIterator
@@ -24,6 +32,7 @@ from fastapi import APIRouter, Depends, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent
from ab_sdk import Event
+from module.api import log as log_api
from module.api.deps import get_context
from module.api.log import _read_log_tail
from module.conf import LOG_PATH, VERSION
@@ -55,6 +64,9 @@ _LOG_EVERY = 10
# 取值需明显小于 downloader 的 5s 推送间隔。
_DOWNLOADER_TIMEOUT_SECONDS = 3.0
+# 下载器种子列表的共享缓存有效期(秒):所有 SSE 连接合计每 5s 最多查询一次
+_DOWNLOADER_CACHE_SECONDS = 5.0
+
# 单个连接最多积压的待转发总线事件数
_FORWARD_BACKLOG = 200
@@ -87,12 +99,91 @@ async def _notification_payload() -> dict:
}
-async def _log_payload() -> str | None:
- """读取日志尾部;日志文件不存在时返回 None(跳过本次推送)。"""
- if not LOG_PATH.exists():
- return None
- data = await asyncio.to_thread(_read_log_tail)
- return data.decode("utf-8", errors="replace")
+class _TorrentCache:
+ """所有 SSE 连接共用的种子列表缓存(已编码为 JSON)。
+
+ 锁保证并发连接同一时刻只查一次下载器;时间戳记在查询开始时,
+ 连接按约 5s 的节拍来取时不会拿到隔一轮的旧数据。
+ """
+
+ def __init__(self) -> None:
+ self._lock = asyncio.Lock()
+ self._fetched_at = -_DOWNLOADER_CACHE_SECONDS
+ self._data: str | None = None
+
+ async def get(self) -> str:
+ async with self._lock:
+ now = time.monotonic()
+ if (
+ self._data is None
+ or now - self._fetched_at >= _DOWNLOADER_CACHE_SECONDS
+ ):
+ # 查询被取消(持锁的连接断开)时不更新时间戳,下一个连接重新查询
+ self._data = json.dumps(await _downloader_payload())
+ self._fetched_at = now
+ return self._data
+
+
+_torrents = _TorrentCache()
+
+
+class _LogCursor:
+ """一个 SSE 连接的日志读取位置。
+
+ 首帧及日志被轮转(inode 变化)、截断(大小小于已读偏移)、mtime 回退或
+ 经 /log/clear 清空(清空计数变化)时推整段尾部 ``log``(为空时推
+ ``log_clear``);否则只推偏移
+ 之后新增的完整行 ``log_append``。增量截到最后一个换行:不切断多字节
+ 字符,且每帧都是完整行,前端可直接追加。
+ """
+
+ def __init__(self) -> None:
+ self._ino: int | None = None
+ self._mtime_ns = 0
+ self._offset = 0
+ self._generation = -1
+
+ def _read(self) -> dict | None:
+ try:
+ st = os.stat(LOG_PATH)
+ generation = log_api._clear_generation
+ if (
+ generation == self._generation
+ and st.st_ino == self._ino
+ and st.st_size >= self._offset
+ and st.st_mtime_ns >= self._mtime_ns
+ ):
+ self._mtime_ns = st.st_mtime_ns
+ if st.st_size == self._offset:
+ return None
+ with open(LOG_PATH, "rb") as f:
+ f.seek(self._offset)
+ data = f.read(st.st_size - self._offset)
+ data = data[: data.rfind(b"\n") + 1]
+ if not data:
+ return None
+ self._offset += len(data)
+ event = "log_append"
+ else:
+ data, offset = _read_log_tail()
+ self._ino, self._mtime_ns, self._offset, self._generation = (
+ st.st_ino,
+ st.st_mtime_ns,
+ offset,
+ generation,
+ )
+ event = "log"
+ if not data:
+ # EventSource 不派发 data 为空的事件:空日志改推 log_clear,
+ # 别的标签页清空后、或清空后重连时前端才不会残留旧日志
+ event, data = "log_clear", b"1"
+ except FileNotFoundError:
+ # 尚未创建,或正处于轮转改名的间隙:本轮跳过
+ return None
+ return {"event": event, "data": data.decode("utf-8", errors="replace")}
+
+ async def frame(self) -> dict | None:
+ return await asyncio.to_thread(self._read)
class _InboxWatch:
@@ -175,8 +266,10 @@ async def _event_generator(request: Request, ctx: AppContext):
"""按各自节拍推送 status/downloader/log 事件、按需推送 notification 的核心循环。"""
loop = asyncio.get_running_loop()
inbox = _InboxWatch()
+ log = _LogCursor()
tick = 0
last_update: str | None = None
+ last_downloader: str | None = None
try:
while True:
if await request.is_disconnected():
@@ -197,13 +290,15 @@ async def _event_generator(request: Request, ctx: AppContext):
if tick % _DOWNLOADER_EVERY == 0:
# 不可用时推送 null 而非跳过,前端据此感知下载器降级状态
# (store/downloader.ts 对 null 保留旧数据,不会崩溃)。
- torrents = await _downloader_payload()
- yield {"event": "downloader", "data": json.dumps(torrents)}
+ torrents = await _torrents.get()
+ if torrents != last_downloader:
+ last_downloader = torrents
+ yield {"event": "downloader", "data": torrents}
if tick % _LOG_EVERY == 0:
- log_text = await _log_payload()
- if log_text is not None:
- yield {"event": "log", "data": log_text}
+ log_frame = await log.frame()
+ if log_frame is not None:
+ yield log_frame
# 通知中心:连接建立时的初始帧,以及此前查库失败后的重试
frame = await inbox.frame()
diff --git a/backend/src/module/api/log.py b/backend/src/module/api/log.py
index 5600444c..5e6482f5 100644
--- a/backend/src/module/api/log.py
+++ b/backend/src/module/api/log.py
@@ -16,11 +16,12 @@ router = APIRouter(
_TAIL_BYTES = 512 * 1024 # 512 KB
-def _read_file_tail(path: Path, budget: int) -> tuple[bytes, bool]:
+def _read_file_tail(path: Path, budget: int) -> tuple[bytes, bool, int]:
"""同步读取单个文件的最后 budget 字节(掐掉开头的半行)。
- 返回 (数据, 是否截断)。文件在 stat 与 open 之间被轮转改名时按
- 空文件处理——抛异常会炸掉 GET /log 与整条 SSE 流。
+ 返回 (数据, 是否截断, 读到的文件末尾偏移)。末尾偏移供 SSE 作为增量读取的
+ 起点。文件在 stat 与 open 之间被轮转改名时按空文件处理——抛异常会炸掉
+ GET /log 与整条 SSE 流。
"""
try:
with open(path, "rb") as f:
@@ -29,52 +30,63 @@ def _read_file_tail(path: Path, budget: int) -> tuple[bytes, bool]:
if size > budget:
f.seek(-budget, 2)
data = f.read()
+ end = f.tell()
# Drop first partial line
idx = data.find(b"\n")
if idx != -1:
data = data[idx + 1 :]
- return data, True
+ return data, True, end
f.seek(0)
- return f.read(), False
+ data = f.read()
+ return data, False, f.tell()
except FileNotFoundError:
- return b"", False
+ return b"", False, 0
-def _read_log_tail() -> bytes:
+def _read_log_tail() -> tuple[bytes, int]:
"""读取日志尾部,供 asyncio.to_thread 在线程池中执行。
轮转(RotatingFileHandler / 启动轮转)刚发生后 log.txt 近乎为空;
仅当 log.txt 完整读入(未截断)且预算有剩时,把 log.txt.1 的尾部
拼在前面,UI 不出现"日志突然清空"的断崖。截断读取时绝不拼接:
中间内容已缺失,再贴更旧的备份会造成时间倒跳的假象。
+
+ 返回 (数据, log.txt 读到的末尾偏移)。
"""
- data, truncated = _read_file_tail(LOG_PATH, _TAIL_BYTES)
+ data, truncated, end = _read_file_tail(LOG_PATH, _TAIL_BYTES)
if truncated:
- return data
+ return data, end
remaining = _TAIL_BYTES - len(data)
if remaining > 0:
- backup_data, _ = _read_file_tail(
+ backup_data, _, _ = _read_file_tail(
LOG_PATH.with_name(f"{LOG_PATH.name}.1"), remaining
)
data = backup_data + data
- return data
+ return data, end
@router.get("", response_model=str)
async def get_log():
if LOG_PATH.exists():
# Up to 512 KB of sync file I/O; keep it off the event loop.
- data = await asyncio.to_thread(_read_log_tail)
+ data, _ = await asyncio.to_thread(_read_log_tail)
return Response(data, media_type="text/plain")
else:
return Response("Log file not found", status_code=404)
+# 日志清空次数。原地截断后若在 SSE 下次读取前又长过旧偏移,大小比较发现不了;
+# SSE 的读取游标比对这个计数,变化时重推整段
+_clear_generation = 0
+
+
def _clear_log_files() -> None:
"""清空日志并删除轮转备份,否则拼接读取会把旧内容带回来。"""
+ global _clear_generation
LOG_PATH.write_text("")
for backup in LOG_PATH.parent.glob(f"{LOG_PATH.name}.*"):
backup.unlink(missing_ok=True)
+ _clear_generation += 1
@router.post("/clear", response_model=APIResponse)
diff --git a/backend/src/module/api/passkey.py b/backend/src/module/api/passkey.py
index c8cf3910..bcaf7d24 100644
--- a/backend/src/module/api/passkey.py
+++ b/backend/src/module/api/passkey.py
@@ -30,7 +30,6 @@ from module.security.api import (
require_session_principal,
)
from module.security.auth_strategy import PasskeyAuthStrategy
-from module.security.webauthn import get_webauthn_service
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/passkey", tags=["passkey"])
@@ -56,6 +55,9 @@ def _get_webauthn_from_request(request: Request):
"""
from urllib.parse import urlparse
+ # webauthn 库较重(约 11 MB),只在用到 Passkey 时导入
+ from module.security.webauthn import get_webauthn_service
+
origin = settings.security.webauthn_origin
if not origin:
origin = request.headers.get("origin", "")
diff --git a/backend/src/module/conf/log.py b/backend/src/module/conf/log.py
index 3df327a6..76f32836 100644
--- a/backend/src/module/conf/log.py
+++ b/backend/src/module/conf/log.py
@@ -86,6 +86,9 @@ def setup_logger(level: int = logging.INFO, reset: bool = False):
logging.basicConfig(
level=level,
handlers=[queue_handler],
+ # 先导入的第三方库(如 mcp)可能已让根日志器带上处理器,
+ # 此时 basicConfig 会静默跳过,日志文件就收不到任何内容
+ force=True,
)
# 压低 HTTP 请求日志:httpx2 是本项目的客户端;google-genai 与 mcp 仍依赖旧 httpx
diff --git a/backend/src/module/core/loops.py b/backend/src/module/core/loops.py
index da405e96..78659476 100644
--- a/backend/src/module/core/loops.py
+++ b/backend/src/module/core/loops.py
@@ -25,29 +25,8 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None
"""Analyse aggregate RSS feeds and refresh the RSS engine once."""
async with DownloaderPool() as downloaders:
async with Database() as db:
- engine = RSSEngine(db)
- # Analyse RSS
- rss_list = await db.rss.search_aggregate()
- for rss in rss_list:
- try:
- await analyser.rss_to_data(rss, engine)
- except Exception:
- # 仅当该 RSS 确实已在遍历期间被 API 删除时才视为预期情况;
- # 其余异常可能是真实 bug,需在 WARNING 级别带完整堆栈。
- if await db.rss.search_id(rss.id) is None:
- logger.debug(
- "Skipping RSS id=%s, deleted during iteration",
- rss.id,
- )
- else:
- logger.warning(
- "Error analysing RSS id=%s (%s)",
- rss.id,
- rss.url,
- exc_info=True,
- )
- # Run RSS Engine
- events = await engine.refresh_rss(downloaders)
+ # 每个订阅本轮只拉取一次,聚合分析与下载匹配共用结果(#1026)
+ events = await RSSEngine(db).refresh_rss(downloaders, analyser=analyser)
if events:
# 站内通知中心要求事件总是送达 notifier(send_event 内部先落库,
# 再按 settings.notification.enable 决定是否外发)。
diff --git a/backend/src/module/database/bangumi.py b/backend/src/module/database/bangumi.py
index 97ec7472..c5a5be96 100644
--- a/backend/src/module/database/bangumi.py
+++ b/backend/src/module/database/bangumi.py
@@ -601,8 +601,15 @@ class BangumiDatabase:
result = await self.session.execute(statement)
return list(result.scalars().all())
- async def match_list(self, torrent_list: list, rss_link: str) -> list:
- match_datas = await self.search_all()
+ async def match_list(
+ self,
+ torrent_list: list,
+ rss_link: str,
+ match_datas: list[Bangumi] | None = None,
+ ) -> list:
+ # 调用方可传入本轮已加载的番剧列表,省去每次 search_all
+ if match_datas is None:
+ match_datas = await self.search_all()
if not match_datas:
return torrent_list
diff --git a/backend/src/module/database/rename_operation.py b/backend/src/module/database/rename_operation.py
index f44e2311..9a6a24e2 100644
--- a/backend/src/module/database/rename_operation.py
+++ b/backend/src/module/database/rename_operation.py
@@ -63,7 +63,7 @@ class RenameOperationDatabase:
RenameOperation.target_path == target_path,
)
if active_only:
- statement = statement.where(RenameOperation.state != "done")
+ statement = statement.where(col(RenameOperation.state).not_in(["done"]))
statement = statement.order_by( # type: ignore[assignment]
col(RenameOperation.updated_at).desc()
)
diff --git a/backend/src/module/database/rss.py b/backend/src/module/database/rss.py
index 2316ab26..740131dd 100644
--- a/backend/src/module/database/rss.py
+++ b/backend/src/module/database/rss.py
@@ -116,12 +116,6 @@ class RSSDatabase:
result = await self.session.execute(select(RSSItem).where(RSSItem.enabled))
return list(result.scalars().all())
- async def search_aggregate(self) -> list[RSSItem]:
- result = await self.session.execute(
- select(RSSItem).where(and_(RSSItem.aggregate, RSSItem.enabled))
- )
- return list(result.scalars().all())
-
async def delete(self, _id: int) -> bool:
try:
# 先删除引用该 RSS 的 torrent,避免外键约束报错
diff --git a/backend/src/module/database/torrent.py b/backend/src/module/database/torrent.py
index ed532653..130a4e2a 100644
--- a/backend/src/module/database/torrent.py
+++ b/backend/src/module/database/torrent.py
@@ -5,7 +5,7 @@ from sqlalchemy import delete, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import and_, select, true
-from module.models import Torrent
+from module.models import Bangumi, RSSItem, Torrent
logger = logging.getLogger(__name__)
@@ -47,13 +47,31 @@ class TorrentDatabase:
)
return list(result.scalars().all())
- async def check_new(self, torrents_list: list[Torrent]) -> list[Torrent]:
+ async def check_new(
+ self, torrents_list: list[Torrent], rss: RSSItem | None = None
+ ) -> list[Torrent]:
+ """返回库中没有记录的种子。
+
+ 给出 ``rss`` 时只看该订阅已匹配、且所属番剧的 ``rss_link`` 已包含该
+ 订阅当前 URL 的记录(与 BangumiDatabase.match_list 的追加判断一致):
+ 未匹配的孤儿每轮仍要重新分析,新番或上次分析失败的种子才能生成规则;
+ 订阅改了 URL 时仍要匹配,才能把新 URL 追加到 rss_link(#1053)。
+ """
if not torrents_list:
return []
urls = [t.url for t in torrents_list]
# SQLModel 类属性在 mypy 看来是普通字段类型而非 InstrumentedAttribute,
# 无法识别 .in_()/.is_() 等查询方法(无官方 mypy 插件支持)。
statement = select(Torrent.url).where(Torrent.url.in_(urls)) # type: ignore[attr-defined]
+ if rss is not None:
+ # instr 区分大小写,与 Python 的 ``in`` 一致(LIKE 不区分)
+ statement = statement.join(
+ Bangumi,
+ Bangumi.id == Torrent.bangumi_id, # type: ignore[arg-type]
+ ).where(
+ Torrent.rss_id == rss.id,
+ func.instr(Bangumi.rss_link, rss.url) > 0,
+ )
result = await self.session.execute(statement)
existing_urls = set(result.scalars().all())
return [t for t in torrents_list if t.url not in existing_urls]
diff --git a/backend/src/module/mcp/__init__.py b/backend/src/module/mcp/__init__.py
index 7a994017..bff6b396 100644
--- a/backend/src/module/mcp/__init__.py
+++ b/backend/src/module/mcp/__init__.py
@@ -8,9 +8,24 @@ Usage::
from module.mcp import create_mcp_app
app = create_mcp_app(ctx) # returns a Starlette ASGI app, route it at /mcp
+
+``mcp`` 与 ``create_mcp_app`` 在首次访问时才导入(mcp 库较重):main 启动时
+只用到 ``module.mcp.security``,不加载 mcp 库。
"""
-# 导入即注册内置工具与资源
-from . import resources, tools # noqa: F401
-from .server import create_mcp_starlette_app as create_mcp_app
-from .server import mcp
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+ from .server import create_mcp_starlette_app as create_mcp_app
+ from .server import mcp
+
+
+def __getattr__(name: str) -> Any:
+ if name not in ("mcp", "create_mcp_app"):
+ raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+ # 导入即注册内置工具与资源
+ from . import resources, tools # noqa: F401
+ from .server import create_mcp_starlette_app, mcp
+
+ globals().update(mcp=mcp, create_mcp_app=create_mcp_starlette_app)
+ return globals()[name]
diff --git a/backend/src/module/mcp/server.py b/backend/src/module/mcp/server.py
index b998aaa1..d7e59c0b 100644
--- a/backend/src/module/mcp/server.py
+++ b/backend/src/module/mcp/server.py
@@ -28,7 +28,6 @@ from .plugins import (
read_plugin_resource,
)
from .runtime import set_context
-from .security import McpAccessMiddleware
logger = logging.getLogger(__name__)
@@ -98,8 +97,8 @@ def create_mcp_starlette_app(ctx=None) -> Starlette:
新建会话管理器,调用方须在应用 lifespan 中运行 ``mcp.session_manager.run()``。
``ctx`` is the application :class:`AppContext`; it is stored so status
- tools/resources can report live program state. ``McpAccessMiddleware`` is
- applied to enforce configurable IP whitelist and bearer token access control.
+ tools/resources can report live program state. 访问控制
+ (``McpAccessMiddleware``)由 ``main.py`` 套在 /mcp 路由外层,先于本函数生效。
"""
set_context(ctx)
app = mcp.streamable_http_app(
@@ -111,5 +110,4 @@ def create_mcp_starlette_app(ctx=None) -> Starlette:
enable_dns_rebinding_protection=False
),
)
- app.add_middleware(McpAccessMiddleware)
return app
diff --git a/backend/src/module/parser/analyser/tmdb_parser.py b/backend/src/module/parser/analyser/tmdb_parser.py
index 83359339..9917aa45 100644
--- a/backend/src/module/parser/analyser/tmdb_parser.py
+++ b/backend/src/module/parser/analyser/tmdb_parser.py
@@ -4,6 +4,7 @@ import re
import time
from collections import OrderedDict
from dataclasses import dataclass
+from typing import Any
from urllib.parse import urlencode
from module.conf import TMDB_API, settings
@@ -26,7 +27,11 @@ def _api_key() -> str:
# In-memory cache for TMDB lookups to avoid repeated API calls
_TMDB_CACHE_MAX = 512
-_tmdb_cache: OrderedDict[str, "TMDBInfo | None"] = OrderedDict()
+# 否定/回退结果(无匹配、回退到电影)的存活时间:既避免对同一个查不到的标题
+# 反复请求 TMDB,又不让一次临时故障把标题永久毒化
+_TMDB_NEGATIVE_TTL = 6 * 3600.0
+# 值为 (过期的 monotonic 时刻, 结果);inf 表示不过期
+_tmdb_cache: OrderedDict[str, "tuple[float, TMDBInfo | None]"] = OrderedDict()
def reset_cache() -> None:
@@ -64,6 +69,18 @@ class TMDBInfo:
LANGUAGE = {"zh": "zh-CN", "jp": "ja-JP", "en": "en-US"}
+class _TMDBRequestFailed(Exception):
+ """TMDB 请求失败(网络错误、5xx、重试耗尽的 429 等),区别于「无匹配」。"""
+
+
+async def _get_json(req: RequestContent, url: str) -> Any:
+ """get_json 返回 None 表示请求失败:抛出,让本次查询整体作废且不缓存。"""
+ contents = await req.get_json(url)
+ if contents is None:
+ raise _TMDBRequestFailed(url)
+ return contents
+
+
def search_url(e, key="zh"):
query = urlencode(
{
@@ -100,7 +117,7 @@ def season_url(tv_id, season_number, key):
async def is_animation(tv_id, language, req: RequestContent) -> bool:
url_info = info_url(tv_id, language)
- type_id = await req.get_json(url_info)
+ type_id = await _get_json(req, url_info)
if type_id:
for type in type_id.get("genres", []):
if type.get("id") == 16:
@@ -200,12 +217,12 @@ async def _search_movie(
电影没有季度概念,因此不复用剧集的季度/集数聚合逻辑,仅返回标题、原名、
年份与海报等基本信息。"""
url = search_movie_url(title, language)
- contents = await req.get_json(url)
- results = (contents or {}).get("results") or []
+ contents = await _get_json(req, url)
+ results = contents.get("results") or []
if not results:
url = search_movie_url(title.replace(" ", ""), language)
- contents = await req.get_json(url)
- results = (contents or {}).get("results") or []
+ contents = await _get_json(req, url)
+ results = contents.get("results") or []
if not results:
return None
movie = results[0]
@@ -234,26 +251,48 @@ async def tmdb_parser(
# URL instead of a locally-saved one, so mixing the two would poison
# whichever caller queries second.
cache_key = f"{title}:{language}:{test}:{is_movie}"
- if cache_key in _tmdb_cache:
- return _tmdb_cache[cache_key]
+ hit = _tmdb_cache.get(cache_key)
+ if hit is not None:
+ if hit[0] > time.monotonic():
+ _tmdb_cache.move_to_end(cache_key)
+ return hit[1]
+ del _tmdb_cache[cache_key]
+ try:
+ result = await _tmdb_lookup(title, language, test, is_movie)
+ except _TMDBRequestFailed as e:
+ # 临时故障不写缓存,下次照常查询
+ logger.warning("TMDB request failed for %s: %s", title, e)
+ return None
+ # 剧集结果的 season 非空;None 与电影回退(season=[])都按否定结果限时缓存
+ expires = (
+ time.monotonic() + _TMDB_NEGATIVE_TTL
+ if result is None or not result.season
+ else float("inf")
+ )
+ _tmdb_cache[cache_key] = (expires, result)
+ _tmdb_cache.move_to_end(cache_key)
+ if len(_tmdb_cache) > _TMDB_CACHE_MAX:
+ _tmdb_cache.popitem(last=False)
+ return result
+
+
+async def _tmdb_lookup(title, language, test: bool, is_movie: bool) -> TMDBInfo | None:
async with RequestContent() as req:
if is_movie:
# 已知是电影/剧场版,直接查询 search/movie,跳过剧集搜索
- result = await _search_movie(title, language, req)
- _tmdb_cache[cache_key] = result
- return result
+ return await _search_movie(title, language, req)
url = search_url(title, language)
- contents = await req.get_json(url)
+ contents = await _get_json(req, url)
if not contents:
return await _search_movie(title, language, req)
- contents = (contents or {}).get("results") or []
+ contents = contents.get("results") or []
if not contents:
url = search_url(title.replace(" ", ""), language)
- contents_resp = await req.get_json(url)
+ contents_resp = await _get_json(req, url)
if not contents_resp:
return await _search_movie(title, language, req)
- contents = (contents_resp or {}).get("results") or []
+ contents = contents_resp.get("results") or []
if not contents:
# search/tv 无结果:回退到 search/movie (剧场版等)
return await _search_movie(title, language, req)
@@ -266,12 +305,10 @@ async def tmdb_parser(
matched_id = cid
break
if matched_id is None:
- # search/tv 有结果但都不是动画:回退到 search/movie (剧场版等)。
- # Don't cache the negative result permanently — a temporary
- # TMDB hiccup shouldn't poison this title for the process lifetime.
+ # search/tv 有结果但都不是动画:回退到 search/movie (剧场版等)
return await _search_movie(title, language, req)
url_info = info_url(matched_id, language)
- info_content = await req.get_json(url_info)
+ info_content = await _get_json(req, url_info)
season = [
{
"season": s.get("name"),
@@ -355,13 +392,8 @@ async def tmdb_parser(
virtual_season_starts if virtual_season_starts else None
),
)
- if len(_tmdb_cache) >= _TMDB_CACHE_MAX:
- _tmdb_cache.popitem(last=False)
- _tmdb_cache[cache_key] = result
return result
else:
- # No results at all — don't cache the negative result permanently,
- # see the matched_id is None case above.
return None
diff --git a/backend/src/module/rss/analyser.py b/backend/src/module/rss/analyser.py
index 516805e6..93d9471a 100644
--- a/backend/src/module/rss/analyser.py
+++ b/backend/src/module/rss/analyser.py
@@ -114,28 +114,36 @@ class RSSAnalyser:
return None
async def rss_to_data(
- self, rss: RSSItem, engine: RSSEngine, full_parse: bool = True
+ self,
+ rss: RSSItem,
+ torrents: list[Torrent],
+ engine: RSSEngine,
+ bangumi_list: list[Bangumi],
) -> list[Bangumi]:
+ """用本轮已拉取的 ``torrents`` 为聚合订阅生成新规则。
+
+ ``bangumi_list`` 是本轮已加载的番剧列表,避免每个订阅各查一次。
+ """
# One RSS workflow can cross several await boundaries and parse the
# same resource during Movie match, Bangumi match, and persistence.
# Keep those stages on the engine selected when the workflow started.
with parser_engine_snapshot():
- rss_torrents = await self.get_rss_torrents(rss.url, full_parse)
+ # 只跳过追加 rss_link 已是空操作的种子:其他订阅已知、或本订阅
+ # 改过 URL 的种子仍要匹配,才能把当前 URL 追加到 rss_link(#1053)
+ torrents = await engine.db.torrent.check_new(torrents, rss=rss)
+ if not torrents:
+ return []
# Filter out already-known movies first
- torrents_after_movies = await engine.db.movie.match_list(
- rss_torrents, rss.url
- )
+ torrents_after_movies = await engine.db.movie.match_list(torrents, rss.url)
# Then filter out already-known bangumi
torrents_to_add = await engine.db.bangumi.match_list(
- torrents_after_movies, rss.url
+ torrents_after_movies, rss.url, bangumi_list
)
if not torrents_to_add:
logger.debug("No new title has been found.")
return []
# Parse remaining torrents
- new_bangumi, new_movies = await self.torrents_to_data(
- torrents_to_add, rss, full_parse
- )
+ new_bangumi, new_movies = await self.torrents_to_data(torrents_to_add, rss)
for movie in new_movies:
await engine.db.movie.add(movie)
if new_bangumi:
diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py
index 5cb30579..e4210546 100644
--- a/backend/src/module/rss/engine.py
+++ b/backend/src/module/rss/engine.py
@@ -3,7 +3,7 @@ import logging
import re
from collections import defaultdict
from datetime import datetime, timezone
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
from urllib.parse import urlparse
from ab_sdk import points
@@ -40,6 +40,9 @@ from module.parser.title_parser import _apply_title_hooks
from module.plugin import host as plugin_host
from module.plugin.views import bangumi_info, torrent_info
+if TYPE_CHECKING:
+ from .analyser import RSSAnalyser
+
logger = logging.getLogger(__name__)
# Delay between consecutive requests to the same host. Firing all feeds of one
@@ -162,16 +165,12 @@ class RSSEngine:
msg_zh="删除 RSS 成功。",
)
- async def pull_rss(self, rss_item: RSSItem) -> list[Torrent]:
- torrents = await self._get_torrents(rss_item)
- new_torrents = await self.db.torrent.check_new(torrents)
- return new_torrents
-
async def _pull_rss_with_status(
self, rss_item: RSSItem
) -> tuple[list[Torrent], Optional[str]]:
+ """拉取订阅的全部条目(不去重),聚合分析与下载匹配共用这一份结果。"""
try:
- torrents = await self.pull_rss(rss_item)
+ torrents = await self._get_torrents(rss_item)
return torrents, None
except Exception as e:
logger.warning(f"Failed to fetch RSS {rss_item.name}: {e}")
@@ -279,12 +278,17 @@ class RSSEngine:
``refresh_rss`` 内部过滤,不做跨请求持久化。
"""
skip_ids: set[int] = set()
+ candidate_ids = {b.id for _, b in matched if b.id in preference_bangumi}
+ if not candidate_ids:
+ # 本批次没有偏好番剧的候选:无需解析已下载版本
+ return skip_ids
# 已下载版本:按完整内容身份记录当前最高 (偏好得分, 修订号)。
+ # 只解析本批次有候选的番剧。
existing_best: dict[_PreferenceKey, _PreferenceRank] = {}
for bangumi_id, torrents in existing_downloaded.items():
bangumi = preference_bangumi.get(bangumi_id)
- if not bangumi:
+ if not bangumi or bangumi_id not in candidate_ids:
continue
for torrent in torrents:
release = parse_configured_release_title(torrent.name)
@@ -335,17 +339,55 @@ class RSSEngine:
return skip_ids
async def refresh_rss(
- self, downloaders: DownloaderPool, rss_id: Optional[int] = None
+ self,
+ downloaders: DownloaderPool,
+ rss_id: Optional[int] = None,
+ analyser: "RSSAnalyser | None" = None,
) -> list[SystemEvent]:
"""Refresh feeds with one parser engine for the complete workflow.
匹配到的种子投递到规则的下载器实例,规则未指定时用订阅的,再退回默认实例。
+ 给出 ``analyser`` 时,先用同一次拉取的结果为聚合订阅生成新规则。
"""
with parser_engine_snapshot():
- return await self._refresh_rss(downloaders, rss_id)
+ return await self._refresh_rss(downloaders, rss_id, analyser)
+
+ async def _analyse_aggregate(
+ self,
+ analyser: "RSSAnalyser",
+ item_results: list[tuple[RSSItem, tuple[list[Torrent], Optional[str]]]],
+ bangumi_list: list[Bangumi],
+ ) -> list[Bangumi]:
+ """为聚合订阅生成新规则,返回之后用于匹配的番剧列表。"""
+ for rss_item, (torrents, error) in item_results:
+ if not rss_item.aggregate or error:
+ continue
+ try:
+ if await analyser.rss_to_data(rss_item, torrents, self, bangumi_list):
+ # 新规则要能被后续订阅与本轮下载匹配到
+ bangumi_list = await self.db.bangumi.search_all()
+ except Exception:
+ # 仅当该 RSS 确实已在遍历期间被 API 删除时才视为预期情况;
+ # 其余异常可能是真实 bug,需在 WARNING 级别带完整堆栈。
+ if await self.db.rss.search_id(rss_item.id) is None:
+ logger.debug(
+ "Skipping RSS id=%s, deleted during iteration",
+ rss_item.id,
+ )
+ else:
+ logger.warning(
+ "Error analysing RSS id=%s (%s)",
+ rss_item.id,
+ rss_item.url,
+ exc_info=True,
+ )
+ return bangumi_list
async def _refresh_rss(
- self, downloaders: DownloaderPool, rss_id: Optional[int] = None
+ self,
+ downloaders: DownloaderPool,
+ rss_id: Optional[int] = None,
+ analyser: "RSSAnalyser | None" = None,
) -> list[SystemEvent]:
# Get All RSS Items
if not rss_id:
@@ -384,6 +426,10 @@ class RSSEngine:
# torrent against it in memory (this is the job the old module-level
# TTL cache existed to do).
bangumi_list = await self.db.bangumi.search_all()
+ if analyser is not None:
+ bangumi_list = await self._analyse_aggregate(
+ analyser, item_results, bangumi_list
+ )
events: list[SystemEvent] = []
# Bangumi with a release-group/resolution preference need per-episode
@@ -393,13 +439,6 @@ class RSSEngine:
for b in bangumi_list
if b.id is not None and (b.preferred_group or b.preferred_resolution)
}
- existing_downloaded: dict[int, list[Torrent]] = {}
- if preference_bangumi:
- existing_downloaded = (
- await self.db.torrent.search_downloaded_by_bangumi_ids(
- list(preference_bangumi)
- )
- )
# First pass: match every torrent against the bangumi list (this also
# tags torrent.bangumi_id as a side effect) without downloading yet,
@@ -408,20 +447,38 @@ class RSSEngine:
item_matches: list[
tuple[RSSItem, list[Torrent], Optional[str], list[Optional[Bangumi]]]
] = []
- for rss_item, (new_torrents, error) in item_results:
+ for rss_item, (torrents, error) in item_results:
+ try:
+ new_torrents = await self.db.torrent.check_new(torrents)
+ except Exception as e:
+ # 查库出错只算该订阅失败,其余订阅照常处理
+ logger.warning(f"Failed to dedup RSS {rss_item.name}: {e}")
+ new_torrents, error = [], str(e)
matches = [self.match_torrent(t, bangumi_list) for t in new_torrents]
matches = await self._apply_plugin_filters(new_torrents, matches)
item_matches.append((rss_item, new_torrents, error, matches))
+ matched_pairs = [
+ (torrent, matched)
+ for _, new_torrents, _, matches in item_matches
+ for torrent, matched in zip(new_torrents, matches)
+ if matched is not None
+ ]
+ # 只为本批次有候选的偏好番剧加载已下载版本;没有候选时不查库
+ candidate_ids = {
+ b.id
+ for _, b in matched_pairs
+ if b.id is not None and b.id in preference_bangumi
+ }
+ existing_downloaded: dict[int, list[Torrent]] = {}
+ if candidate_ids:
+ existing_downloaded = (
+ await self.db.torrent.search_downloaded_by_bangumi_ids(
+ list(candidate_ids)
+ )
+ )
skip_ids = self._select_preference_skips(
- [
- (torrent, matched)
- for _, new_torrents, _, matches in item_matches
- for torrent, matched in zip(new_torrents, matches)
- if matched is not None
- ],
- preference_bangumi,
- existing_downloaded,
+ matched_pairs, preference_bangumi, existing_downloaded
)
# Process results sequentially (DB operations)
diff --git a/backend/src/test/test_api_events.py b/backend/src/test/test_api_events.py
index 3de84483..be9bb65c 100644
--- a/backend/src/test/test_api_events.py
+++ b/backend/src/test/test_api_events.py
@@ -2,21 +2,35 @@
import asyncio
import json
+import os
import time
+from itertools import chain, repeat
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
-from module.api import v1
-from module.api.events import _downloader_payload, _log_payload, _status_payload
+from module.api import events, v1
+from module.api.events import (
+ _downloader_payload,
+ _LogCursor,
+ _status_payload,
+ _TorrentCache,
+)
+from module.api.log import _clear_log_files
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
+@pytest.fixture(autouse=True)
+def fresh_torrent_cache(monkeypatch):
+ """种子列表缓存是模块级共享状态(含绑定事件循环的锁),每个用例换一份。"""
+ monkeypatch.setattr(events, "_torrents", _TorrentCache())
+
+
@pytest.fixture
def app():
"""Create a FastAPI app with v1 routes for testing."""
@@ -191,31 +205,154 @@ class TestEventGeneratorNonBlocking:
assert json.loads(downloader_event["data"]) is None
-class TestLogPayload:
- async def test_log_payload_missing_file_returns_none(self, tmp_path):
- """_log_payload returns None when the log file doesn't exist yet."""
- missing = tmp_path / "does-not-exist.log"
- with patch("module.api.events.LOG_PATH", missing):
- result = await _log_payload()
+class TestTorrentCache:
+ async def test_concurrent_streams_share_one_fetch_per_ttl(self, monkeypatch):
+ """N 个连接同时取种子列表,TTL 内只查询一次下载器;过期后再查一次。"""
+ fetch = AsyncMock(return_value=[{"hash": "abc"}])
+ monkeypatch.setattr(events, "_downloader_payload", fetch)
+ cache = _TorrentCache()
- assert result is None
+ results = await asyncio.gather(*(cache.get() for _ in range(10)))
+ await cache.get()
+ assert fetch.await_count == 1
+ assert set(results) == {json.dumps([{"hash": "abc"}])}
- async def test_log_payload_reads_tail(self, tmp_path):
- """_log_payload decodes the tail bytes read via _read_log_tail.
+ monkeypatch.setattr(events, "_DOWNLOADER_CACHE_SECONDS", 0)
+ await cache.get()
+ assert fetch.await_count == 2
- _read_log_tail (imported from module.api.log) resolves LOG_PATH via
- its own module binding, so both names must be patched together.
- """
+
+class TestLogCursor:
+ """log 帧:首帧整段尾部;之后只推新增完整行;轮转/截断后重推整段。"""
+
+ @pytest.fixture
+ def log_file(self, tmp_path):
+ """_read_log_tail 通过 module.api.log 自己的 LOG_PATH 绑定读文件,两处一起替换。"""
log_file = tmp_path / "app.log"
- log_file.write_text("hello world\n")
-
with (
patch("module.api.events.LOG_PATH", log_file),
patch("module.api.log.LOG_PATH", log_file),
):
- result = await _log_payload()
+ yield log_file
- assert result == "hello world\n"
+ async def test_missing_file_yields_no_frame(self, log_file):
+ assert await _LogCursor().frame() is None
+
+ async def test_first_frame_is_full_tail(self, log_file):
+ log_file.write_text("hello world\n")
+
+ frame = await _LogCursor().frame()
+
+ assert frame == {"event": "log", "data": "hello world\n"}
+
+ @pytest.mark.parametrize(
+ ("appended", "expected"),
+ [
+ ("second 第二行\n", "second 第二行\n"),
+ # 半行留到写完整再推,偏移只前进到最后一个换行
+ ("second\nthi", "second\n"),
+ ("", None),
+ ("thi", None),
+ ],
+ )
+ async def test_append_yields_only_new_complete_lines(
+ self, log_file, appended, expected
+ ):
+ log_file.write_text("hello world\n")
+ cursor = _LogCursor()
+ await cursor.frame()
+
+ with log_file.open("a") as f:
+ f.write(appended)
+ frame = await cursor.frame()
+
+ if expected is None:
+ assert frame is None
+ else:
+ assert frame == {"event": "log_append", "data": expected}
+
+ async def test_partial_line_is_sent_once_completed(self, log_file):
+ log_file.write_text("hello world\n")
+ cursor = _LogCursor()
+ await cursor.frame()
+
+ with log_file.open("a") as f:
+ f.write("thi")
+ assert await cursor.frame() is None
+ with log_file.open("a") as f:
+ f.write("rd\n")
+
+ assert await cursor.frame() == {"event": "log_append", "data": "third\n"}
+
+ async def test_empty_log_on_connect_yields_log_clear(self, log_file):
+ """清空后重连:空日志推 log_clear,前端才会丢掉上个连接的旧文本。"""
+ log_file.write_text("")
+ cursor = _LogCursor()
+
+ assert await cursor.frame() == {"event": "log_clear", "data": "1"}
+ assert await cursor.frame() is None
+
+ @pytest.mark.parametrize("clear", ["truncate", "api"])
+ async def test_cleared_log_yields_log_clear_then_appends(self, log_file, clear):
+ """截断或经 /log/clear 清空:推一次 log_clear,之后的新行按增量推。"""
+ log_file.write_text("old line\n")
+ cursor = _LogCursor()
+ await cursor.frame()
+
+ if clear == "truncate":
+ log_file.write_text("")
+ else:
+ _clear_log_files()
+ assert await cursor.frame() == {"event": "log_clear", "data": "1"}
+ assert await cursor.frame() is None
+ with log_file.open("a") as f:
+ f.write("fresh\n")
+
+ assert await cursor.frame() == {"event": "log_append", "data": "fresh\n"}
+
+ async def test_clear_then_regrow_past_offset_sends_full_frame(self, log_file):
+ """清空后、下一次读取前日志已长过旧偏移:仍推整段,而不是从旧偏移增量。"""
+ log_file.write_text("old line\n")
+ cursor = _LogCursor()
+ await cursor.frame()
+
+ _clear_log_files()
+ with log_file.open("a") as f:
+ f.write("fresh line after clear\n")
+
+ assert await cursor.frame() == {
+ "event": "log",
+ "data": "fresh line after clear\n",
+ }
+
+ @pytest.mark.parametrize(
+ ("change", "expected"),
+ [
+ ("truncate", "new\n"),
+ # 轮转后 log.txt 近乎为空,整段尾部会拼上 .1 备份
+ ("rotate", "old line one\nold line two\nnew\n"),
+ # 同 inode、同大小,但 mtime 回退(被整体替换写回)
+ ("mtime_reset", "NEW line one\nNEW line two\n"),
+ ],
+ )
+ async def test_rotation_or_truncation_yields_full_frame(
+ self, log_file, change, expected
+ ):
+ log_file.write_text("old line one\nold line two\n")
+ cursor = _LogCursor()
+ await cursor.frame()
+
+ if change == "truncate":
+ log_file.write_text("new\n")
+ elif change == "rotate":
+ log_file.rename(log_file.with_name(f"{log_file.name}.1"))
+ log_file.write_text("new\n")
+ else:
+ st = log_file.stat()
+ log_file.write_text("NEW line one\nNEW line two\n")
+ os.utime(log_file, ns=(st.st_atime_ns, st.st_mtime_ns - 10**9))
+
+ assert await cursor.frame() == {"event": "log", "data": expected}
# ---------------------------------------------------------------------------
@@ -235,7 +372,7 @@ class TestNotificationEvent:
monkeypatch.setattr(host, "_bus", bus)
return bus
- def _generator(self, tmp_path, *, tick=0.01, payload=None):
+ def _generator(self, tmp_path, *, tick=0.01, payload=None, downloader=None):
from module.api.events import _event_generator
request = AsyncMock()
@@ -249,7 +386,7 @@ class TestNotificationEvent:
patches = (
patch(
"module.api.events._downloader_payload",
- new=AsyncMock(return_value=None),
+ new=downloader or AsyncMock(return_value=None),
),
patch("module.api.events.LOG_PATH", tmp_path / "missing.log"),
patch("module.api.events._notification_payload", new=notification),
@@ -283,12 +420,41 @@ class TestNotificationEvent:
with patches[0], patches[1], patches[2], patches[3]:
try:
await self._collect(gen, 3) # 第一个 tick
- # tick3 status、tick5 downloader —— 中间不应再有 notification
+ # tick3 status、tick6 status(tick5 的 downloader 负载未变,不推)
+ # —— 中间不应再有 notification
more = await self._collect(gen, 2)
finally:
await gen.aclose()
- assert [e["event"] for e in more] == ["status", "downloader"]
+ assert [e["event"] for e in more] == ["status", "status"]
+
+ @pytest.mark.parametrize(
+ ("second", "expected"),
+ [
+ (
+ [{"hash": "a"}],
+ ["status", "downloader", "notification", "status", "status"],
+ ),
+ (
+ [{"hash": "b"}],
+ ["status", "downloader", "notification", "status", "downloader"],
+ ),
+ ],
+ )
+ async def test_downloader_frame_only_when_payload_changes(
+ self, tmp_path, bus, monkeypatch, second, expected
+ ):
+ fetch = AsyncMock(side_effect=chain([[{"hash": "a"}]], repeat(second)))
+ monkeypatch.setattr(events, "_DOWNLOADER_CACHE_SECONDS", 0)
+ gen, patches = self._generator(tmp_path, downloader=fetch)
+ with patches[0], patches[1], patches[2], patches[3]:
+ try:
+ got = await self._collect(gen, 5) # tick0 … tick5
+ finally:
+ await gen.aclose()
+
+ assert [e["event"] for e in got] == expected
+ assert fetch.await_count == 2
async def test_inbox_change_pushes_frame_immediately(self, tmp_path, bus):
from module.notification.inbox import bump_inbox_revision, inbox_revision
diff --git a/backend/src/test/test_api_log.py b/backend/src/test/test_api_log.py
index f04848dc..82eff90c 100644
--- a/backend/src/test/test_api_log.py
+++ b/backend/src/test/test_api_log.py
@@ -149,7 +149,7 @@ class TestGetLog:
from module.api.log import _read_file_tail
missing = temp_log_file.with_name("gone.txt")
- assert _read_file_tail(missing, 100) == (b"", False)
+ assert _read_file_tail(missing, 100) == (b"", False, 0)
# ---------------------------------------------------------------------------
diff --git a/backend/src/test/test_conf_log.py b/backend/src/test/test_conf_log.py
index 72607c83..1494f914 100644
--- a/backend/src/test/test_conf_log.py
+++ b/backend/src/test/test_conf_log.py
@@ -1,5 +1,8 @@
"""启动时日志轮转(rotate_boot_log):历史进备份而非被删除。"""
+import os
+import subprocess
+import sys
from pathlib import Path
import pytest
@@ -66,3 +69,21 @@ class TestRotateBootLog:
conf_log.rotate_boot_log()
assert log_path.read_text() == "content"
+
+
+def test_setup_logger_replaces_handlers_installed_by_earlier_imports(tmp_path: Path):
+ """生产导入顺序里 mcp 早于 setup_logger 导入;根日志器已有处理器时
+ basicConfig 不能被静默跳过,日志必须落到 data/log.txt。"""
+ code = (
+ "import logging, main\n"
+ "assert [type(h).__name__ for h in logging.getLogger().handlers]"
+ " == ['QueueHandler']\n"
+ "logging.getLogger('probe').info('probe-marker')\n"
+ )
+ src = Path(__file__).resolve().parents[1]
+ env = {**os.environ, "PYTHONPATH": str(src)}
+ subprocess.run(
+ [sys.executable, "-c", code], cwd=tmp_path, env=env, check=True, timeout=60
+ )
+ # 子进程退出时 atexit 会排空日志队列
+ assert "probe-marker" in (tmp_path / "data" / "log.txt").read_text()
diff --git a/backend/src/test/test_database_rename_operation.py b/backend/src/test/test_database_rename_operation.py
index ffcf8864..32c59b1c 100644
--- a/backend/src/test/test_database_rename_operation.py
+++ b/backend/src/test/test_database_rename_operation.py
@@ -4,6 +4,7 @@ from datetime import datetime, timedelta, timezone
from typing import Any
import pytest
+from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from module.database import Database
@@ -268,3 +269,28 @@ async def test_database_facade_exposes_rename_operation_repository(db_engine):
row, created = await db.rename_operation.get_or_create(_operation())
assert created is True
assert await db.rename_operation.get(row.id) is not None
+
+
+async def test_get_by_target_active_filter_uses_partial_unique_index(db_session):
+ """state 过滤必须写成 NOT IN ('done'),才能命中同样条件的部分唯一索引。"""
+ repo = RenameOperationDatabase(db_session)
+ captured: list[Any] = []
+ original = db_session.execute
+
+ async def spy(statement, *args, **kwargs):
+ captured.append(statement)
+ return await original(statement, *args, **kwargs)
+
+ db_session.execute = spy # type: ignore[method-assign]
+ await repo.get_by_target(
+ downloader_type="qbittorrent", save_path="/d", target_path="t.mp4"
+ )
+ db_session.execute = original # type: ignore[method-assign]
+
+ plan = await db_session.execute(
+ text(
+ "EXPLAIN QUERY PLAN "
+ + str(captured[0].compile(compile_kwargs={"literal_binds": True}))
+ )
+ )
+ assert "ux_rename_operation_active_target" in " ".join(r[3] for r in plan)
diff --git a/backend/src/test/test_http_perf.py b/backend/src/test/test_http_perf.py
new file mode 100644
index 00000000..b92dac31
--- /dev/null
+++ b/backend/src/test/test_http_perf.py
@@ -0,0 +1,107 @@
+"""HTTP 层性能头:gzip 压缩与 SPA 静态文件的缓存策略。"""
+
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+import main
+from module.api.deps import get_context
+from module.security.api import get_current_user
+
+HTML = {"accept": "text/html"}
+GZIP = {"accept-encoding": "gzip"}
+IMMUTABLE = "public, max-age=31536000, immutable"
+
+
+@pytest.fixture
+def app(tmp_path):
+ dist = tmp_path / "dist"
+ for sub in ("assets", "fonts", "images"):
+ (dist / sub).mkdir(parents=True)
+ (dist / "index.html").write_text("spa")
+ (dist / "theme-init.js").write_text("// theme")
+ (dist / "assets" / "app-abc.js").write_text("export const x = 1;\n" * 200)
+ (dist / "fonts" / "a.woff2").write_text("font")
+ (dist / "images" / "a.svg").write_text("")
+ app = main.create_app()
+ main.mount_webui(app, dist)
+ return app
+
+
+class TestSpaCacheControl:
+ @pytest.mark.parametrize("root_path", ["", "/ab"])
+ @pytest.mark.parametrize(
+ ("path", "cache_control"),
+ [
+ ("assets/app-abc.js", IMMUTABLE),
+ ("", "no-cache"),
+ ("index.html", "no-cache"),
+ # 前端路由深链接回落到 index.html
+ ("bangumi", "no-cache"),
+ ("bangumi/detail/3", "no-cache"),
+ # assets/ 下不存在的路径同样回落到 index.html,不能永久缓存
+ ("assets/missing", "no-cache"),
+ ("fonts/a.woff2", None),
+ ("images/a.svg", None),
+ ("theme-init.js", None),
+ ],
+ )
+ def test_cache_control_by_path(self, app, root_path, path, cache_control):
+ resp = TestClient(app, root_path=root_path).get(
+ f"{root_path}/{path}", headers=HTML
+ )
+ assert resp.status_code == 200
+ assert resp.headers.get("cache-control") == cache_control
+
+
+class TestGzip:
+ def test_large_static_asset_is_gzipped(self, app):
+ resp = TestClient(app).get("/assets/app-abc.js", headers=GZIP)
+ assert resp.headers["content-encoding"] == "gzip"
+
+ def test_small_response_is_not_gzipped(self, app):
+ resp = TestClient(app).get("/", headers={**HTML, **GZIP})
+ assert "content-encoding" not in resp.headers
+
+ def test_event_stream_is_not_gzipped(self, app):
+ app.dependency_overrides[get_current_user] = lambda: "user"
+ app.dependency_overrides[get_context] = lambda: MagicMock(
+ is_running=True, first_run=False
+ )
+ started: dict = {}
+
+ async def drive() -> None:
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/api/v1/events/stream",
+ "raw_path": b"/api/v1/events/stream",
+ "query_string": b"",
+ "headers": [(b"accept-encoding", b"gzip")],
+ "scheme": "http",
+ "server": ("test", 80),
+ "client": ("test", 1),
+ "http_version": "1.1",
+ "root_path": "",
+ }
+
+ async def receive() -> dict:
+ await asyncio.Event().wait() # 客户端保持连接,不发 disconnect
+ return {}
+
+ async def send(message: dict) -> None:
+ if message["type"] == "http.response.start":
+ started.update(message)
+ raise asyncio.CancelledError # 只取响应头,随后断开
+
+ try:
+ await app(scope, receive, send)
+ except asyncio.CancelledError:
+ pass
+
+ asyncio.run(drive())
+ headers = {k.decode(): v.decode() for k, v in started["headers"]}
+ assert headers["content-type"].startswith("text/event-stream")
+ assert "content-encoding" not in headers
diff --git a/backend/src/test/test_loops.py b/backend/src/test/test_loops.py
index 54cd1604..9f55b240 100644
--- a/backend/src/test/test_loops.py
+++ b/backend/src/test/test_loops.py
@@ -10,7 +10,7 @@ returned events into ``notifier.send_event`` calls, gated by
import asyncio
import time
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
from module.core.loops import (
offset_scan_tick,
@@ -93,7 +93,6 @@ class TestRssTick:
analyser = AsyncMock()
mock_db = AsyncMock()
- mock_db.rss.search_aggregate = AsyncMock(return_value=[])
mock_engine = AsyncMock()
mock_engine.refresh_rss = AsyncMock(return_value=events)
@@ -110,6 +109,7 @@ class TestRssTick:
mock_settings.bangumi_manage.eps_complete = False
await rss_tick(analyser, notifier)
+ mock_engine.refresh_rss.assert_awaited_once_with(ANY, analyser=analyser)
notifier.send_event.assert_awaited_once_with(events[0])
async def test_events_forwarded_even_when_external_push_disabled(self):
@@ -119,7 +119,6 @@ class TestRssTick:
analyser = AsyncMock()
mock_db = AsyncMock()
- mock_db.rss.search_aggregate = AsyncMock(return_value=[])
mock_engine = AsyncMock()
mock_engine.refresh_rss = AsyncMock(return_value=events)
diff --git a/backend/src/test/test_main_lifespan.py b/backend/src/test/test_main_lifespan.py
index 41b17ae5..359fcda1 100644
--- a/backend/src/test/test_main_lifespan.py
+++ b/backend/src/test/test_main_lifespan.py
@@ -1,5 +1,10 @@
"""Tests for the application lifespan wiring in main.create_app."""
+import asyncio
+import os
+import subprocess
+import sys
+from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -21,6 +26,22 @@ def _lifespan_ctx(first_run_boot: bool = False) -> MagicMock:
class TestCreateApp:
+ def test_import_main_skips_mcp_and_webauthn(self):
+ """启动不加载 mcp 与 webauthn 库(其它测试已导入它们,须在子进程里检查)。"""
+ src = Path(main.__file__).parent
+ code = (
+ "import sys, main; print(sorted({'mcp', 'webauthn'} & sys.modules.keys()))"
+ )
+ out = subprocess.run(
+ [sys.executable, "-c", code],
+ cwd=src.parent,
+ env={**os.environ, "PYTHONPATH": str(src)},
+ capture_output=True,
+ text=True,
+ )
+ assert out.returncode == 0, out.stderr
+ assert out.stdout.strip().splitlines()[-1] == "[]"
+
def test_stores_real_context_on_state(self):
app = main.create_app()
assert isinstance(app.state.ctx, AppContext)
@@ -98,6 +119,10 @@ class TestMcpEndpoint:
},
)
assert init.status_code == 200
+ from module.mcp import mcp
+
+ manager = mcp.session_manager
+ assert manager._task_group is not None
session = {"mcp-session-id": init.headers["mcp-session-id"]}
client.post(
"/mcp",
@@ -113,6 +138,26 @@ class TestMcpEndpoint:
assert '"list_anime"' in listed.text
# 旧的 SSE 端点已删除
assert client.get("/mcp/sse", headers=self.HEADERS).status_code == 404
+ # 关停时会话管理器随 lifespan 的任务组停止
+ assert manager._task_group is None
+
+ async def test_concurrent_first_requests_build_once(self):
+ lazy = main._LazyMcpApp(MagicMock())
+ inner = AsyncMock()
+ with (
+ patch("module.mcp.create_mcp_app", return_value=inner) as build,
+ patch("module.mcp.mcp") as server,
+ ):
+ async with lazy.running():
+ await asyncio.gather(
+ *(
+ lazy({"type": "http"}, AsyncMock(), AsyncMock())
+ for _ in range(5)
+ )
+ )
+ build.assert_called_once()
+ server.session_manager.run.assert_called_once()
+ assert inner.await_count == 5
def test_rejected_without_token_or_whitelisted_ip(self):
with patch("main.AppContext.build", return_value=_lifespan_ctx()):
diff --git a/backend/src/test/test_rss_analyser_typed_identity.py b/backend/src/test/test_rss_analyser_typed_identity.py
index 7c6db5c8..23933f0a 100644
--- a/backend/src/test/test_rss_analyser_typed_identity.py
+++ b/backend/src/test/test_rss_analyser_typed_identity.py
@@ -8,7 +8,7 @@ from unittest.mock import patch
import pytest
from module.conf import settings
-from module.models import Movie, Torrent
+from module.models import Movie, RSSItem, Torrent
from module.models.config import LLM
from module.parser.analyser import selector
from module.parser.analyser.selector import parse_configured_release_title
@@ -97,8 +97,8 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits(
torrent = Torrent(name="resource", url="https://example.com/resource")
analyser = RSSAnalyser()
- async def get_rss_torrents(rss_link: str, full_parse: bool = True):
- return [torrent]
+ async def check_new(torrents, rss: RSSItem | None = None):
+ return torrents
async def match_movies(torrents, rss_link: str):
assert parse_configured_release_title(torrent.name) is classic_release
@@ -107,7 +107,7 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits(
assert parse_configured_release_title(torrent.name) is classic_release
return torrents
- async def match_bangumi(torrents, rss_link: str):
+ async def match_bangumi(torrents, rss_link: str, match_datas):
assert parse_configured_release_title(torrent.name) is classic_release
return torrents
@@ -115,19 +115,19 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits(
assert parse_configured_release_title(torrent.name) is classic_release
return [], []
- monkeypatch.setattr(analyser, "get_rss_torrents", get_rss_torrents)
monkeypatch.setattr(analyser, "torrents_to_data", torrents_to_data)
engine = cast(
RSSEngine,
SimpleNamespace(
db=SimpleNamespace(
+ torrent=SimpleNamespace(check_new=check_new),
movie=SimpleNamespace(match_list=match_movies),
bangumi=SimpleNamespace(match_list=match_bangumi),
)
),
)
- await analyser.rss_to_data(make_rss_item(parser="none"), engine)
+ await analyser.rss_to_data(make_rss_item(parser="none"), [torrent], engine, [])
# The task-local scope is gone, so a later workflow sees the new setting.
assert parse_configured_release_title(torrent.name) is preview_release
diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py
index 1576a7f9..8409a711 100644
--- a/backend/src/test/test_rss_engine_new.py
+++ b/backend/src/test/test_rss_engine_new.py
@@ -1,18 +1,24 @@
-"""Tests for RSS engine: pull_rss, match_torrent, refresh_rss, add_rss."""
+"""Tests for RSS engine: aggregate analysis, match_torrent, refresh_rss, add_rss."""
import asyncio
from unittest.mock import AsyncMock, patch
import pytest
import pytest_asyncio
+from sqlalchemy.exc import OperationalError
from module.conf import settings
from module.database import Database
+from module.database.bangumi import BangumiDatabase
+from module.database.movie import MovieDatabase
+from module.database.torrent import TorrentDatabase
from module.downloader import AddResult
from module.models import Torrent
+from module.network import RequestContent
from module.notification.events import DownloadFailureEvent, RssFailureEvent
from module.parser.analyser.selector import parse_configured_release_title
from module.parser.analyser.tokenizer import ReleaseKind
+from module.rss.analyser import RSSAnalyser
from module.rss.engine import RSSEngine
from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent
@@ -24,68 +30,135 @@ async def rss_engine(db_engine):
# ---------------------------------------------------------------------------
-# pull_rss
+# 聚合订阅分析(refresh_rss 带 analyser,即 rss_tick)
# ---------------------------------------------------------------------------
+FEED_A = "https://mikanani.me/RSS/MyBangumi?token=a"
+FEED_B = "https://mikanani.me/RSS/MyBangumi?token=b"
+KNOWN_URL = "https://example.com/known.torrent"
-class TestPullRss:
- async def test_returns_only_new_torrents(self, rss_engine):
- """pull_rss filters out torrents already in the database."""
- rss_item = make_rss_item()
- await rss_engine.db.rss.add(rss_item)
- rss_item = await rss_engine.db.rss.search_id(1)
- # Pre-insert one torrent into DB
- existing = make_torrent(url="https://example.com/existing.torrent", rss_id=1)
- await rss_engine.db.torrent.add(existing)
-
- # Mock _get_torrents to return 3 torrents (1 existing + 2 new)
- all_torrents = [
- Torrent(name="existing", url="https://example.com/existing.torrent"),
- Torrent(name="new1", url="https://example.com/new1.torrent"),
- Torrent(name="new2", url="https://example.com/new2.torrent"),
+async def _tick_with_feeds(
+ engine: RSSEngine, items: dict[str, list[Torrent]]
+) -> AsyncMock:
+ """按订阅 URL 返回 items 跑一轮刷新,返回 HTTP 层 get_torrents 的替身。"""
+ client = AsyncMock()
+ client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
+ get = AsyncMock(
+ side_effect=lambda url, *_a, **_k: [
+ Torrent(name=t.name, url=t.url) for t in items[url]
]
- with patch.object(
- RSSEngine, "_get_torrents", new_callable=AsyncMock
- ) as mock_get:
- mock_get.return_value = all_torrents
- result = await rss_engine.pull_rss(rss_item)
+ )
+ with (
+ patch.object(RequestContent, "get_torrents", get),
+ patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
+ patch("module.rss.analyser.enrich_bangumi", AsyncMock()),
+ patch("module.rss.analyser.enrich_movie", AsyncMock()),
+ ):
+ await engine.refresh_rss(SingleClientPool(client), analyser=RSSAnalyser())
+ return get
- assert len(result) == 2
- assert all(t.url != "https://example.com/existing.torrent" for t in result)
- async def test_all_existing_returns_empty(self, rss_engine):
- """When all torrents already exist, returns empty list."""
- rss_item = make_rss_item()
- await rss_engine.db.rss.add(rss_item)
- rss_item = await rss_engine.db.rss.search_id(1)
+class TestAggregateAnalysis:
+ async def test_item_matched_for_same_feed_skips_matching(self, rss_engine):
+ """本订阅已匹配入库、番剧 rss_link 已含本订阅 URL 的种子不再匹配。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_A))
+ await rss_engine.db.bangumi.add(
+ make_bangumi(title_raw="Dungeon Meshi", filter="", rss_link=FEED_A)
+ )
+ await rss_engine.db.torrent.add(
+ make_torrent(
+ name="[Sub] Dungeon Meshi - 01 [1080p].mkv",
+ url=KNOWN_URL,
+ rss_id=1,
+ bangumi_id=1,
+ )
+ )
+ known = Torrent(name="[Sub] Dungeon Meshi - 01 [1080p].mkv", url=KNOWN_URL)
- existing = make_torrent(url="https://example.com/only.torrent", rss_id=1)
- await rss_engine.db.torrent.add(existing)
+ with (
+ patch.object(
+ BangumiDatabase, "match_list", new_callable=AsyncMock
+ ) as bangumi_match,
+ patch.object(
+ MovieDatabase, "match_list", new_callable=AsyncMock
+ ) as movie_match,
+ ):
+ await _tick_with_feeds(rss_engine, {FEED_A: [known]})
- with patch.object(
- RSSEngine, "_get_torrents", new_callable=AsyncMock
- ) as mock_get:
- mock_get.return_value = [
- Torrent(name="only", url="https://example.com/only.torrent")
- ]
- result = await rss_engine.pull_rss(rss_item)
+ bangumi_match.assert_not_awaited()
+ movie_match.assert_not_awaited()
+ assert len(await rss_engine.db.bangumi.search_all()) == 1
- assert result == []
+ async def test_edited_feed_url_appends_to_rss_link(self, rss_engine):
+ """订阅改了 URL(rss_id 不变):本订阅已匹配的种子仍要匹配,
+ 把新 URL 追加到 rss_link,补下载才不会用过期链接(#1053)。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_B))
+ await rss_engine.db.bangumi.add(
+ make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_A)
+ )
+ name = "[Sub] Mushoku Tensei - 12 [1080p].mkv"
+ await rss_engine.db.torrent.add(
+ make_torrent(name=name, url=KNOWN_URL, rss_id=1, bangumi_id=1)
+ )
- async def test_empty_feed_returns_empty(self, rss_engine):
- """When RSS feed has no torrents, returns empty list."""
- rss_item = make_rss_item()
- await rss_engine.db.rss.add(rss_item)
- rss_item = await rss_engine.db.rss.search_id(1)
+ await _tick_with_feeds(
+ rss_engine, {FEED_B: [Torrent(name=name, url=KNOWN_URL)]}
+ )
- with patch.object(
- RSSEngine, "_get_torrents", new_callable=AsyncMock
- ) as mock_get:
- mock_get.return_value = []
- result = await rss_engine.pull_rss(rss_item)
+ bangumi = await rss_engine.db.bangumi.search_id(1)
+ assert bangumi is not None
+ assert bangumi.rss_link.split(",") == [FEED_A, FEED_B]
- assert result == []
+ async def test_orphan_for_same_feed_is_analysed_again(self, rss_engine):
+ """本订阅记录的未匹配种子(孤儿)每轮仍参与分析,新番能生成规则。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_A))
+ name = "[Sub] Dungeon Meshi - 01 [1080p].mkv"
+ await rss_engine.db.torrent.add(
+ make_torrent(name=name, url=KNOWN_URL, rss_id=1)
+ )
+
+ await _tick_with_feeds(
+ rss_engine, {FEED_A: [Torrent(name=name, url=KNOWN_URL)]}
+ )
+
+ rules = await rss_engine.db.bangumi.search_all()
+ assert [b.title_raw for b in rules] == ["Dungeon Meshi"]
+
+ async def test_known_torrent_in_second_feed_appends_rss_link(self, rss_engine):
+ """其他订阅已知的种子出现在新聚合订阅里:仍把新订阅追加到 rss_link(#1053)。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_A))
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_B))
+ await rss_engine.db.bangumi.add(
+ make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_A)
+ )
+ name = "[Sub] Mushoku Tensei - 12 [1080p].mkv"
+ await rss_engine.db.torrent.add(
+ make_torrent(name=name, url=KNOWN_URL, rss_id=1, bangumi_id=1)
+ )
+ known = Torrent(name=name, url=KNOWN_URL)
+
+ await _tick_with_feeds(rss_engine, {FEED_A: [known], FEED_B: [known]})
+
+ bangumi = await rss_engine.db.bangumi.search_id(1)
+ assert bangumi is not None
+ assert bangumi.rss_link.split(",") == [FEED_A, FEED_B]
+
+ async def test_each_feed_fetched_once_per_tick(self, rss_engine):
+ """聚合订阅不再单独请求:每轮每个订阅只有一次 HTTP GET。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_A))
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_B, aggregate=False))
+ await rss_engine.db.bangumi.add(
+ make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_A)
+ )
+ items = {
+ FEED_A: [Torrent(name="[Sub] Mushoku Tensei - 12 [1080p].mkv", url="a")],
+ FEED_B: [Torrent(name="[Sub] Mushoku Tensei - 13 [1080p].mkv", url="b")],
+ }
+
+ get = await _tick_with_feeds(rss_engine, items)
+
+ assert sorted(c.args[0] for c in get.await_args_list) == [FEED_A, FEED_B]
# ---------------------------------------------------------------------------
@@ -884,6 +957,40 @@ class TestRefreshRssPerHostThrottle:
item = await rss_engine.db.rss.search_id(rss_id)
assert item.connection_status == "healthy"
+ async def test_dedup_error_marks_only_that_feed_failed(self, rss_engine):
+ """某订阅去重查库出错只把该订阅标为失败,其余订阅照常下载。"""
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_A, aggregate=False))
+ await rss_engine.db.rss.add(make_rss_item(url=FEED_B, aggregate=False))
+ await rss_engine.db.bangumi.add(
+ make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_B)
+ )
+ items = {
+ FEED_A: [Torrent(name="[Sub] Mushoku Tensei - 12 [1080p].mkv", url="a")],
+ FEED_B: [Torrent(name="[Sub] Mushoku Tensei - 13 [1080p].mkv", url="b")],
+ }
+ check_new = TorrentDatabase.check_new
+
+ async def flaky_check_new(self, torrents, rss=None):
+ if any(t.url == "a" for t in torrents):
+ raise OperationalError("SELECT", {}, Exception("disk I/O error"))
+ return await check_new(self, torrents, rss)
+
+ client = AsyncMock()
+ client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
+ get = AsyncMock(side_effect=lambda url, *_a, **_k: items[url])
+ with (
+ patch.object(RequestContent, "get_torrents", get),
+ patch.object(TorrentDatabase, "check_new", flaky_check_new),
+ patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
+ ):
+ await rss_engine.refresh_rss(SingleClientPool(client))
+
+ feed_a = await rss_engine.db.rss.search_id(1)
+ feed_b = await rss_engine.db.rss.search_id(2)
+ assert feed_a.connection_status == "error"
+ assert feed_b.connection_status == "healthy"
+ assert [c.args[0].url for c in client.add_torrent.await_args_list] == ["b"]
+
# ---------------------------------------------------------------------------
# Release-group / resolution preference dedup
@@ -1403,3 +1510,59 @@ class TestRefreshRssPreferenceDedup:
await rss_engine.refresh_rss(SingleClientPool(client))
assert client.add_torrent.call_count == 2
+
+
+class TestPreferencePreload:
+ @pytest.fixture(autouse=True)
+ def use_preview_engine(self):
+ with patch.object(settings.rss_parser, "engine", "tokenizer"):
+ yield
+
+ @pytest.mark.parametrize(
+ ("name", "expected_ids"),
+ [
+ ("[GroupA] Dungeon Meshi - 01 [1080p].mkv", None),
+ ("[GroupA] Mushoku Tensei - 12 [1080p].mkv", [1]),
+ ],
+ ids=["no-preference-candidate", "one-preference-candidate"],
+ )
+ async def test_downloaded_loaded_only_for_candidate_bangumi(
+ self, rss_engine, name, expected_ids
+ ):
+ """只为本批次有候选的偏好番剧加载已下载版本;没有候选时不查库。"""
+ await rss_engine.db.rss.add(make_rss_item(aggregate=False))
+ for title_raw, preferred_group in (
+ ("Mushoku Tensei", "GroupA"),
+ ("Dungeon Meshi", None),
+ ("Sousou no Frieren", "GroupA"),
+ ):
+ await rss_engine.db.bangumi.add(
+ make_bangumi(
+ official_title=title_raw,
+ title_raw=title_raw,
+ filter="",
+ preferred_group=preferred_group,
+ )
+ )
+ client = AsyncMock()
+ client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
+
+ with (
+ patch.object(
+ RSSEngine,
+ "_get_torrents",
+ AsyncMock(return_value=[Torrent(name=name, url="https://x/1")]),
+ ),
+ patch.object(
+ rss_engine.db.torrent,
+ "search_downloaded_by_bangumi_ids",
+ AsyncMock(return_value={}),
+ ) as preload,
+ ):
+ await rss_engine.refresh_rss(SingleClientPool(client))
+
+ client.add_torrent.assert_awaited_once()
+ if expected_ids is None:
+ preload.assert_not_awaited()
+ else:
+ preload.assert_awaited_once_with(expected_ids)
diff --git a/backend/src/test/test_tmdb.py b/backend/src/test/test_tmdb.py
index 45dbbcb1..3bf6dfdb 100644
--- a/backend/src/test/test_tmdb.py
+++ b/backend/src/test/test_tmdb.py
@@ -318,3 +318,85 @@ async def test_tmdb_parser_live():
assert tmdb_info.title == "冰海战记"
assert tmdb_info.year == bangumi_year
assert tmdb_info.last_season == bangumi_season
+
+
+async def test_tmdb_parser_negative_result_is_cached_until_ttl_expires(mocker):
+ """查不到的标题在 TTL 内只请求一次,过期后重新请求。"""
+ requests: list[str] = []
+
+ async def fake_get_json(url: str) -> dict:
+ requests.append(url)
+ return {"results": []}
+
+ mocker.patch.object(
+ tmdb_parser_module.RequestContent, "get_json", side_effect=fake_get_json
+ )
+ now = mocker.patch.object(tmdb_parser_module.time, "monotonic", return_value=0.0)
+ tmdb_parser_module._tmdb_cache.clear()
+
+ assert await tmdb_parser("不存在", "zh", test=True) is None
+ first = len(requests)
+ assert await tmdb_parser("不存在", "zh", test=True) is None
+ assert len(requests) == first
+
+ now.return_value = tmdb_parser_module._TMDB_NEGATIVE_TTL + 1
+ assert await tmdb_parser("不存在", "zh", test=True) is None
+ assert len(requests) == 2 * first
+
+
+async def test_tmdb_parser_cache_hit_refreshes_lru_order(mocker):
+ """命中的条目移到队尾,满员淘汰的是最久未用而非最早写入的条目。"""
+ mocker.patch.object(tmdb_parser_module, "_TMDB_CACHE_MAX", 2)
+ mocker.patch.object(
+ tmdb_parser_module.RequestContent, "get_json", side_effect=_fake_get_json
+ )
+ tmdb_parser_module._tmdb_cache.clear()
+
+ await tmdb_parser("A", "zh", test=True)
+ await tmdb_parser("B", "zh", test=True)
+ await tmdb_parser("A", "zh", test=True) # 命中,A 变为最近使用
+ await tmdb_parser("C", "zh", test=True) # 淘汰 B
+
+ assert list(tmdb_parser_module._tmdb_cache) == [
+ "A:zh:True:False",
+ "C:zh:True:False",
+ ]
+
+
+async def test_tmdb_parser_request_failure_is_not_cached(mocker):
+ """请求失败(get_json 返回 None)不当作「无匹配」缓存,下次照常请求。"""
+ requests: list[str] = []
+
+ async def fake_get_json(url: str) -> dict | None:
+ requests.append(url)
+ return None
+
+ mocker.patch.object(
+ tmdb_parser_module.RequestContent, "get_json", side_effect=fake_get_json
+ )
+ tmdb_parser_module._tmdb_cache.clear()
+
+ assert await tmdb_parser("海盗战记", "zh", test=True) is None
+ assert tmdb_parser_module._tmdb_cache == {}
+ first = len(requests)
+ assert await tmdb_parser("海盗战记", "zh", test=True) is None
+ assert len(requests) == 2 * first
+
+
+async def test_tmdb_parser_failed_animation_check_does_not_cache_movie(mocker):
+ """判断动画的请求失败时不回退成电影,也不缓存。"""
+
+ async def fake_get_json(url: str) -> dict | None:
+ if "/search/tv" in url:
+ return {"results": [{"id": 82684}]}
+ if "/search/movie" in url:
+ return _MOVIE_SEARCH_RESULT
+ return None
+
+ mocker.patch.object(
+ tmdb_parser_module.RequestContent, "get_json", side_effect=fake_get_json
+ )
+ tmdb_parser_module._tmdb_cache.clear()
+
+ assert await tmdb_parser("海盗战记", "zh", test=True) is None
+ assert tmdb_parser_module._tmdb_cache == {}
diff --git a/backend/vulture_whitelist.py b/backend/vulture_whitelist.py
index 266ebecb..f3a36888 100644
--- a/backend/vulture_whitelist.py
+++ b/backend/vulture_whitelist.py
@@ -42,6 +42,9 @@ exc_value # __aexit__
traceback # __aexit__
hash_ # for 循环解包
+# --- 模块级 __getattr__(PEP 562):解释器在属性缺失时调用 ---
+__getattr__
+
# --- 测试 / e2e 支撑:生产路径不调用 ---
_.check_single # e2e worker
_.set_renamed_path # aria2 测试构造状态
diff --git a/webui/.storybook/main.ts b/webui/.storybook/main.ts
index c83ea5a6..f48fd73b 100644
--- a/webui/.storybook/main.ts
+++ b/webui/.storybook/main.ts
@@ -10,6 +10,13 @@ const config: StorybookConfig = {
},
viteFinal(config) {
config.plugins?.push(Unocss());
+ // 故事用字符串 template 选项,需要完整版(含编译器)的 vue
+ const alias = config.resolve?.alias;
+ if (Array.isArray(alias)) {
+ config.resolve!.alias = alias.filter(
+ (a) => !(a.find instanceof RegExp && a.find.source === '^vue$')
+ );
+ }
// Add other configuration here depending on your use case
return config;
},
diff --git a/webui/src/components/ab-bangumi-card.vue b/webui/src/components/ab-bangumi-card.vue
index eefca765..2dc6e979 100644
--- a/webui/src/components/ab-bangumi-card.vue
+++ b/webui/src/components/ab-bangumi-card.vue
@@ -200,18 +200,40 @@ const posterSrc = computed(() => resolvePosterUrl(bangumi.poster_link));
display: block;
}
-// Card glow animation when needs review - yellow
+// 待审核的黄色光晕:静态 box-shadow 放在 ::after 上,只动画 opacity(合成器线程,
+// 不触发每帧重绘)。光晕在海报外侧,所以这里取消 overflow 裁剪,由子元素继承圆角
.card-poster--needs-review {
- animation: card-glow 2.5s ease-in-out infinite;
+ overflow: visible;
+
+ > * {
+ border-radius: inherit;
+ }
+
+ &::after {
+ content: '';
+ position: absolute;
+ inset: 0;
+ border-radius: inherit;
+ box-shadow: 0 0 16px 4px rgba(251, 191, 36, 0.6);
+ pointer-events: none;
+ animation: card-glow 2.5s ease-in-out infinite;
+ }
+
+ @media (prefers-reduced-motion: reduce) {
+ &::after {
+ animation: none;
+ opacity: 0.5;
+ }
+ }
}
@keyframes card-glow {
0%,
100% {
- box-shadow: var(--shadow-md), 0 0 0 0 rgba(251, 191, 36, 0);
+ opacity: 0;
}
50% {
- box-shadow: var(--shadow-md), 0 0 16px 4px rgba(251, 191, 36, 0.6);
+ opacity: 1;
}
}
diff --git a/webui/src/components/ab-search-bar.vue b/webui/src/components/ab-search-bar.vue
index 5b9b8ecb..6b84a939 100644
--- a/webui/src/components/ab-search-bar.vue
+++ b/webui/src/components/ab-search-bar.vue
@@ -1,9 +1,13 @@
@@ -302,30 +350,29 @@ function groupCheckedKeys(group: TorrentGroup): string[] {
{{ $t('downloader.empty_torrents') }}
-
-
- onCheckedChange(group, keys as string[])"
- />
-
-
+
+
-
+
- {{ selectedKeys.length }} {{ $t('downloader.selected') }}
+ {{ selectedKeys.size }} {{ $t('downloader.selected') }}