feat: implement data cleanup chain for batch deletion of expired records

app/chain/data_cleanup.py

@@ -0,0 +1,160 @@
+import json
+from datetime import datetime, timedelta
+from typing import Callable, Optional, Dict, Any
+
+from sqlalchemy.orm import Session
+
+from app.db import SessionFactory
+from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
+from app.db.models.message import Message
+from app.db.models.siteuserdata import SiteUserData
+from app.db.models.transferhistory import TransferHistory
+from app.log import logger
+
+
+class DataCleanupChain:
+    """
+    系统数据清理链。
+    """
+
+    DEFAULT_BATCH_SIZE = 500
+    MESSAGE_RETENTION_DAYS = 90
+    DOWNLOAD_HISTORY_RETENTION_DAYS = 180
+    SITE_USERDATA_RETENTION_DAYS = 180
+    TRANSFER_HISTORY_RETENTION_DAYS = 365 * 3
+
+    def cleanup(self, batch_size: Optional[int] = None) -> Dict[str, Any]:
+        """
+        按预设保留期执行分批清理。
+        """
+        started_at = datetime.now()
+        batch_size = batch_size or self.DEFAULT_BATCH_SIZE
+        if batch_size <= 0:
+            batch_size = self.DEFAULT_BATCH_SIZE
+
+        message_cutoff = (
+            started_at - timedelta(days=self.MESSAGE_RETENTION_DAYS)
+        ).strftime("%Y-%m-%d %H:%M:%S")
+        download_history_cutoff = (
+            started_at - timedelta(days=self.DOWNLOAD_HISTORY_RETENTION_DAYS)
+        ).strftime("%Y-%m-%d %H:%M:%S")
+        site_userdata_cutoff = (
+            started_at - timedelta(days=self.SITE_USERDATA_RETENTION_DAYS)
+        ).strftime("%Y-%m-%d")
+        transfer_history_cutoff = (
+            started_at - timedelta(days=self.TRANSFER_HISTORY_RETENTION_DAYS)
+        ).strftime("%Y-%m-%d %H:%M:%S")
+
+        report: Dict[str, Any] = {
+            "started_at": started_at.strftime("%Y-%m-%d %H:%M:%S"),
+            "batch_size": batch_size,
+            "tables": {},
+            "total_deleted": 0,
+        }
+        errors = []
+
+        plans = [
+            {
+                "name": "message",
+                "cutoff": message_cutoff,
+                "handler": lambda db: Message.delete_before(
+                    db=db,
+                    before_time=message_cutoff,
+                    limit=batch_size,
+                ),
+            },
+            {
+                "name": "downloadhistory",
+                "cutoff": download_history_cutoff,
+                "handler": lambda db: DownloadHistory.delete_before(
+                    db=db,
+                    before_time=download_history_cutoff,
+                    limit=batch_size,
+                ),
+            },
+            {
+                "name": "downloadfiles",
+                "cutoff": "follow-parent-history",
+                "handler": lambda db: DownloadFiles.delete_orphans(
+                    db=db,
+                    limit=batch_size,
+                ),
+            },
+            {
+                "name": "siteuserdata",
+                "cutoff": site_userdata_cutoff,
+                "handler": lambda db: SiteUserData.delete_before(
+                    db=db,
+                    before_day=site_userdata_cutoff,
+                    limit=batch_size,
+                ),
+            },
+            {
+                "name": "transferhistory",
+                "cutoff": transfer_history_cutoff,
+                "handler": lambda db: TransferHistory.delete_before(
+                    db=db,
+                    before_time=transfer_history_cutoff,
+                    limit=batch_size,
+                ),
+            },
+        ]
+
+        with SessionFactory() as db:
+            for plan in plans:
+                name = plan["name"]
+                try:
+                    table_report = self._cleanup_in_batches(
+                        db=db,
+                        table_name=name,
+                        delete_batch=plan["handler"],
+                    )
+                    table_report["cutoff"] = plan["cutoff"]
+                    report["tables"][name] = table_report
+                    report["total_deleted"] += table_report["deleted"]
+                except Exception as err:
+                    errors.append(f"{name}: {str(err)}")
+                    logger.error(f"数据表 {name} 清理失败：{str(err)}")
+                    report["tables"][name] = {
+                        "deleted": 0,
+                        "batches": 0,
+                        "cutoff": plan["cutoff"],
+                        "error": str(err),
+                    }
+
+        if errors:
+            report["errors"] = errors
+            logger.error(
+                f"数据表清理部分失败：{json.dumps(report, ensure_ascii=False)}"
+            )
+            raise RuntimeError("；".join(errors))
+
+        logger.info(f"数据表清理完成：{json.dumps(report, ensure_ascii=False)}")
+        return report
+
+    @staticmethod
+    def _cleanup_in_batches(
+        db: Session,
+        table_name: str,
+        delete_batch: Callable[[Session], int],
+    ) -> Dict[str, int]:
+        """
+        循环执行单表分批删除，直到没有可删除数据。
+        """
+        total_deleted = 0
+        batches = 0
+
+        while True:
+            deleted = delete_batch(db) or 0
+            if deleted <= 0:
+                break
+            batches += 1
+            total_deleted += deleted
+            logger.info(
+                f"数据表 {table_name} 清理第 {batches} 批完成，删除 {deleted} 条记录"
+            )
+
+        return {
+            "deleted": total_deleted,
+            "batches": batches,
+        }

app/db/models/downloadhistory.py

@@ -1,7 +1,7 @@
 import time
 from typing import List, Optional
 
-from sqlalchemy import Column, Integer, String, JSON, select
+from sqlalchemy import Column, Integer, String, JSON, select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
@@ -336,6 +336,33 @@ class DownloadHistory(Base):
             .all()
         )
 
+    @classmethod
+    @db_update
+    def delete_before(
+        cls,
+        db: Session,
+        before_time: str,
+        limit: Optional[int] = 500,
+    ) -> int:
+        """
+        分批删除指定时间之前的下载历史。
+        """
+        ids = [
+            row[0]
+            for row in db.query(cls.id)
+            .filter(cls.date < before_time)
+            .order_by(cls.id.asc())
+            .limit(limit)
+            .all()
+        ]
+        if not ids:
+            return 0
+        return (
+            db.query(cls)
+            .filter(cls.id.in_(ids))
+            .delete(synchronize_session=False)
+        )
+
 
 class DownloadFiles(Base):
     """
@@ -399,3 +426,36 @@ class DownloadFiles(Base):
         db.query(cls).filter(cls.fullpath == fullpath, cls.state == 1).update(
             {"state": 0}
         )
+
+    @classmethod
+    @db_update
+    def delete_orphans(
+        cls,
+        db: Session,
+        limit: Optional[int] = 500,
+    ) -> int:
+        """
+        分批删除已找不到父下载历史的文件记录。
+
+        downloadfiles 没有时间字段，无法安全地按时间直接裁剪，
+        因此只清理明确失去父记录的孤儿数据。
+        """
+        ids = [
+            row[0]
+            for row in db.query(cls.id)
+            .outerjoin(
+                DownloadHistory,
+                DownloadHistory.download_hash == cls.download_hash,
+            )
+            .filter(DownloadHistory.id.is_(None))
+            .order_by(cls.id.asc())
+            .limit(limit)
+            .all()
+        ]
+        if not ids:
+            return 0
+        return (
+            db.query(cls)
+            .filter(cls.id.in_(ids))
+            .delete(synchronize_session=False)
+        )

app/db/models/message.py

@@ -4,7 +4,7 @@ from sqlalchemy import Column, Integer, String, JSON, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
-from app.db import db_query, Base, get_id_column, async_db_query
+from app.db import db_query, db_update, Base, get_id_column, async_db_query
 
 
 class Message(Base):
@@ -47,3 +47,30 @@ class Message(Base):
             select(cls).order_by(cls.reg_time.desc()).offset((page - 1) * count).limit(count)
         )
         return result.scalars().all()
+
+    @classmethod
+    @db_update
+    def delete_before(
+        cls,
+        db: Session,
+        before_time: str,
+        limit: Optional[int] = 500,
+    ) -> int:
+        """
+        分批删除指定时间之前的消息记录。
+        """
+        ids = [
+            row[0]
+            for row in db.query(cls.id)
+            .filter(cls.reg_time < before_time)
+            .order_by(cls.id.asc())
+            .limit(limit)
+            .all()
+        ]
+        if not ids:
+            return 0
+        return (
+            db.query(cls)
+            .filter(cls.id.in_(ids))
+            .delete(synchronize_session=False)
+        )

app/db/models/siteuserdata.py

@@ -5,7 +5,7 @@ from sqlalchemy import Column, Integer, String, Float, JSON, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
-from app.db import db_query, Base, get_id_column, async_db_query
+from app.db import db_query, db_update, Base, get_id_column, async_db_query
 
 
 class SiteUserData(Base):
@@ -129,3 +129,31 @@ class SiteUserData(Base):
                 (cls.updated_day == subquery.c.latest_update_day)
             ).order_by(cls.updated_time.desc()))
         return result.scalars().all()
+
+    @classmethod
+    @db_update
+    def delete_before(
+        cls,
+        db: Session,
+        before_day: str,
+        limit: Optional[int] = 500,
+    ) -> int:
+        """
+        分批删除指定日期之前的站点用户快照。
+        """
+        ids = [
+            row[0]
+            for row in db.query(cls.id)
+            .filter(cls.updated_day < before_day)
+            .order_by(cls.id.asc())
+            .limit(limit)
+            .all()
+        ]
+        if not ids:
+            return 0
+        deleted = (
+            db.query(cls)
+            .filter(cls.id.in_(ids))
+            .delete(synchronize_session=False)
+        )
+        return deleted

app/db/models/transferhistory.py

@@ -344,3 +344,30 @@ class TransferHistory(Base):
         查询某时间之后的转移历史
         """
         return db.query(cls).filter(cls.date > date).order_by(cls.id.desc()).all()
+
+    @classmethod
+    @db_update
+    def delete_before(
+        cls,
+        db: Session,
+        before_time: str,
+        limit: Optional[int] = 500,
+    ) -> int:
+        """
+        分批删除指定时间之前的整理历史。
+        """
+        ids = [
+            row[0]
+            for row in db.query(cls.id)
+            .filter(cls.date < before_time)
+            .order_by(cls.id.asc())
+            .limit(limit)
+            .all()
+        ]
+        if not ids:
+            return 0
+        return (
+            db.query(cls)
+            .filter(cls.id.in_(ids))
+            .delete(synchronize_session=False)
+        )

app/scheduler.py

@@ -15,6 +15,7 @@ from apscheduler.triggers.cron import CronTrigger
 
 from app import schemas
 from app.chain import ChainBase
+from app.chain.data_cleanup import DataCleanupChain
 from app.chain.mediaserver import MediaServerChain
 from app.chain.recommend import RecommendChain
 from app.chain.site import SiteChain
@@ -145,6 +146,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                     "func": self.clear_cache,
                     "running": False,
                 },
+                "data_cleanup": {
+                    "name": "数据表清理",
+                    "func": DataCleanupChain().cleanup,
+                    "running": False,
+                },
                 "user_auth": {
                     "name": "用户认证检查",
                     "func": self.user_auth,
@@ -345,6 +351,17 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 kwargs={"job_id": "clear_cache"},
             )
 
+            # 数据表清理服务，每天凌晨执行一次
+            self._scheduler.add_job(
+                self.start,
+                "cron",
+                id="data_cleanup",
+                name="数据表清理",
+                hour=3,
+                minute=30,
+                kwargs={"job_id": "data_cleanup"},
+            )
+
             # 定时检查用户认证，每隔10分钟
             self._scheduler.add_job(
                 self.start,

tests/test_data_cleanup_chain.py

@@ -0,0 +1,180 @@
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+from pathlib import Path
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.chain.data_cleanup import DataCleanupChain
+from app.db import Base
+from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
+from app.db.models.message import Message
+from app.db.models.siteuserdata import SiteUserData
+from app.db.models.transferhistory import TransferHistory
+
+
+class DataCleanupChainTest(unittest.TestCase):
+    """
+    数据清理链测试。
+    """
+
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        db_path = Path(self.temp_dir.name) / "cleanup.db"
+        self.engine = create_engine(f"sqlite:///{db_path}")
+        self.SessionFactory = sessionmaker(bind=self.engine)
+        Base.metadata.create_all(bind=self.engine)
+
+    def tearDown(self):
+        self.engine.dispose()
+        self.temp_dir.cleanup()
+
+    def test_cleanup_removes_expired_rows_in_batches(self):
+        """
+        指定表应按保留期分批删除，并保留仍在有效期内的数据。
+        """
+        now = datetime.now()
+        old_message_time = (now - timedelta(days=120)).strftime("%Y-%m-%d %H:%M:%S")
+        keep_message_time = (now - timedelta(days=10)).strftime("%Y-%m-%d %H:%M:%S")
+        old_download_time = (now - timedelta(days=240)).strftime("%Y-%m-%d %H:%M:%S")
+        keep_download_time = (now - timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
+        old_site_day = (now - timedelta(days=240)).strftime("%Y-%m-%d")
+        keep_site_day = (now - timedelta(days=2)).strftime("%Y-%m-%d")
+        old_transfer_time = (now - timedelta(days=365 * 3 + 30)).strftime(
+            "%Y-%m-%d %H:%M:%S"
+        )
+        keep_transfer_time = (now - timedelta(days=30)).strftime(
+            "%Y-%m-%d %H:%M:%S"
+        )
+
+        with self.SessionFactory() as db:
+            db.add_all(
+                [
+                    Message(reg_time=old_message_time, title="old-1"),
+                    Message(reg_time=old_message_time, title="old-2"),
+                    Message(reg_time=old_message_time, title="old-3"),
+                    Message(reg_time=keep_message_time, title="keep"),
+                ]
+            )
+            db.add_all(
+                [
+                    DownloadHistory(
+                        path="/downloads/old-1",
+                        type="电影",
+                        title="old-1",
+                        download_hash="hash-old-1",
+                        date=old_download_time,
+                    ),
+                    DownloadHistory(
+                        path="/downloads/old-2",
+                        type="电影",
+                        title="old-2",
+                        download_hash="hash-old-2",
+                        date=old_download_time,
+                    ),
+                    DownloadHistory(
+                        path="/downloads/keep",
+                        type="电影",
+                        title="keep",
+                        download_hash="hash-keep",
+                        date=keep_download_time,
+                    ),
+                ]
+            )
+            db.add_all(
+                [
+                    DownloadFiles(
+                        download_hash="hash-old-1",
+                        fullpath="/downloads/old-1/file.mkv",
+                        savepath="/downloads/old-1",
+                        filepath="file.mkv",
+                    ),
+                    DownloadFiles(
+                        download_hash="hash-old-2",
+                        fullpath="/downloads/old-2/file.mkv",
+                        savepath="/downloads/old-2",
+                        filepath="file.mkv",
+                    ),
+                    DownloadFiles(
+                        download_hash="hash-keep",
+                        fullpath="/downloads/keep/file.mkv",
+                        savepath="/downloads/keep",
+                        filepath="file.mkv",
+                    ),
+                    DownloadFiles(
+                        download_hash="hash-orphan",
+                        fullpath="/downloads/orphan/file.mkv",
+                        savepath="/downloads/orphan",
+                        filepath="file.mkv",
+                    ),
+                ]
+            )
+            db.add_all(
+                [
+                    SiteUserData(domain="old-1", name="old-1", updated_day=old_site_day),
+                    SiteUserData(domain="old-2", name="old-2", updated_day=old_site_day),
+                    SiteUserData(domain="keep", name="keep", updated_day=keep_site_day),
+                ]
+            )
+            db.add_all(
+                [
+                    TransferHistory(
+                        src="/src/old",
+                        title="old",
+                        date=old_transfer_time,
+                    ),
+                    TransferHistory(
+                        src="/src/keep",
+                        title="keep",
+                        date=keep_transfer_time,
+                    ),
+                ]
+            )
+            db.commit()
+
+        with patch("app.chain.data_cleanup.SessionFactory", self.SessionFactory):
+            report = DataCleanupChain().cleanup(batch_size=1)
+
+        self.assertEqual(report["tables"]["message"]["deleted"], 3)
+        self.assertEqual(report["tables"]["message"]["batches"], 3)
+        self.assertEqual(report["tables"]["downloadhistory"]["deleted"], 2)
+        self.assertEqual(report["tables"]["downloadfiles"]["deleted"], 3)
+        self.assertEqual(report["tables"]["siteuserdata"]["deleted"], 2)
+        self.assertEqual(report["tables"]["transferhistory"]["deleted"], 1)
+
+        with self.SessionFactory() as db:
+            self.assertEqual(db.query(Message).count(), 1)
+            self.assertEqual(db.query(DownloadHistory).count(), 1)
+            self.assertEqual(db.query(DownloadFiles).count(), 1)
+            self.assertEqual(db.query(SiteUserData).count(), 1)
+            self.assertEqual(db.query(TransferHistory).count(), 1)
+
+            keep_download_file = db.query(DownloadFiles).first()
+            self.assertEqual(keep_download_file.download_hash, "hash-keep")
+
+    def test_transferhistory_keeps_boundary_records(self):
+        """
+        恰好位于保留边界上的整理历史不应被提前清理。
+        """
+        now = datetime.now()
+        cutoff_time = (now - timedelta(days=365 * 3)).strftime("%Y-%m-%d %H:%M:%S")
+
+        with self.SessionFactory() as db:
+            db.add(
+                TransferHistory(
+                    src="/src/boundary",
+                    title="boundary",
+                    date=cutoff_time,
+                )
+            )
+            db.commit()
+
+        with patch("app.chain.data_cleanup.SessionFactory", self.SessionFactory):
+            report = DataCleanupChain().cleanup(batch_size=10)
+
+        self.assertEqual(report["tables"]["transferhistory"]["deleted"], 0)
+
+        with self.SessionFactory() as db:
+            self.assertEqual(db.query(TransferHistory).count(), 1)