refactor(sdk): isolate plugin query persistence

app/db/oper/downloadhistory.py

@@ -1,10 +1,25 @@
 from typing import Dict, List, Optional, cast
 
-from sqlalchemy import delete as sqlalchemy_delete, update as sqlalchemy_update
+from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, select
+from sqlalchemy import update as sqlalchemy_update
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
-from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
+from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    literal_contains,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    QueryPageRequest,
+    QuerySortField,
+)
 from app.schemas.types import MediaSource
 
 
@@ -13,6 +28,95 @@ class DownloadHistoryOper(DbOper):
     下载历史管理
     """
 
+    def get_by_id(self, record_id: int) -> Optional[DownloadHistory]:
+        """按稳定记录 ID 读取单条下载历史。"""
+        return cast(
+            Optional[DownloadHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(DownloadHistory).where(DownloadHistory.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: DownloadHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[DownloadHistory], int]:
+        """按稳定筛选和分页合同读取下载历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[DownloadHistory], int]:
+            """在同一会话中构造并执行下载历史 count/page 查询。"""
+            conditions = media_identity_conditions(DownloadHistory, filters)
+            ids = enum_values(filters.ids)
+            media_types = enum_values(filters.media_types)
+            usernames = enum_values(filters.usernames)
+            if ids:
+                conditions.append(DownloadHistory.id.in_(ids))
+            if media_types:
+                conditions.append(DownloadHistory.type.in_(media_types))
+            for column, value in (
+                (DownloadHistory.title, filters.title),
+                (DownloadHistory.year, filters.year),
+                (DownloadHistory.seasons, filters.seasons),
+                (DownloadHistory.episodes, filters.episodes),
+                (DownloadHistory.path, filters.path),
+                (DownloadHistory.download_hash, filters.download_hash),
+                (DownloadHistory.username, filters.username),
+                (DownloadHistory.episode_group, filters.episode_group),
+            ):
+                if value is not None and value != "":
+                    conditions.append(column == value)
+            if filters.text:
+                conditions.append(
+                    literal_contains(DownloadHistory.title, filters.text)
+                    | literal_contains(DownloadHistory.path, filters.text)
+                )
+            if usernames:
+                conditions.append(DownloadHistory.username.in_(usernames))
+            music_condition = music_type_condition(
+                DownloadHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(DownloadHistory.id))
+            page_statement = select(DownloadHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    DownloadHistory.id.desc()
+                    if descending_order
+                    else DownloadHistory.id.asc()
+                )
+                secondary = (
+                    DownloadHistory.date.desc()
+                    if descending_order
+                    else DownloadHistory.date.asc()
+                )
+            else:
+                primary = (
+                    DownloadHistory.date.desc().nullslast()
+                    if descending_order
+                    else DownloadHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    DownloadHistory.id.desc()
+                    if descending_order
+                    else DownloadHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[DownloadHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     def get_by_path(self, path: str) -> Optional[DownloadHistory]:
         """
         按路径查询下载记录

app/db/oper/query.py

@@ -0,0 +1,113 @@
+"""只读查询 Oper 共享的持久化原语。
+
+本模块只承载跨表完全相同的查询表示规则和分页执行步骤。每个具体 Oper 仍负责
+声明本表模型、筛选字段、SQLAlchemy 条件、count/page 语句以及排序，避免这里退化
+成任意表的 Repository。
+"""
+
+from collections.abc import Iterable
+from enum import Enum
+from typing import Any, TypeVar
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from app.db.base import Base
+from app.schemas.query import QueryPageRequest, QuerySortDirection
+from app.schemas.types import MEDIA_SOURCE_IDENTIFIER_PATTERN, MUSIC_ENTITY_RECORDING
+
+ModelT = TypeVar("ModelT", bound=Base)
+
+
+def enum_value(value: Any) -> Any:
+    """返回筛选枚举对应的数据库值。"""
+    return value.value if isinstance(value, Enum) else value
+
+
+def enum_values(values: Iterable[Any]) -> tuple[Any, ...]:
+    """去除空筛选值、归一枚举，并保留调用方顺序。"""
+    normalized: list[Any] = []
+    for value in values:
+        value = enum_value(value)
+        if value in (None, ""):
+            continue
+        normalized.append(value)
+    return tuple(dict.fromkeys(normalized))
+
+
+def literal_contains(column: Any, value: str) -> Any:
+    """构造不区分大小写且不解释 ``%``、``_`` 通配符的字面包含条件。"""
+    escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    return column.ilike(f"%{escaped}%", escape="\\")
+
+
+def media_identity_conditions(model: Any, query: Any) -> list[Any]:
+    """按媒体来源与原生 ID 的成对合同构造条件，并对非法身份 fail-closed。"""
+    media_source = query.media_source
+    media_id = query.media_id
+    if (media_source is None) != (media_id is None):
+        raise ValueError("media_source 和 media_id 必须同时提供")
+    if media_source is None:
+        return []
+    normalized_id = str(media_id).strip()
+    if not normalized_id or normalized_id == "0":
+        raise ValueError("media_id 必须是非零的来源原生 ID")
+    return [
+        model.media_source == enum_value(media_source),
+        model.media_id == normalized_id,
+    ]
+
+
+def required_media_identity_conditions(model: Any) -> list[Any]:
+    """构造只保留可解析来源与非空、非零原生 ID 的条件。"""
+    return [
+        model.media_source.is_not(None),
+        func.trim(model.media_source) != "",
+        func.lower(func.trim(model.media_source)).regexp_match(
+            MEDIA_SOURCE_IDENTIFIER_PATTERN
+        ),
+        model.media_id.is_not(None),
+        func.trim(model.media_id) != "",
+        func.trim(model.media_id) != "0",
+    ]
+
+
+def music_type_condition(column: Any, music_type: str | None) -> Any | None:
+    """兼容未标注音乐类型的历史单曲记录。"""
+    music_type = enum_value(music_type)
+    if not music_type:
+        return None
+    if music_type == MUSIC_ENTITY_RECORDING:
+        return or_(column == music_type, column.is_(None))
+    return column == music_type
+
+
+def execute_page(
+    session: Session,
+    count_statement: Any,
+    page_statement: Any,
+    page: QueryPageRequest,
+) -> tuple[list[ModelT], int]:
+    """在同一同步 Session 内先统计再读取一页已构造的查询语句。"""
+    total = int(session.execute(count_statement).scalar_one() or 0)
+    records = list(
+        session.execute(page_statement.offset((page.page - 1) * page.count).limit(page.count)).scalars().all()
+    )
+    return records, total
+
+
+def descending(page: QueryPageRequest) -> bool:
+    """返回公开分页排序是否为降序，集中处理枚举表示。"""
+    return page.sort.direction == QuerySortDirection.DESC
+
+
+__all__ = [
+    "descending",
+    "enum_value",
+    "enum_values",
+    "execute_page",
+    "literal_contains",
+    "media_identity_conditions",
+    "music_type_condition",
+    "required_media_identity_conditions",
+]

app/db/oper/subscribe.py

@@ -9,9 +9,10 @@
 """
 import time
 from collections.abc import Awaitable, Callable
-from typing import Any, Tuple, List, Optional
+from typing import Any, List, Optional, Tuple, cast
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
@@ -19,6 +20,14 @@ from app.application.subscription.delete import SubscribeDeletionCandidate
 from app.db.base import DbOper
 from app.db.models.subscribe import Subscribe
 from app.db.models.subscribehistory import SubscribeHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import QueryPageRequest, QuerySortField, SubscriptionFilter
 from app.schemas.types import MediaSource
 
 INTEGER_FLAG_FIELDS = ("best_version", "best_version_full", "search_imdbid", "manual_total_episode")
@@ -276,7 +285,77 @@ class SubscribeOper(DbOper):
         """
         获取订阅
         """
-        return self._execute_sync_query(lambda session: Subscribe.get(session, sid))
+        return self.get_by_id(sid)
+
+    def get_by_id(self, record_id: int) -> Optional[Subscribe]:
+        """按稳定记录 ID 读取单条订阅。"""
+        return cast(
+            Optional[Subscribe],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(Subscribe).where(Subscribe.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: SubscriptionFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[Subscribe], int]:
+        """按稳定筛选和分页合同读取订阅记录及总数。"""
+        def execute(session: Session) -> tuple[list[Subscribe], int]:
+            """在同一会话中构造并执行订阅 count/page 查询。"""
+            conditions = media_identity_conditions(Subscribe, filters)
+            ids = enum_values(filters.ids)
+            names = enum_values(filters.names)
+            states = enum_values(filters.states)
+            usernames = enum_values(filters.usernames)
+            media_types = enum_values(filters.media_types)
+            if ids:
+                conditions.append(Subscribe.id.in_(ids))
+            if names:
+                conditions.append(Subscribe.name.in_(names))
+            if states:
+                conditions.append(Subscribe.state.in_(states))
+            if usernames:
+                conditions.append(Subscribe.username.in_(usernames))
+            if media_types:
+                conditions.append(Subscribe.type.in_(media_types))
+            if filters.season is not None:
+                conditions.append(Subscribe.season == filters.season)
+            if filters.episode_group is not None:
+                conditions.append(Subscribe.episode_group == filters.episode_group)
+            music_condition = music_type_condition(
+                Subscribe.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(Subscribe.id))
+            page_statement = select(Subscribe)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = Subscribe.id.desc() if descending_order else Subscribe.id.asc()
+                secondary = Subscribe.date.desc() if descending_order else Subscribe.date.asc()
+            else:
+                primary = (
+                    Subscribe.date.desc().nullslast()
+                    if descending_order
+                    else Subscribe.date.asc().nullsfirst()
+                )
+                secondary = Subscribe.id.desc() if descending_order else Subscribe.id.asc()
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[Subscribe], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
 
     async def async_get(self, sid: int) -> Optional[Subscribe]:
         """

app/db/oper/subscribehistory.py

@@ -1,10 +1,23 @@
-from typing import List, Optional
+from typing import List, Optional, cast
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.subscribehistory import SubscribeHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import (
+    QueryPageRequest,
+    QuerySortField,
+    SubscriptionHistoryFilter,
+)
 
 
 class SubscribeHistoryOper(DbOper):
@@ -12,6 +25,87 @@ class SubscribeHistoryOper(DbOper):
     订阅历史管理。
     """
 
+    def get_by_id(self, record_id: int) -> Optional[SubscribeHistory]:
+        """按稳定记录 ID 读取单条订阅历史。"""
+        return cast(
+            Optional[SubscribeHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(SubscribeHistory).where(SubscribeHistory.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: SubscriptionHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[SubscribeHistory], int]:
+        """按稳定筛选和分页合同读取订阅历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[SubscribeHistory], int]:
+            """在同一会话中构造并执行订阅历史 count/page 查询。"""
+            conditions = media_identity_conditions(SubscribeHistory, filters)
+            ids = enum_values(filters.ids)
+            names = enum_values(filters.names)
+            usernames = enum_values(filters.usernames)
+            media_types = enum_values(filters.media_types)
+            if ids:
+                conditions.append(SubscribeHistory.id.in_(ids))
+            if names:
+                conditions.append(SubscribeHistory.name.in_(names))
+            if usernames:
+                conditions.append(SubscribeHistory.username.in_(usernames))
+            if media_types:
+                conditions.append(SubscribeHistory.type.in_(media_types))
+            if filters.season is not None:
+                conditions.append(SubscribeHistory.season == filters.season)
+            if filters.episode_group is not None:
+                conditions.append(
+                    SubscribeHistory.episode_group == filters.episode_group
+                )
+            music_condition = music_type_condition(
+                SubscribeHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(SubscribeHistory.id))
+            page_statement = select(SubscribeHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    SubscribeHistory.id.desc()
+                    if descending_order
+                    else SubscribeHistory.id.asc()
+                )
+                secondary = (
+                    SubscribeHistory.date.desc()
+                    if descending_order
+                    else SubscribeHistory.date.asc()
+                )
+            else:
+                primary = (
+                    SubscribeHistory.date.desc().nullslast()
+                    if descending_order
+                    else SubscribeHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    SubscribeHistory.id.desc()
+                    if descending_order
+                    else SubscribeHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[SubscribeHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     async def async_list_by_type(
         self,
         mtype: str,

app/db/oper/transferhistory.py

@@ -1,12 +1,27 @@
 import time
-from typing import Any, List, Optional
+from typing import Any, List, Optional, cast
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.transferhistory import TransferHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    literal_contains,
+    media_identity_conditions,
+    music_type_condition,
+    required_media_identity_conditions,
+)
+from app.schemas.query import (
+    QueryPageRequest,
+    QuerySortField,
+    TransferHistoryFilter,
+)
 from app.schemas.types import MediaSource
 
 
@@ -20,10 +35,111 @@ class TransferHistoryOper(DbOper):
         获取转移历史
         :param historyid: 转移历史id
         """
-        return self._execute_sync_query(
-            lambda session: TransferHistory.get(session, historyid)
+        return self.get_by_id(historyid)
+
+    def get_by_id(self, record_id: int) -> Optional[TransferHistory]:
+        """按稳定记录 ID 读取单条整理历史。"""
+        return cast(
+            Optional[TransferHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(TransferHistory).where(TransferHistory.id == record_id)
+                ).scalars().first()
+            ),
         )
 
+    def query(
+        self,
+        filters: TransferHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[TransferHistory], int]:
+        """按稳定筛选和分页合同读取整理历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[TransferHistory], int]:
+            """在同一会话中构造并执行整理历史 count/page 查询。"""
+            conditions = media_identity_conditions(TransferHistory, filters)
+            ids = enum_values(filters.ids)
+            media_types = enum_values(filters.media_types)
+            media_sources = enum_values(filters.media_sources)
+            if ids:
+                conditions.append(TransferHistory.id.in_(ids))
+            if media_types:
+                conditions.append(TransferHistory.type.in_(media_types))
+            if media_sources:
+                conditions.append(TransferHistory.media_source.in_(media_sources))
+            if filters.require_media_identity:
+                conditions.extend(required_media_identity_conditions(TransferHistory))
+            if filters.title:
+                conditions.append(TransferHistory.title == filters.title)
+            if filters.text:
+                conditions.append(
+                    literal_contains(TransferHistory.title, filters.text)
+                    | literal_contains(TransferHistory.src, filters.text)
+                    | literal_contains(TransferHistory.dest, filters.text)
+                )
+            for column, value in (
+                (TransferHistory.year, filters.year),
+                (TransferHistory.seasons, filters.seasons),
+                (TransferHistory.episodes, filters.episodes),
+                (TransferHistory.src, filters.src),
+                (TransferHistory.dest, filters.dest),
+                (TransferHistory.download_hash, filters.download_hash),
+                (TransferHistory.episode_group, filters.episode_group),
+            ):
+                if value is not None and value != "":
+                    conditions.append(column == value)
+            if filters.status is not None:
+                if filters.status:
+                    conditions.append(TransferHistory.status.is_(True))
+                else:
+                    conditions.append(
+                        or_(
+                            TransferHistory.status.is_(False),
+                            TransferHistory.status.is_(None),
+                        )
+                    )
+            music_condition = music_type_condition(
+                TransferHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(TransferHistory.id))
+            page_statement = select(TransferHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    TransferHistory.id.desc()
+                    if descending_order
+                    else TransferHistory.id.asc()
+                )
+                secondary = (
+                    TransferHistory.date.desc()
+                    if descending_order
+                    else TransferHistory.date.asc()
+                )
+            else:
+                primary = (
+                    TransferHistory.date.desc().nullslast()
+                    if descending_order
+                    else TransferHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    TransferHistory.id.desc()
+                    if descending_order
+                    else TransferHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[TransferHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     async def async_get(self, historyid: int) -> Optional[TransferHistory]:
         """
         异步获取转移历史。