dsh/dual_camera_system/core/file_cleanup.py

"""本地图片文件清理工具."""
import logging
import os
import threading
import time
from pathlib import Path
from typing import Callable, List, Optional

logger = logging.getLogger(__name__)


def _file_mtime(path: Path) -> float:
    """获取文件修改时间，失败返回 0。"""
    try:
        return path.stat().st_mtime
    except OSError:
        return 0.0


def cleanup_directory(
    directory: str,
    retention_days: int,
    max_files: Optional[int] = None,
    pattern: str = "*",
    now: Optional[float] = None,
) -> int:
    """
    清理目录中的过期文件。

    规则：
    1. 删除修改时间超过 retention_days 的文件。
    2. 如果指定 max_files 且文件总数超过，删除最旧的文件直到数量符合限制。

    Args:
        directory: 要清理的目录。
        retention_days: 保留天数。
        max_files: 最大文件数，None 表示不限制。
        pattern: 匹配模式，默认匹配所有文件。
        now: 当前时间戳，默认 time.time()。

    Returns:
        删除的文件数量。
    """
    if retention_days <= 0 and max_files is None:
        return 0

    dir_path = Path(directory)
    if not dir_path.exists():
        return 0

    now = now or time.time()
    cutoff = now - retention_days * 86400.0

    try:
        files = [p for p in dir_path.rglob(pattern) if p.is_file()]
    except OSError as exc:
        logger.warning("[cleanup] 扫描目录失败: %s, %s", directory, exc)
        return 0

    deleted = 0
    for p in files:
        mtime = _file_mtime(p)
        if retention_days > 0 and mtime < cutoff:
            try:
                p.unlink()
                deleted += 1
                logger.debug("[cleanup] 删除过期文件: %s", p)
            except OSError as exc:
                logger.warning("[cleanup] 删除文件失败: %s, %s", p, exc)

    # 如果仍然超过最大数量，按修改时间删除最旧的
    if max_files is not None and max_files > 0:
        try:
            remaining = sorted(
                [p for p in dir_path.rglob(pattern) if p.is_file()],
                key=_file_mtime,
            )
        except OSError:
            remaining = []
        while len(remaining) > max_files:
            oldest = remaining.pop(0)
            try:
                oldest.unlink()
                deleted += 1
                logger.debug("[cleanup] 删除超量旧文件: %s", oldest)
            except OSError as exc:
                logger.warning("[cleanup] 删除文件失败: %s, %s", oldest, exc)
                break

    if deleted:
        logger.info("[cleanup] %s 清理完成，删除 %d 个文件", directory, deleted)
    return deleted


class CleanupWorker:
    """定时清理 worker，每个实例管理一个目录。"""

    def __init__(
        self,
        directory: str,
        retention_days: int,
        max_files: Optional[int] = None,
        interval_seconds: float = 3600.0,
        pattern: str = "*",
    ):
        self.directory = directory
        self.retention_days = retention_days
        self.max_files = max_files
        self.interval_seconds = interval_seconds
        self.pattern = pattern
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None

    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._loop, daemon=True)
        self._thread.start()
        logger.info(
            "[cleanup] worker 启动: %s, retention=%d天, max=%s, interval=%.0fs",
            self.directory,
            self.retention_days,
            self.max_files,
            self.interval_seconds,
        )

    def stop(self) -> None:
        self._stop_event.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)

    def run_once(self) -> int:
        return cleanup_directory(
            self.directory,
            self.retention_days,
            self.max_files,
            self.pattern,
        )

    def _loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                self.run_once()
            except Exception as exc:
                logger.error("[cleanup] 清理异常: %s", exc)
            # 等待下一次清理或停止信号
            self._stop_event.wait(timeout=self.interval_seconds)


def make_cleanup_workers(
    storage_config: dict,
    group_ids: List[str],
    base_path: str = ".",
) -> List[CleanupWorker]:
    """
    根据 STORAGE_CONFIG 为 captures 和 previews 创建清理 worker。

    Args:
        storage_config: STORAGE_CONFIG 字典。
        group_ids: 摄像头组 ID 列表，用于为 captures 每组创建 worker。
        base_path: 项目根目录，相对路径基于此。

    Returns:
        CleanupWorker 列表。
    """
    workers: List[CleanupWorker] = []

    captures_cfg = storage_config.get("captures", {})
    captures_base = captures_cfg.get("base_dir", "data/captures")
    captures_retention = captures_cfg.get("retention_days", 7)
    captures_max = captures_cfg.get("max_files")
    captures_interval = captures_cfg.get("cleanup_interval_seconds", 3600)

    for gid in group_ids:
        workers.append(
            CleanupWorker(
                directory=os.path.join(base_path, captures_base, gid),
                retention_days=captures_retention,
                max_files=captures_max,
                interval_seconds=captures_interval,
                pattern="*",
            )
        )

    previews_cfg = storage_config.get("previews", {})
    previews_base = previews_cfg.get("base_dir", "data/previews")
    previews_retention = previews_cfg.get("retention_days", 7)
    previews_max = previews_cfg.get("max_files")
    previews_interval = previews_cfg.get("cleanup_interval_seconds", 3600)

    workers.append(
        CleanupWorker(
            directory=os.path.join(base_path, previews_base),
            retention_days=previews_retention,
            max_files=previews_max,
            interval_seconds=previews_interval,
            pattern="*",
        )
    )

    return workers