Merge pull request #1141 from EstrellaXD/feat/4.0-perf

perf: 4.0 性能优化(日志回归、下载页虚拟滚动、SSE 增量、RSS 刷新、懒加载)
This commit is contained in:
Xingyu (Estrella) Pan
2026-10-20 18:28:02 +02:00
committed by GitHub
45 changed files with 1801 additions and 344 deletions

View File

@@ -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

View File

@@ -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)

View File

@@ -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()

View File

@@ -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)

View File

@@ -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", "")

View File

@@ -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

View File

@@ -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 决定是否外发)。

View File

@@ -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

View File

@@ -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()
)

View File

@@ -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,避免外键约束报错

View File

@@ -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]

View File

@@ -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]

View File

@@ -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

View File

@@ -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

View File

@@ -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:

View File

@@ -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)

View File

@@ -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

View File

@@ -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)
# ---------------------------------------------------------------------------

View File

@@ -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()

View File

@@ -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)

View File

@@ -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("<html>spa</html>")
(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("<svg/>")
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

View File

@@ -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)

View File

@@ -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()):

View File

@@ -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

View File

@@ -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)

View File

@@ -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 == {}

View File

@@ -42,6 +42,9 @@ exc_value # __aexit__
traceback # __aexit__
hash_ # for 循环解包
# --- 模块级 __getattr__(PEP 562):解释器在属性缺失时调用 ---
__getattr__
# --- 测试 / e2e 支撑:生产路径不调用 ---
_.check_single # e2e worker
_.set_renamed_path # aria2 测试构造状态

View File

@@ -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;
},

View File

@@ -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;
}
}

View File

@@ -1,9 +1,13 @@
<script lang="ts" setup>
import AbSearchModal from './search/ab-search-modal.vue';
const { showModal, provider, loading } = storeToRefs(useSearchStore());
const { toggleModal, getProviders } = useSearchStore();
// 搜索弹窗首次打开时才加载
const { comp: SearchModal } = useLazyModal(
showModal,
() => import('./search/ab-search-modal.vue')
);
onMounted(() => {
getProviders();
});
@@ -18,5 +22,5 @@ onMounted(() => {
/>
<!-- Search Modal -->
<AbSearchModal @close="toggleModal" />
<component :is="SearchModal" v-if="SearchModal" @close="toggleModal" />
</template>

View File

@@ -18,6 +18,10 @@ const { isMobile } = useBreakpointQuery();
const showAccount = ref(false);
const rssRule = ref(ruleTemplate);
const { comp: AddRss, shown: addRssShown } = useLazyModal(
showAddRSS,
() => import('@/components/ab-add-rss.vue')
);
const { start, pause, shutdown, restart, resetRule } = useProgramStore();
const { refreshPoster } = useBangumiStore();
@@ -138,7 +142,13 @@ onUnmounted(() => {
</div>
<ab-change-account v-model:show="showAccount" />
<ab-add-rss v-model:show="showAddRSS" v-model:rule="rssRule" />
<component
:is="AddRss"
v-if="AddRss"
v-model:rule="rssRule"
:show="addRssShown"
@update:show="showAddRSS = $event"
/>
</div>
</template>

View File

@@ -615,12 +615,12 @@ function handleFilterClick(
<template>
<Teleport to="body">
<!-- Backdrop -->
<transition name="overlay">
<transition name="overlay" appear>
<div v-if="showModal" class="modal-backdrop" />
</transition>
<!-- Modal -->
<transition name="modal">
<transition name="modal" appear>
<div
v-if="showModal"
class="modal-container"

View File

@@ -4,7 +4,7 @@
* store) when available.
*/
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { ref } from 'vue';
import { withSetup } from './test-utils';
@@ -59,12 +59,31 @@ async function freshEventStream() {
}
describe('useEventStream', () => {
// 单例在模块级注册 visibilitychange 监听且从不卸载;逐测试移除,避免旧实例串扰
const visibilityHandlers: EventListener[] = [];
const realAddEventListener = document.addEventListener.bind(document);
beforeEach(() => {
vi.spyOn(document, 'addEventListener').mockImplementation(
(type: string, handler: unknown, opts?: unknown) => {
if (type === 'visibilitychange') {
visibilityHandlers.push(handler as EventListener);
}
realAddEventListener(type, handler as EventListener, opts as never);
}
);
isLoggedIn.value = false;
MockEventSource.instances = [];
vi.stubGlobal('EventSource', MockEventSource);
});
afterEach(() => {
for (const h of visibilityHandlers.splice(0)) {
document.removeEventListener('visibilitychange', h);
}
vi.restoreAllMocks();
});
it('should open a connection to the events stream when logged in', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
@@ -176,4 +195,103 @@ describe('useEventStream', () => {
expect(second).toHaveBeenCalledWith(1);
});
it('should append log_append frames and replace on a log frame', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
const result = withSetup(() => useEventStream());
const es = MockEventSource.instances.at(-1)!;
es.emit('log_append', 'a\n');
expect(result.logData.value).toBe('a\n');
es.emit('log', 'snapshot\n');
es.emit('log_append', 'b\n');
expect(result.logData.value).toBe('snapshot\nb\n');
es.emit('log', 'rotated\n');
expect(result.logData.value).toBe('rotated\n');
});
it('should empty the log when a log_clear frame arrives', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
const result = withSetup(() => useEventStream());
const es = MockEventSource.instances.at(-1)!;
es.emit('log', 'stale\n');
es.emit('log_clear', '1');
expect(result.logData.value).toBe('');
es.emit('log_append', 'fresh\n');
expect(result.logData.value).toBe('fresh\n');
});
it('should trim the oldest lines when appended log exceeds the cap', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
const result = withSetup(() => useEventStream());
const es = MockEventSource.instances.at(-1)!;
const line = `${'x'.repeat(99)}\n`;
es.emit('log', line.repeat(5000));
es.emit('log_append', 'tail\n');
const text = result.logData.value!;
expect(text.length).toBeLessThanOrEqual(512 * 1024);
expect(text.endsWith('x\ntail\n')).toBe(true);
expect(text.startsWith(`${'x'.repeat(99)}\n`)).toBe(true);
});
describe('hidden tab', () => {
let hidden = false;
function setHidden(value: boolean) {
hidden = value;
document.dispatchEvent(new Event('visibilitychange'));
}
beforeEach(() => {
vi.useFakeTimers();
hidden = false;
vi.spyOn(document, 'hidden', 'get').mockImplementation(() => hidden);
});
afterEach(() => {
vi.useRealTimers();
});
it('should close the stream after 30 s hidden and reconnect on visible', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
const result = withSetup(() => useEventStream());
const es = MockEventSource.instances.at(-1)!;
es.onopen?.();
// 旧测试遗留的单例也会连接,故以当前数量为基线
const baseline = MockEventSource.instances.length;
setHidden(true);
vi.advanceTimersByTime(30_000);
expect(es.closed).toBe(true);
expect(result.connected.value).toBe(false);
// 隐藏期间不会重连
vi.advanceTimersByTime(60_000);
expect(MockEventSource.instances).toHaveLength(baseline);
setHidden(false);
expect(MockEventSource.instances).toHaveLength(baseline + 1);
expect(MockEventSource.instances.at(-1)!.closed).toBe(false);
});
it('should keep the stream when hidden for less than 30 s', async () => {
isLoggedIn.value = true;
const useEventStream = await freshEventStream();
withSetup(() => useEventStream());
const es = MockEventSource.instances.at(-1)!;
const baseline = MockEventSource.instances.length;
setHidden(true);
vi.advanceTimersByTime(29_000);
setHidden(false);
vi.advanceTimersByTime(60_000);
expect(es.closed).toBe(false);
expect(MockEventSource.instances).toHaveLength(baseline);
});
});
});

View File

@@ -0,0 +1,38 @@
/**
* Tests for useMyI18n: a slow locale import must not override a later switch.
*/
import { ref } from 'vue';
import { flushPromises } from '@vue/test-utils';
import { describe, expect, it, vi } from 'vitest';
const gate = vi.hoisted(() => {
let release!: () => void;
const promise = new Promise<void>((r) => (release = r));
return { promise, release };
});
// setup.ts 把 createI18n 置空,这里换成带 locale ref 的最小实现
vi.mock('vue-i18n', () => ({
createI18n: () => ({
global: { locale: ref('en'), setLocaleMessage: vi.fn(), t: vi.fn() },
}),
}));
vi.mock('@/i18n/zh-CN.json', async () => {
await gate.promise;
return { default: {} };
});
describe('useMyI18n', () => {
it('should keep the latest locale when an earlier import resolves late', async () => {
const { useMyI18n } = await import('@/hooks/useMyI18n');
const { lang, locale } = useMyI18n();
lang.value = 'zh-CN';
await flushPromises();
lang.value = 'en';
await flushPromises();
expect(locale.value).toBe('en');
gate.release();
await new Promise((r) => setTimeout(r, 50));
expect(locale.value).toBe('en');
});
});

View File

@@ -22,6 +22,10 @@ export interface BusEvent {
const RECONNECT_BASE_DELAY_MS = 1000;
const RECONNECT_MAX_DELAY_MS = 15000;
/** 标签页持续隐藏超过该时长后关闭流,可见时重连(服务端重发快照) */
const HIDDEN_CLOSE_DELAY_MS = 30000;
/** 累积日志文本的上限,与后端 /log 的尾部读取预算一致 */
const LOG_MAX_CHARS = 512 * 1024;
/**
* 单一 SSE 连接(api/v1/events/stream),推送 status/downloader/log 更新,
@@ -56,6 +60,9 @@ export const useEventStream = createSharedComposable(() => {
let source: EventSource | null = null;
let retryCount = 0;
let retryTimer: ReturnType<typeof setTimeout> | undefined;
let hiddenTimer: ReturnType<typeof setTimeout> | undefined;
/** 因标签页长时间隐藏而主动关闭;此状态下不重连,直到页面可见 */
let suspended = false;
function teardown() {
source?.close();
@@ -71,7 +78,7 @@ export const useEventStream = createSharedComposable(() => {
);
retryCount += 1;
retryTimer = setTimeout(() => {
if (isLoggedIn.value) connect();
if (isLoggedIn.value && !suspended) connect();
}, delay);
}
@@ -111,6 +118,22 @@ export const useEventStream = createSharedComposable(() => {
logData.value = (e as MessageEvent).data;
});
// EventSource 不派发 data 为空的事件,空日志由服务端改推 log_clear
es.addEventListener('log_clear', () => {
logData.value = '';
});
es.addEventListener('log_append', (e) => {
const text = (logData.value ?? '') + (e as MessageEvent).data;
if (text.length <= LOG_MAX_CHARS) {
logData.value = text;
return;
}
// 超出上限时丢弃最旧部分,并对齐到下一行起点
const tail = text.slice(text.length - LOG_MAX_CHARS);
logData.value = tail.slice(tail.indexOf('\n') + 1);
});
es.addEventListener('update', (e) => {
try {
updateData.value = JSON.parse((e as MessageEvent).data);
@@ -153,10 +176,28 @@ export const useEventStream = createSharedComposable(() => {
function stop() {
clearTimeout(retryTimer);
clearTimeout(hiddenTimer);
suspended = false;
retryCount = 0;
teardown();
}
function onVisibilityChange() {
clearTimeout(hiddenTimer);
if (document.hidden) {
hiddenTimer = setTimeout(() => {
suspended = true;
clearTimeout(retryTimer);
teardown();
}, HIDDEN_CLOSE_DELAY_MS);
} else if (suspended) {
suspended = false;
retryCount = 0;
if (isLoggedIn.value) connect();
}
}
document.addEventListener('visibilitychange', onVisibilityChange);
// 生命周期跟随登录状态,而非某个调用方组件的挂载/卸载——这是一个跨组件
// 共享的单例连接(createSharedComposable),不能绑定到任意一个消费者的
// onBeforeUnmount 上。

View File

@@ -0,0 +1,41 @@
import type { Component, Ref } from 'vue';
/**
* 弹窗首次打开时才加载并挂载。
* `comp` 就绪后才渲染,`shown` 在挂载后的下一拍再置为 true,
* 这样 NModal 的进入过渡(appear)仍然生效。
*/
export function useLazyModal(
open: Ref<boolean>,
load: () => Promise<{ default: Component }>
) {
const comp = shallowRef<Component>();
const shown = ref(false);
watch(
open,
async (val) => {
if (val && !comp.value) {
try {
comp.value = (await load()).default;
} catch {
// 升级后旧的分包 404:整页刷新一次拿新入口,sessionStorage 标记防止循环刷新
try {
if (!sessionStorage.getItem('lazy-modal-reload')) {
sessionStorage.setItem('lazy-modal-reload', '1');
location.reload();
}
} catch {
// sessionStorage 不可用时不刷新,避免无限循环
}
return;
}
await nextTick();
}
shown.value = open.value;
},
{ immediate: true }
);
return { comp, shown };
}

View File

@@ -1,15 +1,9 @@
import { createI18n } from 'vue-i18n';
import { createSharedComposable, useLocalStorage } from '@vueuse/core';
import enUS from '@/i18n/en.json';
import zhCN from '@/i18n/zh-CN.json';
import type { ApiSuccess } from '#/api';
const messages = {
en: enUS,
'zh-CN': zhCN,
};
type Languages = keyof typeof messages;
type Languages = 'en' | 'zh-CN';
function normalizeLocale(locale: string): Languages {
if (locale.startsWith('zh')) return 'zh-CN';
@@ -20,19 +14,44 @@ export const i18n = createI18n({
legacy: false,
locale: normalizeLocale(navigator.language),
fallbackLocale: 'en',
messages,
// zh-CN 先放空壳以固定语言类型,真正的消息由 loadLocale 载入
messages: { en: enUS, 'zh-CN': {} },
});
// 默认语言(en)静态打包;其它语言首次使用时才加载
const loaded = new Set<Languages>(['en']);
export async function loadLocale(locale: Languages) {
if (loaded.has(locale)) return;
const { default: msgs } = await import('@/i18n/zh-CN.json');
i18n.global.setLocaleMessage(locale, msgs);
loaded.add(locale);
}
/** 首次渲染前预载已保存的语言,避免先闪出英文 */
export function loadSavedLocale() {
let saved: string | null = null;
try {
saved = localStorage.getItem('lang');
} catch {
// 隐私模式等场景下读取会抛错,回退到浏览器语言
}
return loadLocale(normalizeLocale(saved ?? navigator.language));
}
export const useMyI18n = createSharedComposable(() => {
const lang = useLocalStorage<Languages>(
'lang',
normalizeLocale(navigator.language)
);
i18n.global.locale.value = lang.value as unknown as Languages;
i18n.global.locale.value = lang.value;
watch(lang, (val) => {
i18n.global.locale.value = val as unknown as Languages;
// 先加载语言包再切换,切换期间沿用旧语言
watch(lang, async (val) => {
await loadLocale(val);
// 慢的导入返回时用户可能已切到别的语言,只应用最新一次
if (lang.value === val) i18n.global.locale.value = val;
});
function changeLocale() {

View File

@@ -1,7 +1,7 @@
import { createApp } from 'vue';
import { createPinia } from 'pinia';
import { router } from './router';
import { i18n } from './hooks/useMyI18n';
import { i18n, loadSavedLocale } from './hooks/useMyI18n';
import App from './App.vue';
import '@unocss/reset/tailwind-compat.css';
@@ -13,4 +13,8 @@ const app = createApp(App);
app.use(router);
app.use(pinia);
app.use(i18n);
app.mount('#app');
// 已保存的语言包加载完再渲染,避免先闪出英文
// 加载失败也要挂载,回退到默认语言
loadSavedLocale()
.catch(() => {})
.then(() => app.mount('#app'));

View File

@@ -8,6 +8,14 @@ definePage({
});
const { editRule } = storeToRefs(useBangumiStore());
const editRuleOpen = computed({
get: () => editRule.value.show,
set: (val) => (editRule.value.show = val),
});
const { comp: EditRule, shown: editRuleShown } = useLazyModal(
editRuleOpen,
() => import('@/components/ab-edit-rule.vue')
);
// 插件页面的标题取自清单(清单到达前留空,避免闪出路由名 Plugin;清单已到达
// 或加载失败仍无匹配页面时回退为插件 id);插件市场用翻译后的标题,其它页面沿用路由名
@@ -74,9 +82,12 @@ function onApplyRule(rule: BangumiRule) {
</div>
</main>
<ab-edit-rule
v-model:show="editRule.show"
<component
:is="EditRule"
v-if="EditRule"
v-model:rule="editRule.item"
:show="editRuleShown"
@update:show="editRuleOpen = $event"
@enable="onEnableRule"
@archive="onArchiveRule"
@unarchive="onUnarchiveRule"

View File

@@ -0,0 +1,158 @@
import { flushPromises, mount } from '@vue/test-utils';
import { createPinia, setActivePinia } from 'pinia';
import { ref } from 'vue';
import type { Component } from 'vue';
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { useConfigStore } from '@/store/config';
import { useDownloaderStore } from '@/store/downloader';
import { mockConfig } from '@/test/mocks/api';
import type { Config } from '#/config';
import type { QbTorrentInfo } from '#/downloader';
// happy-dom 没有布局:给虚拟列表一个固定视口高度,否则一行都不渲染。
// 必须在 naive-ui(vueuc 的 ResizeObserver 单例)加载前替换
vi.hoisted(() => {
class FixedResizeObserver {
callback: ResizeObserverCallback;
constructor(callback: ResizeObserverCallback) {
this.callback = callback;
}
observe(target: Element) {
const entry = { target, contentRect: { width: 1000, height: 600 } };
this.callback(
[entry as unknown as ResizeObserverEntry],
this as unknown as ResizeObserver
);
}
unobserve() {}
disconnect() {}
}
globalThis.ResizeObserver = FixedResizeObserver;
});
const downloaderData = ref<QbTorrentInfo[] | null>(null);
vi.mock('@/hooks/useEventStream', () => ({
useEventStream: () => ({ connected: ref(true), downloaderData }),
}));
vi.mock('@/utils/axios', () => ({
axios: { get: vi.fn(), post: vi.fn() },
}));
vi.mock('@/api/downloader', () => ({
apiDownloader: {
getTorrents: vi.fn().mockResolvedValue([]),
instances: vi.fn().mockResolvedValue({ default: '', instances: [] }),
},
}));
vi.mock('@/hooks/useMyI18n', () => ({
useMyI18n: () => ({
t: (key: string) => key,
returnUserLangMsg: () => '',
}),
}));
vi.mock('@/hooks/useMessage', () => ({
useMessage: () => ({ success: vi.fn(), error: vi.fn() }),
}));
function torrent(
hash: string,
savePath: string,
progress = 0.5
): QbTorrentInfo {
return {
hash,
name: `torrent-${hash}`,
size: 1024,
progress,
dlspeed: 0,
upspeed: 0,
num_seeds: 0,
num_leechs: 0,
state: 'downloading',
eta: 0,
category: 'Bangumi',
save_path: savePath,
added_on: 0,
downloader_id: 'default',
};
}
describe('Downloader page', () => {
let DownloaderPage: Component;
beforeAll(async () => {
vi.stubGlobal('definePage', vi.fn());
// vitest 的自动导入不含 @vueuse/core
vi.stubGlobal('useIntervalFn', () => ({ pause: vi.fn(), resume: vi.fn() }));
DownloaderPage = (await import('./downloader.vue')).default;
});
beforeEach(() => {
setActivePinia(createPinia());
useConfigStore().config = mockConfig as unknown as Config;
downloaderData.value = null;
});
async function mountPage() {
const wrapper = mount(DownloaderPage, {
attachTo: document.body,
global: { stubs: { 'ab-button': true, RouterLink: true } },
});
// 首帧在挂载后到达,与真实 SSE 一致(store 的 watch 不是 immediate)
downloaderData.value = [
torrent('a1', '/dl/Show A'),
torrent('a2', '/dl/Show A'),
torrent('b1', '/dl/Show B'),
];
await flushPromises();
return wrapper;
}
function rowOf(wrapper: Awaited<ReturnType<typeof mountPage>>, text: string) {
const row = wrapper
.findAll('.n-data-table-tbody .n-data-table-tr')
.find((r) => r.text().includes(text));
if (!row) throw new Error(`row not found: ${text}`);
return row;
}
it('should update the row when the store receives a new SSE payload', async () => {
const wrapper = await mountPage();
expect(rowOf(wrapper, 'torrent-a1').text()).toContain('50%');
downloaderData.value = [torrent('a1', '/dl/Show A', 1)];
await flushPromises();
expect(rowOf(wrapper, 'torrent-a1').text()).toContain('100%');
});
it('should select every torrent of the group when its header checkbox is checked', async () => {
const wrapper = await mountPage();
await rowOf(wrapper, 'Show A (2)').find('.n-checkbox').trigger('click');
expect([...useDownloaderStore().selectedKeys].sort()).toEqual([
'default:a1',
'default:a2',
]);
expect(wrapper.find('.action-bar-count').text()).toContain('2');
});
it('should hide the group rows when the group is collapsed', async () => {
const wrapper = await mountPage();
await rowOf(wrapper, 'Show A (2)')
.find('.n-data-table-expand-trigger')
.trigger('click');
const text = wrapper.find('.n-data-table-tbody').text();
expect(text).not.toContain('torrent-a1');
expect(text).toContain('torrent-b1');
});
});

View File

@@ -1,5 +1,10 @@
<script lang="ts" setup>
import { type DataTableColumns, NDataTable } from 'naive-ui';
import type { VNodeChild } from 'vue';
import {
type DataTableColumns,
type DataTableRowKey,
NDataTable,
} from 'naive-ui';
import AbProgress from '@/components/basic/ab-progress.vue';
import AbTag from '@/components/basic/ab-tag.vue';
import { useConfirm } from '@/hooks/useConfirm';
@@ -20,8 +25,6 @@ const {
pauseSelected,
resumeSelected,
deleteSelected,
toggleKey,
toggleGroup,
clearSelection,
} = useDownloaderStore();
const { confirm } = useConfirm();
@@ -120,30 +123,89 @@ function stateType(state: string): 'neutral' | 'success' | 'info' | 'danger' {
return 'info';
}
function isGroupAllSelected(group: TorrentGroup): boolean {
return group.torrents.every((t) =>
selectedKeys.value.includes(torrentKey(t))
/** 分组头行:同一个虚拟表里以树形父行显示,children 为该组的种子 */
interface GroupRow {
key: string;
group: TorrentGroup;
children: QbTorrentInfo[];
}
type Row = QbTorrentInfo | GroupRow;
function isGroupRow(row: Row): row is GroupRow {
return 'group' in row;
}
function rowKey(row: Row): string {
return isGroupRow(row) ? row.key : torrentKey(row);
}
function rowClassName(row: Row): string {
return isGroupRow(row) ? 'group-row' : '';
}
/** 分组头行的名称格跨满其余列,种子列的渲染只处理种子行 */
function torrentCell(render: (row: QbTorrentInfo) => VNodeChild) {
return (row: Row) => (isGroupRow(row) ? null : render(row));
}
const rows = computed<GroupRow[]>(() =>
groups.value.map((group) => ({
key: `group:${group.savePath}`,
group,
children: group.torrents,
}))
);
// 记录折叠的分组,新出现的分组默认展开
const collapsed = shallowRef(new Set<string>());
const expandedKeys = computed(() =>
rows.value.filter((r) => !collapsed.value.has(r.key)).map((r) => r.key)
);
function onExpandedChange(keys: DataTableRowKey[]) {
const open = new Set(keys);
collapsed.value = new Set(
rows.value.map((r) => r.key).filter((k) => !open.has(k))
);
}
const checkedKeys = computed(() => [...selectedKeys.value]);
// 级联勾选会带上分组头的键,只保留种子键
function onCheckedChange(keys: DataTableRowKey[]) {
const groupKeys = new Set(rows.value.map((r) => r.key));
selectedKeys.value = new Set(
keys.filter((k) => !groupKeys.has(k as string)) as string[]
);
}
// 多个下载器实例时标出每个种子所在的实例
const downloaders = useDownloaderInstances();
const tableColumnsValue = computed<DataTableColumns<QbTorrentInfo>>(() => [
const tableColumnsValue = computed<DataTableColumns<Row>>(() => [
{
type: 'selection',
},
{
title: t('downloader.torrent.name'),
key: 'name',
ellipsis: { tooltip: true },
// CSS 省略号 + title:不用逐行测量宽度的 tooltip
ellipsis: true,
className: 'name-col',
minWidth: 200,
colSpan: (row: Row): number =>
isGroupRow(row) ? tableColumnsValue.value.length - 1 : 1,
cellProps: (row) => ({
title: isGroupRow(row) ? row.group.savePath : row.name,
}),
render: (row) =>
isGroupRow(row) ? `${row.group.name} (${row.group.count})` : row.name,
},
{
title: t('downloader.torrent.progress'),
key: 'progress',
width: 160,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return h(AbProgress, {
value: row.progress * 100,
label: `${Math.round(row.progress * 100)}%`,
@@ -152,58 +214,58 @@ const tableColumnsValue = computed<DataTableColumns<QbTorrentInfo>>(() => [
? 'error'
: 'active',
});
},
}),
},
{
title: t('downloader.torrent.status'),
key: 'state',
width: 100,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return h(AbTag, {
type: stateType(row.state),
title: stateLabel(row.state),
});
},
}),
},
{
title: t('downloader.torrent.size'),
key: 'size',
width: 100,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return formatSize(row.size);
},
}),
},
{
title: t('downloader.torrent.dlspeed'),
key: 'dlspeed',
width: 110,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return formatSpeed(row.dlspeed);
},
}),
},
{
title: t('downloader.torrent.upspeed'),
key: 'upspeed',
width: 110,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return formatSpeed(row.upspeed);
},
}),
},
{
title: 'ETA',
key: 'eta',
width: 80,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return formatEta(row.eta);
},
}),
},
{
title: t('downloader.torrent.peers'),
key: 'peers',
width: 110,
render(row: QbTorrentInfo) {
render: torrentCell((row) => {
return `${row.num_seeds} / ${row.num_leechs}`;
},
}),
},
// 放在最后并完整显示 id(触屏没有悬停提示)
...(downloaders.multiple.value
@@ -216,20 +278,6 @@ const tableColumnsValue = computed<DataTableColumns<QbTorrentInfo>>(() => [
]
: []),
]);
function onCheckedChange(group: TorrentGroup, keys: string[]) {
const groupKeys = group.torrents.map(torrentKey);
const otherSelected = selectedKeys.value.filter(
(k) => !groupKeys.includes(k)
);
selectedKeys.value = [...otherSelected, ...keys];
}
function groupCheckedKeys(group: TorrentGroup): string[] {
return group.torrents
.map(torrentKey)
.filter((k) => selectedKeys.value.includes(k));
}
</script>
<template>
@@ -302,30 +350,29 @@ function groupCheckedKeys(group: TorrentGroup): string[] {
{{ $t('downloader.empty_torrents') }}
</div>
<div v-else class="downloader-groups">
<ab-fold-panel
v-for="group in groups"
:key="group.savePath"
:title="`${group.name} (${group.count})`"
:default-open="true"
>
<NDataTable
:columns="tableColumnsValue"
:data="group.torrents"
:row-key="torrentKey"
:pagination="false"
:bordered="false"
:checked-row-keys="groupCheckedKeys(group)"
size="small"
@update:checked-row-keys="(keys: any) => onCheckedChange(group, keys as string[])"
/>
</ab-fold-panel>
</div>
<!-- 单个虚拟滚动表:只渲染可见行;不放进 NCollapseItem,SSE 数据才能刷新行 -->
<NDataTable
v-else
class="downloader-table"
:columns="tableColumnsValue"
:data="rows"
:row-key="rowKey"
:row-class-name="rowClassName"
:pagination="false"
:bordered="false"
:checked-row-keys="checkedKeys"
:expanded-row-keys="expandedKeys"
virtual-scroll
flex-height
size="small"
@update:checked-row-keys="onCheckedChange"
@update:expanded-row-keys="onExpandedChange"
/>
<Transition name="fade">
<div v-if="selectedKeys.length > 0" class="action-bar">
<div v-if="selectedKeys.size > 0" class="action-bar">
<span class="action-bar-count">
{{ selectedKeys.length }} {{ $t('downloader.selected') }}
{{ selectedKeys.size }} {{ $t('downloader.selected') }}
</span>
<div class="action-bar-buttons">
<ab-button variant="primary" size="sm" @click="resumeSelected">
@@ -354,6 +401,7 @@ function groupCheckedKeys(group: TorrentGroup): string[] {
.page-downloader {
overflow: auto;
flex-grow: 1;
min-height: 0;
display: flex;
flex-direction: column;
}
@@ -362,14 +410,28 @@ function groupCheckedKeys(group: TorrentGroup): string[] {
display: flex;
flex-direction: column;
flex: 1;
min-height: 0;
gap: 12px;
padding-bottom: 60px;
}
.downloader-groups {
display: flex;
flex-direction: column;
gap: 12px;
// flex-height:表格填满页面剩余高度,虚拟滚动在表体内进行
.downloader-table {
flex: 1;
min-height: 0;
:deep(.group-row .n-data-table-td) {
font-weight: 600;
}
// 树形缩进与展开按钮占 40px,名称不换行到下一行
:deep(.name-col) {
white-space: nowrap;
}
:deep(.name-col .n-data-table-td__ellipsis) {
max-width: calc(100% - 40px);
}
}
.downloader-empty {

View File

@@ -1,11 +1,12 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { ref } from 'vue';
import { nextTick, ref } from 'vue';
import { apiDownloader } from '@/api/downloader';
import {
hasConfiguredDownloader,
torrentKey,
useDownloaderStore,
} from '@/store/downloader';
import type { Ref } from 'vue';
import type { QbTorrentInfo } from '#/downloader';
import type { PluginInstance } from '#/config';
@@ -18,8 +19,14 @@ vi.mock('@/api/downloader', () => ({
},
}));
const downloaderData = vi.hoisted(() => ({
current: null as Ref<QbTorrentInfo[] | null> | null,
}));
vi.mock('@/hooks/useEventStream', () => ({
useEventStream: () => ({ downloaderData: ref(null) }),
useEventStream: () => {
downloaderData.current = ref(null);
return { downloaderData: downloaderData.current };
},
}));
vi.mock('@/hooks/useApi', () => ({
@@ -59,7 +66,7 @@ describe('downloader store', () => {
torrent('a2', 'default'),
torrent('a3', 'default'),
];
store.selectedKeys = store.torrents.slice(0, 3).map(torrentKey);
store.selectedKeys = new Set(store.torrents.slice(0, 3).map(torrentKey));
await store.pauseSelected();
await store.deleteSelected(true);
@@ -77,7 +84,7 @@ describe('downloader store', () => {
it('should act only on the selected instance when the same hash is in two instances', async () => {
const store = useDownloaderStore();
store.torrents = [torrent('h', 'home'), torrent('h', 'seedbox')];
store.selectedKeys = [torrentKey(store.torrents[1])];
store.selectedKeys = new Set([torrentKey(store.torrents[1])]);
await store.pauseSelected();
@@ -98,7 +105,7 @@ describe('downloader store', () => {
async (_, failure) => {
const store = useDownloaderStore();
store.torrents = [torrent('a', 'home'), torrent('b', 'seedbox')];
store.selectedKeys = store.torrents.map(torrentKey);
store.selectedKeys = new Set(store.torrents.map(torrentKey));
vi.mocked(apiDownloader.deleteTorrents).mockImplementationOnce(failure);
await expect(store.deleteSelected(false)).rejects.toBeTruthy();
@@ -110,10 +117,28 @@ describe('downloader store', () => {
}
);
it('should discard an HTTP response when an SSE frame arrived during the request', async () => {
const store = useDownloaderStore();
let respond!: (list: QbTorrentInfo[]) => void;
vi.mocked(apiDownloader.getTorrents).mockReturnValueOnce(
new Promise((resolve) => {
respond = resolve;
})
);
const request = store.getAll();
downloaderData.current!.value = [torrent('fresh', 'home')];
await nextTick();
respond([torrent('fresh', 'home'), torrent('deleted', 'home')]);
await request;
expect(store.torrents.map((t) => t.hash)).toEqual(['fresh']);
});
it('should show the failure message when an instance returns status false', async () => {
const store = useDownloaderStore();
store.torrents = [torrent('a', 'home')];
store.selectedKeys = store.torrents.map(torrentKey);
store.selectedKeys = new Set(store.torrents.map(torrentKey));
vi.mocked(apiDownloader.deleteTorrents).mockResolvedValueOnce({
status: false,
msg_en: 'Failed to delete torrents',

View File

@@ -14,13 +14,18 @@ export function hasConfiguredDownloader(instances: PluginInstance[]): boolean {
export const useDownloaderStore = defineStore('downloader', () => {
const torrents = shallowRef<QbTorrentInfo[]>([]);
const selectedKeys = ref<string[]>([]);
// Set 而不是响应式数组:成员判断 O(1),整体替换触发更新
const selectedKeys = shallowRef(new Set<string>());
const loading = ref(false);
const { downloaderData } = useEventStream();
// SSE 帧计数:getAll 请求期间收到过新帧时丢弃 HTTP 结果——服务端只在
// 负载变化时重推,旧响应覆盖新帧后已删除的种子会一直留在列表里。
let sseFrames = 0;
// SSE 已连接时使用推送数据,省去 downloader.vue 页面自己的 5s 轮询请求。
watch(downloaderData, (data) => {
if (data === null) return;
sseFrames += 1;
torrents.value = data;
});
@@ -65,8 +70,10 @@ export const useDownloaderStore = defineStore('downloader', () => {
async function getAll() {
loading.value = true;
const frames = sseFrames;
try {
torrents.value = await apiDownloader.getTorrents();
const data = await apiDownloader.getTorrents();
if (frames === sseFrames) torrents.value = data;
} catch {
// Keep the last known list — a failed 5s poll (e.g. during a backend
// restart) must not blank the page into a fake "no torrents" state.
@@ -82,7 +89,7 @@ export const useDownloaderStore = defineStore('downloader', () => {
const opts = {
showMessage: true,
onSuccess() {
selectedKeys.value = [];
selectedKeys.value = new Set();
},
onFinally: getAll,
};
@@ -97,7 +104,7 @@ export const useDownloaderStore = defineStore('downloader', () => {
) {
const byInstance = new Map<string, string[]>();
for (const t of torrents.value) {
if (!selectedKeys.value.includes(torrentKey(t))) continue;
if (!selectedKeys.value.has(torrentKey(t))) continue;
const hashes = byInstance.get(t.downloader_id) ?? [];
hashes.push(t.hash);
byInstance.set(t.downloader_id, hashes);
@@ -134,30 +141,8 @@ export const useDownloaderStore = defineStore('downloader', () => {
opts
);
function toggleKey(key: string) {
const idx = selectedKeys.value.indexOf(key);
if (idx === -1) {
selectedKeys.value.push(key);
} else {
selectedKeys.value.splice(idx, 1);
}
}
function toggleGroup(group: TorrentGroup) {
const groupKeys = group.torrents.map(torrentKey);
const allSelected = groupKeys.every((k) => selectedKeys.value.includes(k));
if (allSelected) {
selectedKeys.value = selectedKeys.value.filter(
(k) => !groupKeys.includes(k)
);
} else {
const toAdd = groupKeys.filter((k) => !selectedKeys.value.includes(k));
selectedKeys.value.push(...toAdd);
}
}
function clearSelection() {
selectedKeys.value = [];
selectedKeys.value = new Set();
}
return {
@@ -170,8 +155,6 @@ export const useDownloaderStore = defineStore('downloader', () => {
pauseSelected,
resumeSelected,
deleteSelected,
toggleKey,
toggleGroup,
clearSelection,
};
});

View File

@@ -67,6 +67,8 @@ declare global {
const isRef: typeof import('vue').isRef
const isShallow: typeof import('vue').isShallow
const isValidNexusPhpCategoryIds: typeof import('../../src/utils/nexusphp').isValidNexusPhpCategoryIds
const loadLocale: typeof import('../../src/hooks/useMyI18n').loadLocale
const loadSavedLocale: typeof import('../../src/hooks/useMyI18n').loadSavedLocale
const localizeFields: typeof import('../../src/utils/plugin-text').localizeFields
const mapActions: typeof import('pinia').mapActions
const mapGetters: typeof import('pinia').mapGetters
@@ -150,6 +152,7 @@ declare global {
const useI18n: typeof import('vue-i18n').useI18n
const useId: typeof import('vue').useId
const useIntervalFn: typeof import('@vueuse/core').useIntervalFn
const useLazyModal: typeof import('../../src/hooks/useLazyModal').useLazyModal
const useLocalStorage: typeof import('@vueuse/core').useLocalStorage
const useLogStore: typeof import('../../src/store/log').useLogStore
const useMessage: typeof import('../../src/hooks/useMessage').useMessage

View File

@@ -46,6 +46,7 @@ export default defineConfig({
}),
VueI18nPlugin({
include: resolve(import.meta.dirname, './src/i18n/**'),
dropMessageCompiler: true,
}),
],
css: {
@@ -61,7 +62,13 @@ export default defineConfig({
output: { minify: { compress: { dropConsole: true } } },
},
},
resolve: { tsconfigPaths: true },
resolve: {
tsconfigPaths: true,
// 只用运行时:模板都在构建期编译,不需要打包 @vue/compiler-*
alias: [
{ find: /^vue$/, replacement: 'vue/dist/vue.runtime.esm-bundler.js' },
],
},
preview: { headers },
server: {
proxy: {