From a7823fb4d1e04aaf0274795ab6420bfd298b73c8 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 9 May 2026 14:04:10 +0800 Subject: [PATCH] feat: implement data cleanup chain for batch deletion of expired records --- app/chain/data_cleanup.py | 160 +++++++++++++++++++++++++++ app/db/models/downloadhistory.py | 62 ++++++++++- app/db/models/message.py | 29 ++++- app/db/models/siteuserdata.py | 30 +++++- app/db/models/transferhistory.py | 27 +++++ app/scheduler.py | 17 +++ tests/test_data_cleanup_chain.py | 180 +++++++++++++++++++++++++++++++ 7 files changed, 502 insertions(+), 3 deletions(-) create mode 100644 app/chain/data_cleanup.py create mode 100644 tests/test_data_cleanup_chain.py diff --git a/app/chain/data_cleanup.py b/app/chain/data_cleanup.py new file mode 100644 index 00000000..69f15dca --- /dev/null +++ b/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, + } diff --git a/app/db/models/downloadhistory.py b/app/db/models/downloadhistory.py index bc7465c6..4b9d2f17 100644 --- a/app/db/models/downloadhistory.py +++ b/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) + ) diff --git a/app/db/models/message.py b/app/db/models/message.py index 6fcad3de..67d5c21d 100644 --- a/app/db/models/message.py +++ b/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) + ) diff --git a/app/db/models/siteuserdata.py b/app/db/models/siteuserdata.py index 451e182a..c3d1f58f 100644 --- a/app/db/models/siteuserdata.py +++ b/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 diff --git a/app/db/models/transferhistory.py b/app/db/models/transferhistory.py index 325bb8ff..2f94f4f6 100644 --- a/app/db/models/transferhistory.py +++ b/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) + ) diff --git a/app/scheduler.py b/app/scheduler.py index 48e86461..74a60472 100644 --- a/app/scheduler.py +++ b/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, diff --git a/tests/test_data_cleanup_chain.py b/tests/test_data_cleanup_chain.py new file mode 100644 index 00000000..c2b4ecea --- /dev/null +++ b/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)