fix: 审查修复——启动时一次性去掉旧搜索订阅关键词里的集号 (#1138)、回退到 none 的轮次不为路径未变的文件建记录、路径未变的完成记录在任务从下载器删除后清理 (#1082)

- migrations:Migration 新增 data_step,在同一 SAVEPOINT 内执行一次性数据修复;v28 按番剧 season_raw 改写搜索源 URL 的关键词参数,并同步番剧 rss_link
- revision_saga:选定策略未登记时,路径未变的文件不建操作记录也不通知,策略恢复后的那一轮再记录并通知
- rename_operation:prune_done 按本实例当前任务清理路径未变的过期 done 记录,任务列表为空时不清理
This commit is contained in:
Estrella Pan
2026-10-21 09:53:47 +02:00
parent 81cbc549b0
commit 1dda39f2d5
10 changed files with 351 additions and 47 deletions

View File

@@ -170,6 +170,7 @@ Schema migrations are tracked via a `schema_version` table in SQLite. To add a n
1. Append a `Migration(version, "description", (…SQL…), already_applied=column_exists("table", "col"))` entry to the `MIGRATIONS` tuple in `backend/src/module/database/migrations.py` (`CURRENT_SCHEMA_VERSION` is derived from the list — do not edit it by hand)
2. Provide an `already_applied` guard (`column_exists` / `table_exists`) so a schema created out-of-band is detected and skipped
- For a one-time data fix, pass `data_step=<fn(conn: Connection)>` (statements may be empty). It runs after the SQL in the same SAVEPOINT. `schema_version` makes it run once, so its guard may only say "nothing to fix here".
3. Migrations run automatically on startup via `run_migrations()` (each in a SAVEPOINT, stopping on first failure)
## Notes

View File

@@ -11,6 +11,7 @@
import logging
from dataclasses import dataclass
from typing import Any, Callable, get_args, get_origin
from urllib.parse import parse_qsl, urlsplit
from pydantic.fields import FieldInfo
from pydantic_core import PydanticUndefined
@@ -87,6 +88,9 @@ class Migration:
statements: tuple[str, ...]
already_applied: AppliedCheck
guarded_statements: tuple[GuardedStatement, ...] = ()
# 一次性的数据修复:在 SQL 语句之后、同一 SAVEPOINT 内执行;
# already_applied 为真时跳过
data_step: Callable[[Connection], None] | None = None
def pending_statements(self, inspector) -> tuple[str, ...]:
if self.guarded_statements:
@@ -120,6 +124,66 @@ def _has_downloader_id(table: str) -> AppliedCheck:
return check
def _strip_search_keyword_episodes(conn: Connection) -> None:
"""旧版本从搜索结果建的订阅把集号写进了搜索关键词(如 S03E02),刷新时
一直按存下的 URL 拉取,订阅锁死在单集(#1138)。
只改搜索源 URL(主机、路径与某个搜索源模板一致,且带模板里 ``%s`` 所在的
关键词参数):按引用该 URL 的番剧存下的 season_raw,把关键词里的季集写法
换成只含季的写法,其余参数与编码原样保留;番剧 rss_link 里的同一 URL 一并
替换。Mikan 番剧 RSS、用户手填的订阅不变。
"""
# 延迟导入:searcher 依赖 rss/network,模块级导入会与 database 循环
from module.searcher.provider import available_sites, encode_keywords
from module.searcher.searcher import strip_season_episode
keyword_params: dict[tuple[str, str], str] = {}
for config in available_sites().values():
template = urlsplit(config["url"])
for name, value in parse_qsl(template.query):
if value == "%s":
keyword_params[(template.netloc, template.path)] = name
season_by_url: dict[str, str] = {}
for raw, rss_link in conn.execute(
text(
"SELECT season_raw, rss_link FROM bangumi "
"WHERE season_raw IS NOT NULL AND rss_link IS NOT NULL"
)
):
for link in rss_link.split(","):
season_by_url.setdefault(link.strip(), raw)
for rss_id, url in conn.execute(text("SELECT id, url FROM rssitem")).all():
parts = urlsplit(url)
param_name = keyword_params.get((parts.netloc, parts.path))
season_raw = season_by_url.get(url)
if param_name is None or season_raw is None:
continue
old = f"+{encode_keywords([season_raw])}+"
new = f"+{encode_keywords([strip_season_episode(season_raw)])}+"
base, _, query = url.partition("?")
params = query.split("&")
for i, param in enumerate(params):
key, sep, value = param.partition("=")
if key == param_name and sep:
params[i] = f"{key}={f'+{value}+'.replace(old, new, 1)[1:-1]}"
new_url = f"{base}?{'&'.join(params)}"
if new_url == url:
continue
conn.execute(
text("UPDATE rssitem SET url = :new WHERE id = :id"),
{"new": new_url, "id": rss_id},
)
conn.execute(
text(
"UPDATE bangumi SET rss_link = replace(rss_link, :old, :new) "
"WHERE instr(rss_link, :old) > 0"
),
{"old": url, "new": new_url},
)
logger.info(f"Search subscription keyword drops episode: {url} -> {new_url}")
# 迁移按版本顺序执行;版本号与 3.2.x 的历史保持一致,旧数据库照常升级。
MIGRATIONS: tuple[Migration, ...] = (
Migration(
@@ -858,6 +922,20 @@ MIGRATIONS: tuple[Migration, ...] = (
),
_has_downloader_id("aria2_gid"),
),
Migration(
28,
"strip episode from search subscription keywords",
(),
# 数据修复没有可检查的 schema 痕迹,由 schema_version 保证只跑一次;
# 缺少这些列的库里没有要修的订阅
lambda inspector: (
not all_checks(
column_exists("bangumi", "season_raw"),
column_exists("bangumi", "rss_link"),
)(inspector)
),
data_step=_strip_search_keyword_episodes,
),
)
# 由迁移列表派生,新增迁移时无需手动同步
@@ -908,8 +986,12 @@ def run_migrations_conn(conn: Connection) -> None:
if migration.version <= current:
continue
# 每轮重新 inspect:前一个迁移的 DDL 需要对守卫可见
pending = migration.pending_statements(inspect(conn))
if not pending:
inspector = inspect(conn)
pending = migration.pending_statements(inspector)
data_step = (
None if migration.already_applied(inspector) else migration.data_step
)
if not pending and data_step is None:
logger.debug(
f"Migration v{migration.version} skipped "
f"(already applied): {migration.description}"
@@ -919,6 +1001,8 @@ def run_migrations_conn(conn: Connection) -> None:
try:
for stmt in pending:
conn.execute(text(stmt))
if data_step is not None:
data_step(conn)
savepoint.commit()
logger.info(f"Migration v{migration.version}: {migration.description}")
except Exception as e:

View File

@@ -384,17 +384,43 @@ class RenameOperationDatabase:
await self.session.commit()
return True
async def prune_done(self, before: datetime) -> int:
async def prune_done(
self,
before: datetime,
live_task_ids: set[str],
downloader_type: str | None,
) -> int:
"""清理过期的 done 记录。
路径未变的记录(重命名方式 none)只在任务已不在下载器里时清理:这类
种子不打 ab:renamed、每轮都会再处理,删掉仍在的任务的记录会重建记录并
再发一次下载完成通知(#1082)。``live_task_ids`` 是该实例当前的全部任务;
为空时不能证明任务已删除,不清理这类记录。``downloader_type`` 为
``None`` 时不限实例。
"""
expired = (
col(RenameOperation.state) == "done",
col(RenameOperation.updated_at) < before,
)
unchanged = col(RenameOperation.source_path) == col(RenameOperation.target_path)
orphan_ids: list[int | None] = []
if live_task_ids:
statement = select(RenameOperation.id, RenameOperation.new_task_id).where(
*expired, unchanged
)
if downloader_type is not None:
statement = statement.where(
RenameOperation.downloader_type == downloader_type
)
rows = await self.session.execute(statement)
orphan_ids = [
row_id for row_id, task_id in rows if task_id not in live_task_ids
]
result = await self.session.execute(
delete(RenameOperation)
.where(
col(RenameOperation.state) == "done",
col(RenameOperation.updated_at) < before,
# 路径未变的记录(重命名方式 none)不清理:种子不打 ab:renamed、
# 每轮都会再处理,删掉后会重建记录并再发一次下载完成通知(#1082)
# ponytail: 这类记录随已删除的种子累积,量级同 torrent 表;
# 需要时按下载器里已不存在的任务清理
col(RenameOperation.source_path) != col(RenameOperation.target_path),
*expired,
or_(~unchanged, col(RenameOperation.id).in_(orphan_ids)),
)
.execution_options(synchronize_session=False)
)

View File

@@ -879,7 +879,7 @@ class Renamer(RevisionSaga):
episode_offset=episode_offset,
episode_type=episode_type,
# 只认本轮实际用的就是选定策略:未登记而回退到 none 时文件
# 保留原名,恢复后还会真正改名并通知,这里再通知就重复了
# 保留原名,不记录也不通知,恢复后那一轮才记录并通知
notify_unchanged=final_names,
)
if report.notification:
@@ -925,7 +925,9 @@ class Renamer(RevisionSaga):
logger.warning(f"{torrent_name} has no media file")
async with Database() as db:
await db.rename_operation.prune_done(
datetime.now(timezone.utc) - timedelta(days=30)
datetime.now(timezone.utc) - timedelta(days=30),
live_task_ids=set(info_by_hash),
downloader_type=instance_key,
)
logger.debug("Rename process finished.")
return renamed_info

View File

@@ -110,7 +110,6 @@ class RevisionSaga:
bangumi_name: str,
_hash: str,
episode_offset: int,
notify_unchanged: bool = False,
) -> MediaRenameReport:
if prepared.source_path == prepared.target_path:
# 文件名已是目标名(重命名方式 none,或已符合命名规则)时不会走真正的
@@ -119,16 +118,12 @@ class RevisionSaga:
return MediaRenameReport(
result=RenameResult(RenameOutcome.ALREADY_APPLIED),
prepared=prepared,
notification=(
Notification(
official_title=bangumi_name,
season=prepared.episode.season,
episode=self._adjust_episode(
prepared.episode.episode, episode_offset
),
)
if notify_unchanged
else None
notification=Notification(
official_title=bangumi_name,
season=prepared.episode.season,
episode=self._adjust_episode(
prepared.episode.episode, episode_offset
),
),
)
result = await self.client.rename_torrent_file(
@@ -856,6 +851,13 @@ class RevisionSaga:
) -> MediaRenameReport:
"""Claim and execute one non-replacement rename exactly once at a time."""
if prepared.source_path == prepared.target_path and not notify_unchanged:
# 选定策略未登记、本轮回退到 none:原名不是最终名,不建操作记录也不
# 通知;策略恢复后那一轮再建记录并通知(#1082)
return MediaRenameReport(
result=RenameResult(RenameOutcome.ALREADY_APPLIED),
prepared=prepared,
)
template = self._build_operation(
info=info,
prepared=prepared,
@@ -984,7 +986,6 @@ class RevisionSaga:
bangumi_name=bangumi_name,
_hash=info["hash"],
episode_offset=episode_offset,
notify_unchanged=notify_unchanged,
)
if report.result.succeeded:
await self._set_operation_state(claimed, "done")

View File

@@ -28,9 +28,13 @@ def available_sites() -> dict[str, ProviderConfig]:
return sites
def encode_keywords(keywords: list[str]) -> str:
"""把关键词拼成搜索 URL 里的查询串:非单词字符都换成 ``+``。"""
return re.sub(r"[\W_ ]", "+", "+".join(keywords))
def search_url(site: str, keywords: list[str]) -> RSSItem:
keyword = "+".join(keywords)
search_str = re.sub(r"[\W_ ]", "+", keyword)
search_str = encode_keywords(keywords)
providers = available_sites()
if site in providers:
# str.replace, not re.sub: the template is a literal "%s" placeholder,

View File

@@ -30,6 +30,11 @@ _SEASON_WITH_EPISODE = re.compile(
)
def strip_season_episode(season_raw: str) -> str:
"""季集合写的 season_raw 只保留季的部分(S03E02 -> S03)。"""
return _SEASON_WITH_EPISODE.sub(r"\1", season_raw)
# Cache for TMDB preview lookups by official_title. Bounded (LRU-ish,
# oldest-evicted) like _tmdb_cache/_mikan_cache — was previously a plain dict
# and grew unbounded for the life of the process. Values are keyed by parser
@@ -112,7 +117,7 @@ class SearchTorrent:
@staticmethod
def special_url(data: Bangumi | Movie, site: str) -> RSSItem:
keywords = [
_SEASON_WITH_EPISODE.sub(r"\1", value) if key == "season_raw" else value
strip_season_episode(value) if key == "season_raw" else value
for key in SEARCH_KEY
if (value := getattr(data, key, None))
]

View File

@@ -261,11 +261,38 @@ async def test_conflict_query_and_done_pruning(db_session):
conflict.updated_at = now - timedelta(days=31)
db_session.add(conflict)
await db_session.commit()
assert await repo.prune_done(before=now - timedelta(days=30)) == 1
assert (
await repo.prune_done(
before=now - timedelta(days=30),
live_task_ids={"conflict-v4"},
downloader_type=None,
)
== 1
)
async def test_prune_done_keeps_unchanged_path_rows(db_session):
"""路径未变(重命名方式 none)的 done 记录是只通知一次的依据,不能被清理(#1082)。"""
@pytest.mark.parametrize(
("live_task_ids", "downloader_type", "pruned"),
[
({"new-v2"}, None, False),
({"other-task"}, None, True),
(set(), None, False),
({"other-task"}, "qbittorrent", True),
({"other-task"}, "aria2", False),
],
ids=[
"task-present",
"task-gone",
"empty-snapshot",
"same-instance",
"other-instance",
],
)
async def test_prune_done_unchanged_path_rows_only_when_task_gone(
db_session, live_task_ids, downloader_type, pruned
):
"""路径未变(重命名方式 none)的 done 记录是只通知一次的依据:任务仍在、
快照为空或属于别的实例时保留,任务已从该实例删除时清理(#1082)。"""
repo = RenameOperationDatabase(db_session)
now = datetime.now(timezone.utc)
row, _ = await repo.get_or_create(
@@ -280,8 +307,13 @@ async def test_prune_done_keeps_unchanged_path_rows(db_session):
db_session.add(row)
await db_session.commit()
assert await repo.prune_done(before=now - timedelta(days=30)) == 0
assert await repo.get(row.id) is not None
assert await repo.prune_done(
before=now - timedelta(days=30),
live_task_ids=live_task_ids,
downloader_type=downloader_type,
) == int(pruned)
db_session.expunge_all()
assert (await repo.get(row.id) is None) is pruned
async def test_database_facade_exposes_rename_operation_repository(db_engine):

View File

@@ -6,7 +6,7 @@ import pytest
from sqlalchemy import inspect, text
from sqlalchemy.engine import Engine
from sqlalchemy.exc import IntegrityError, OperationalError
from sqlmodel import SQLModel, create_engine
from sqlmodel import SQLModel, create_engine, select
import module.database.migrations as migration_module
from module.database.migrations import (
@@ -894,3 +894,82 @@ class TestRunMigrations:
inspector = inspect(engine)
rssitem_cols = {c["name"] for c in inspector.get_columns("rssitem")}
assert "connection_status" not in rssitem_cols
class TestSearchKeywordEpisodeMigration:
NYAA = "https://nyaa.si/?page=rss&q=%s&c=0_0&f=0"
PROVIDERS = {
"nyaa": {"url": NYAA, "parser": "tmdb"},
"dmhy": {
"url": "http://dmhy.org/topics/rss/rss.xml?keyword=%s",
"parser": "tmdb",
},
"mikan": {
"url": "https://mikanani.me/RSS/Search?searchstr=%s",
"parser": "mikan",
},
}
@pytest.mark.parametrize(
("season_raw", "url", "expected"),
[
(
"S03E02",
NYAA.replace(
"%s",
"ToonsHub+The+Apothecary+Diaries+S03E02+Multi+Subs+NF+WEB+DL+1080p",
),
NYAA.replace(
"%s",
"ToonsHub+The+Apothecary+Diaries+S03+Multi+Subs+NF+WEB+DL+1080p",
),
),
(
"Season 2 EPs 01-12",
"http://dmhy.org/topics/rss/rss.xml?keyword=ANi+葬送的芙莉莲+Season+2+EPs+01+12+1080P",
"http://dmhy.org/topics/rss/rss.xml?keyword=ANi+葬送的芙莉莲+Season+2+1080P",
),
("S03", NYAA.replace("%s", "Show+S03+1080p"), None),
(
"S03E02",
"https://mikanani.me/RSS/Bangumi?bangumiId=3519&subgroupid=583",
None,
),
],
ids=["nyaa-episode", "dmhy-multi-token", "no-episode", "mikan-bangumi-feed"],
)
def test_strips_episode_from_search_subscription_once(
self, monkeypatch, season_raw, url, expected
):
"""旧版本搜索订阅关键词里的集号在启动时去掉一次,非搜索 URL 不变(#1138)。"""
from sqlmodel import Session
from module.models import Bangumi, RSSItem
from test.factories import make_bangumi
monkeypatch.setattr(
"module.searcher.provider.get_provider", lambda: self.PROVIDERS
)
engine = create_engine("sqlite://")
SQLModel.metadata.create_all(engine)
with engine.begin() as conn:
ensure_schema_version_table(conn)
set_schema_version(conn, CURRENT_SCHEMA_VERSION - 1)
other = "https://mikanani.me/RSS/Bangumi?bangumiId=1"
with Session(engine) as session:
session.add(RSSItem(id=1, url=url))
session.add(make_bangumi(season_raw=season_raw, rss_link=f"{other},{url}"))
session.commit()
run_migrations(engine)
with engine.begin() as conn:
assert get_schema_version(conn) == CURRENT_SCHEMA_VERSION
# 再跑一次不再改动
migration_module._strip_search_keyword_episodes(conn)
with Session(engine) as session:
rss = session.get(RSSItem, 1)
bangumi = session.exec(select(Bangumi)).one()
assert rss is not None
assert rss.url == (expected or url)
assert bangumi.rss_link == f"{other},{expected or url}"

View File

@@ -1,5 +1,6 @@
"""Tests for Renamer: gen_path, single-file / collection rename, rename flow."""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -7,7 +8,7 @@ import pytest
from module.conf import settings
from module.downloader import DownloadClient, RenameOutcome, RenameResult
from module.manager.renamer import PreparedMediaRename, Renamer
from module.models import EpisodeFile, Notification, SubtitleFile
from module.models import EpisodeFile, Notification, RenameOperation, SubtitleFile
# ---------------------------------------------------------------------------
# gen_path
@@ -517,16 +518,8 @@ class TestRenameSingleFile:
assert second == []
renamer.client.client.torrents_rename_file.assert_not_awaited()
@pytest.mark.parametrize(
("method", "notified"),
[("pn", True), ("not-installed", False)],
ids=["configured", "fallback"],
)
async def test_unchanged_path_notifies_once_for_configured_strategy(
self, renamer, method, notified
):
"""文件名已是目标名时首轮通知一次;选定策略未登记、本轮回退到 none 时
文件名不是最终名,不通知(#1082)。"""
async def test_unchanged_path_notifies_once_for_configured_strategy(self, renamer):
"""文件名已是目标名时首轮通知一次(#1082)。"""
ep = EpisodeFile(
media_path="My Anime S01E05.mkv",
title="My Anime",
@@ -539,16 +532,93 @@ class TestRenameSingleFile:
renamer,
ep,
media_path="My Anime S01E05.mkv",
method=method,
_hash=f"hash-unchanged-{method}",
_hash="hash-unchanged",
)
for _ in range(2)
]
expected = [Notification(official_title="My Anime", season=1, episode=5)]
assert loops == [expected if notified else [], []]
assert loops == [expected, []]
renamer.client.client.torrents_rename_file.assert_not_awaited()
async def test_unchanged_path_fallback_round_defers_record_and_notification(
self, renamer
):
"""选定策略未登记、本轮回退到 none 时原名不是最终名:不建操作记录也不
通知;策略恢复后那一轮文件名仍未变,也要通知一次(#1082)。"""
from module.database import Database
ep = EpisodeFile(
media_path="My Anime S01E05.mkv",
title="My Anime",
season=1,
episode=5,
suffix=".mkv",
)
async def loop(method: str) -> list[Notification]:
return await self._rename(
renamer,
ep,
media_path="My Anime S01E05.mkv",
method=method,
_hash="hash-fallback",
)
async def operation() -> RenameOperation | None:
async with Database() as db:
return await db.rename_operation.get_by_target(
downloader_type=renamer._downloader_type(),
save_path=self.SAVE_PATH,
target_path="My Anime S01E05.mkv",
active_only=False,
)
assert await loop("not-installed") == []
assert await operation() is None
assert [await loop("pn") for _ in range(2)] == [
[Notification(official_title="My Anime", season=1, episode=5)],
[],
]
row = await operation()
assert row is not None and row.state == "done"
renamer.client.client.torrents_rename_file.assert_not_awaited()
async def test_unchanged_path_done_row_pruned_only_after_task_removed(
self, renamer
):
"""路径未变的 done 记录过期后,任务仍在时保留且不再通知;任务从下载器
删除后清理,表不会随已删除的种子无限增长(#1082)。"""
from module.database import Database
ep = EpisodeFile(
media_path="old.mkv", title="My Anime", season=1, episode=5, suffix=".mkv"
)
assert await self._rename(renamer, ep, method="none", _hash="hash-gone")
async with Database() as db:
row = await db.rename_operation.get_by_target(
downloader_type=renamer._downloader_type(),
save_path=self.SAVE_PATH,
target_path="old.mkv",
active_only=False,
)
assert row is not None
row_id = row.id
row.updated_at = datetime.now(timezone.utc) - timedelta(days=31)
db.session.add(row)
await db.session.commit()
assert await self._rename(renamer, ep, method="none", _hash="hash-gone") == []
async with Database() as db:
assert await db.rename_operation.get(row_id) is not None
# 下载器里只剩另一个任务:原任务已删除
await self._rename(
renamer, ep, media_path="other.mkv", method="none", _hash="hash-other"
)
async with Database() as db:
assert await db.rename_operation.get(row_id) is None
@pytest.mark.parametrize("remove_bad_torrent", [False, True])
async def test_parse_fails(self, renamer, remove_bad_torrent):
"""解析失败时不返回通知;仅 remove_bad_torrent=True 时删除种子。"""