Function cleanup_database_logs

src/core/db_logs.py:111–164

清理后台日志： 1) 删除超过 retention_days 的日志 2) 若总量超过 max_rows，删除最旧的超量部分

(
    retention_days: Optional[int] = None,
    max_rows: int = 50000,
)

def cleanup_database_logs(
    retention_days: Optional[int] = None,
    max_rows: int = 50000,
) -> Dict[str, Any]:
    """
    清理后台日志：
    1) 删除超过 retention_days 的日志
    2) 若总量超过 max_rows，删除最旧的超量部分
    """
    settings = get_settings()
    keep_days = int(retention_days if retention_days is not None else settings.log_retention_days or 30)
    keep_days = max(1, keep_days)
    max_rows = max(1000, int(max_rows))
    cutoff = utcnow_naive() - timedelta(days=keep_days)

    deleted_by_age = 0
    deleted_by_limit = 0

    with get_db() as db:
        deleted_by_age = (
            db.query(AppLog)
            .filter(AppLog.created_at < cutoff)
            .delete(synchronize_session=False)
        )
        db.commit()

        total = db.query(func.count(AppLog.id)).scalar() or 0
        if total > max_rows:
            overflow = int(total - max_rows)
            overflow_ids = [
                row_id
                for (row_id,) in db.query(AppLog.id)
                .order_by(AppLog.created_at.asc(), AppLog.id.asc())
                .limit(overflow)
                .all()
            ]
            if overflow_ids:
                deleted_by_limit = (
                    db.query(AppLog)
                    .filter(AppLog.id.in_(overflow_ids))
                    .delete(synchronize_session=False)
                )
                db.commit()

        remaining = db.query(func.count(AppLog.id)).scalar() or 0

    return {
        "retention_days": keep_days,
        "max_rows": max_rows,
        "deleted_by_age": int(deleted_by_age or 0),
        "deleted_by_limit": int(deleted_by_limit or 0),
        "deleted_total": int((deleted_by_age or 0) + (deleted_by_limit or 0)),
        "remaining": int(remaining or 0),
    }