Fix history search case sensitivity

app/db/models/downloadhistory.py

@@ -8,6 +8,11 @@ from sqlalchemy.orm import Session
 from app.db import db_query, db_update, get_id_column, Base, async_db_query
 
 
+def _title_like(column, title: str):
+    """构造跨数据库大小写不敏感的标题匹配条件。"""
+    return column.ilike(f"%{title}%")
+
+
 class DownloadHistory(Base):
     """
     下载历史记录
@@ -148,7 +153,7 @@ class DownloadHistory(Base):
         count: Optional[int] = 30,
     ):
         query = (
-            select(cls).filter(cls.title.like(f"%{title}%")).order_by(cls.date.desc())
+            select(cls).filter(_title_like(cls.title, title)).order_by(cls.date.desc())
         )
         query = query.offset((page - 1) * count).limit(count)
         result = await db.execute(query)
@@ -164,7 +169,7 @@ class DownloadHistory(Base):
     @async_db_query
     async def async_count_by_title(cls, db: AsyncSession, title: str):
         result = await db.execute(
-            select(func.count(cls.id)).filter(cls.title.like(f"%{title}%"))
+            select(func.count(cls.id)).filter(_title_like(cls.title, title))
         )
         return result.scalar()
 

app/db/models/transferhistory.py

@@ -1,13 +1,20 @@
 import time
 from typing import Optional
 
-from sqlalchemy import Column, Integer, String, Boolean, Index, func, or_, JSON, select
+from sqlalchemy import Boolean, Column, Index, Integer, JSON, String, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.db import db_query, db_update, get_id_column, Base, async_db_query
 
 
+def _text_like(column, pattern: str, wildcard: bool = False):
+    """构造跨数据库大小写不敏感的文本匹配条件。"""
+    if wildcard:
+        return column.ilike(pattern, escape='\\')
+    return column.ilike(pattern)
+
+
 class TransferHistory(Base):
     """
     整理记录
@@ -71,15 +78,15 @@ class TransferHistory(Base):
                       status: bool = None, wildcard: bool = False):
         if wildcard:
             text_filter = or_(
-                cls.title.like(title, escape='\\'),
-                cls.src.like(title, escape='\\'),
-                cls.dest.like(title, escape='\\'),
+                _text_like(cls.title, title, wildcard=True),
+                _text_like(cls.src, title, wildcard=True),
+                _text_like(cls.dest, title, wildcard=True),
             )
         else:
             text_filter = or_(
-                cls.title.like(f'%{title}%'),
-                cls.src.like(f'%{title}%'),
-                cls.dest.like(f'%{title}%'),
+                _text_like(cls.title, f'%{title}%'),
+                _text_like(cls.src, f'%{title}%'),
+                _text_like(cls.dest, f'%{title}%'),
             )
         query = db.query(cls).filter(text_filter)
         if status is not None:
@@ -98,15 +105,15 @@ class TransferHistory(Base):
                                   status: bool = None, wildcard: bool = False):
         if wildcard:
             text_filter = or_(
-                cls.title.like(title, escape='\\'),
-                cls.src.like(title, escape='\\'),
-                cls.dest.like(title, escape='\\'),
+                _text_like(cls.title, title, wildcard=True),
+                _text_like(cls.src, title, wildcard=True),
+                _text_like(cls.dest, title, wildcard=True),
             )
         else:
             text_filter = or_(
-                cls.title.like(f'%{title}%'),
-                cls.src.like(f'%{title}%'),
-                cls.dest.like(f'%{title}%'),
+                _text_like(cls.title, f'%{title}%'),
+                _text_like(cls.src, f'%{title}%'),
+                _text_like(cls.dest, f'%{title}%'),
             )
         query = select(cls).filter(text_filter)
         if status is not None:
@@ -239,15 +246,15 @@ class TransferHistory(Base):
     def count_by_title(cls, db: Session, title: str, status: bool = None, wildcard: bool = False):
         if wildcard:
             text_filter = or_(
-                cls.title.like(title, escape='\\'),
-                cls.src.like(title, escape='\\'),
-                cls.dest.like(title, escape='\\'),
+                _text_like(cls.title, title, wildcard=True),
+                _text_like(cls.src, title, wildcard=True),
+                _text_like(cls.dest, title, wildcard=True),
             )
         else:
             text_filter = or_(
-                cls.title.like(f'%{title}%'),
-                cls.src.like(f'%{title}%'),
-                cls.dest.like(f'%{title}%'),
+                _text_like(cls.title, f'%{title}%'),
+                _text_like(cls.src, f'%{title}%'),
+                _text_like(cls.dest, f'%{title}%'),
             )
         query = db.query(func.count(cls.id)).filter(text_filter)
         if status is not None:
@@ -259,15 +266,15 @@ class TransferHistory(Base):
     async def async_count_by_title(cls, db: AsyncSession, title: str, status: bool = None, wildcard: bool = False):
         if wildcard:
             text_filter = or_(
-                cls.title.like(title, escape='\\'),
-                cls.src.like(title, escape='\\'),
-                cls.dest.like(title, escape='\\'),
+                _text_like(cls.title, title, wildcard=True),
+                _text_like(cls.src, title, wildcard=True),
+                _text_like(cls.dest, title, wildcard=True),
             )
         else:
             text_filter = or_(
-                cls.title.like(f'%{title}%'),
-                cls.src.like(f'%{title}%'),
-                cls.dest.like(f'%{title}%'),
+                _text_like(cls.title, f'%{title}%'),
+                _text_like(cls.src, f'%{title}%'),
+                _text_like(cls.dest, f'%{title}%'),
             )
         stmt = select(func.count(cls.id)).filter(text_filter)
         if status is not None:

tests/test_history_search.py

@@ -0,0 +1,100 @@
+import asyncio
+from pathlib import Path
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.db import Base
+from app.db.models.downloadhistory import DownloadHistory
+from app.db.models.transferhistory import TransferHistory
+
+
+def _enable_case_sensitive_like(db):
+    """让 SQLite 在测试中暴露与 PostgreSQL 一致的 LIKE 大小写敏感问题。"""
+    db.execute(text("PRAGMA case_sensitive_like=ON"))
+
+
+def test_transfer_history_search_is_case_insensitive(tmp_path: Path):
+    """整理历史标题和路径搜索应忽略大小写。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'transfer_history.db'}")
+    SessionFactory = sessionmaker(bind=engine)
+    Base.metadata.create_all(bind=engine)
+
+    try:
+        with SessionFactory() as db:
+            _enable_case_sensitive_like(db)
+            db.add_all(
+                [
+                    TransferHistory(
+                        src="/downloads/Avatar.Source.mkv",
+                        dest="/media/Avatar/Avatar.mkv",
+                        title="Avatar",
+                        status=True,
+                        date="2026-06-01 00:00:00",
+                    ),
+                    TransferHistory(
+                        src="/downloads/Interstellar.mkv",
+                        dest="/media/Interstellar/Interstellar.mkv",
+                        title="Interstellar",
+                        status=True,
+                        date="2026-06-02 00:00:00",
+                    ),
+                ]
+            )
+            db.commit()
+
+            title_result = TransferHistory.list_by_title(db, "avatar")
+            src_result = TransferHistory.list_by_title(db, "avatar.source")
+            dest_result = TransferHistory.list_by_title(db, "%avatar.mkv", wildcard=True)
+            total = TransferHistory.count_by_title(db, "avatar")
+
+        assert [item.title for item in title_result] == ["Avatar"]
+        assert [item.title for item in src_result] == ["Avatar"]
+        assert [item.title for item in dest_result] == ["Avatar"]
+        assert total == 1
+    finally:
+        engine.dispose()
+
+
+def test_download_history_title_search_is_case_insensitive(tmp_path: Path):
+    """下载历史标题搜索和计数应忽略大小写。"""
+
+    async def run_case():
+        """执行异步下载历史查询断言。"""
+        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'download_history.db'}")
+        SessionFactory = async_sessionmaker(bind=engine)
+
+        try:
+            async with engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+                await conn.execute(text("PRAGMA case_sensitive_like=ON"))
+
+            async with SessionFactory() as db:
+                db.add_all(
+                    [
+                        DownloadHistory(
+                            path="/downloads/Avatar",
+                            type="电影",
+                            title="Avatar",
+                            date="2026-06-01 00:00:00",
+                        ),
+                        DownloadHistory(
+                            path="/downloads/Interstellar",
+                            type="电影",
+                            title="Interstellar",
+                            date="2026-06-02 00:00:00",
+                        ),
+                    ]
+                )
+                await db.commit()
+
+                result = await DownloadHistory.async_list_by_title(db, "avatar")
+                total = await DownloadHistory.async_count_by_title(db, "avatar")
+
+            assert [item.title for item in result] == ["Avatar"]
+            assert total == 1
+        finally:
+            await engine.dispose()
+
+    asyncio.run(run_case())