qiniu_delete.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import json
import sys
import argparse
from datetime import datetime
from qiniu import Auth, BucketManager

def load_config(config_path="config.json"):
    """加载配置文件"""
    if not os.path.exists(config_path):
        print(f"错误：配置文件 {config_path} 不存在")
        sys.exit(1)
    
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        
        required_keys = ['access_key', 'secret_key', 'bucket']
        for key in required_keys:
            if key not in config or not config[key] or config[key] == f"你的{key}":
                print(f"错误：请在配置文件中设置有效的 {key}")
                sys.exit(1)
                
        return config
    except json.JSONDecodeError:
        print("错误：配置文件格式不正确，请确保是有效的JSON格式")
        sys.exit(1)
    except Exception as e:
        print(f"错误：读取配置文件时出错 - {str(e)}")
        sys.exit(1)

def format_size(size_bytes):
    """将字节大小转换为人类可读的格式"""
    if size_bytes < 1024:
        return f"{size_bytes} B"
    elif size_bytes < 1024 * 1024:
        return f"{size_bytes/1024:.2f} KB"
    elif size_bytes < 1024 * 1024 * 1024:
        return f"{size_bytes/(1024*1024):.2f} MB"
    else:
        return f"{size_bytes/(1024*1024*1024):.2f} GB"

def preview_files_by_date(bucket_manager, bucket, date_str, prefix=None):
    """预览指定日期之前的所有文件"""
    try:
        target_date = datetime.strptime(date_str, "%Y-%m-%d")
    except ValueError:
        print("错误：日期格式不正确，请使用YYYY-MM-DD格式")
        return

    # 设置每次列举的最大数量
    limit = 1000
    marker = None
    file_count = 0
    total_size = 0
    
    print(f"预览 {date_str} 之前" + (f"且前缀为 {prefix} " if prefix else "") + "的文件：")
    print("-" * 80)
    
    while True:
        # 列举文件
        ret, eof, info = bucket_manager.list(bucket, prefix=prefix, marker=marker, limit=limit)
        
        if info.status_code != 200:
            print(f"错误：列举文件失败 - {info.error}")
            break
            
        if not ret or 'items' not in ret:
            if file_count == 0:
                print("没有找到符合条件的文件")
            break
            
        # 处理当前批次的文件
        for item in ret['items']:
            file_key = item['key']
            file_size = item['fsize']
            putTime = item['putTime'] / 10000000  # 七牛时间戳转换为秒级时间戳
            file_date = datetime.fromtimestamp(putTime)
            
            # 如果文件上传日期早于目标日期
            if file_date < target_date:
                file_count += 1
                total_size += file_size
                # 只显示前20个文件，避免输出过多
                if file_count <= 20:
                    print(f"{file_count}. {file_key} ({format_size(file_size)}) - 上传于 {file_date.strftime('%Y-%m-%d %H:%M:%S')}")
        
        # 判断是否已列举完毕
        if eof:
            break
        # 设置下一次列举的起点
        marker = ret.get('marker', '')

    if file_count > 20:
        print(f"... 还有 {file_count - 20} 个文件未显示 ...")
    
    print("-" * 80)
    print(f"符合条件的文件总数：{file_count}")
    print(f"总计大小：{format_size(total_size)}")
    
    return file_count, total_size

def preview_files_by_prefix(bucket_manager, bucket, prefix):
    """预览指定前缀的所有文件"""
    if not prefix:
        print("错误：必须指定文件前缀")
        return
        
    # 设置每次列举的最大数量
    limit = 1000
    marker = None
    file_count = 0
    total_size = 0
    
    print(f"预览前缀为 {prefix} 的文件：")
    print("-" * 80)
    
    while True:
        # 列举文件
        ret, eof, info = bucket_manager.list(bucket, prefix=prefix, marker=marker, limit=limit)
        
        if info.status_code != 200:
            print(f"错误：列举文件失败 - {info.error}")
            break
            
        if not ret or 'items' not in ret:
            if file_count == 0:
                print("没有找到匹配前缀的文件")
            break
            
        # 处理当前批次的文件
        for item in ret['items']:
            file_key = item['key']
            file_size = item['fsize']
            putTime = item['putTime'] / 10000000
            file_date = datetime.fromtimestamp(putTime)
            
            file_count += 1
            total_size += file_size
            # 只显示前20个文件，避免输出过多
            if file_count <= 20:
                print(f"{file_count}. {file_key} ({format_size(file_size)}) - 上传于 {file_date.strftime('%Y-%m-%d %H:%M:%S')}")
        
        # 判断是否已列举完毕
        if eof:
            break
        # 设置下一次列举的起点
        marker = ret.get('marker', '')

    if file_count > 20:
        print(f"... 还有 {file_count - 20} 个文件未显示 ...")
    
    print("-" * 80)
    print(f"符合条件的文件总数：{file_count}")
    print(f"总计大小：{format_size(total_size)}")
    
    return file_count, total_size

def delete_files_by_date(bucket_manager, bucket, date_str, prefix=None):
    """删除指定日期之前的所有文件"""
    try:
        target_date = datetime.strptime(date_str, "%Y-%m-%d")
    except ValueError:
        print("错误：日期格式不正确，请使用YYYY-MM-DD格式")
        return

    # 设置每次列举的最大数量
    limit = 1000
    marker = None
    deleted_count = 0
    
    while True:
        # 列举文件
        ret, eof, info = bucket_manager.list(bucket, prefix=prefix, marker=marker, limit=limit)
        
        if info.status_code != 200:
            print(f"错误：列举文件失败 - {info.error}")
            break
            
        if not ret or 'items' not in ret:
            print("没有找到文件")
            break
            
        # 处理当前批次的文件
        for item in ret['items']:
            file_key = item['key']
            putTime = item['putTime'] / 10000000  # 七牛时间戳转换为秒级时间戳
            file_date = datetime.fromtimestamp(putTime)
            
            # 如果文件上传日期早于目标日期，则删除
            if file_date < target_date:
                ret, info = bucket_manager.delete(bucket, file_key)
                if ret is None and info.status_code == 200:
                    print(f"已删除: {file_key}")
                    deleted_count += 1
                else:
                    print(f"删除失败: {file_key} - {info.error}")
        
        # 判断是否已列举完毕
        if eof:
            break
        # 设置下一次列举的起点
        marker = ret.get('marker', '')

    print(f"共删除 {deleted_count} 个文件")

def delete_files_by_prefix(bucket_manager, bucket, prefix):
    """删除指定前缀的所有文件"""
    if not prefix:
        print("错误：必须指定文件前缀")
        return
        
    # 设置每次列举的最大数量
    limit = 1000
    marker = None
    deleted_count = 0
    
    while True:
        # 列举文件
        ret, eof, info = bucket_manager.list(bucket, prefix=prefix, marker=marker, limit=limit)
        
        if info.status_code != 200:
            print(f"错误：列举文件失败 - {info.error}")
            break
            
        if not ret or 'items' not in ret:
            print("没有找到匹配前缀的文件")
            break
            
        # 处理当前批次的文件
        for item in ret['items']:
            file_key = item['key']
            ret, info = bucket_manager.delete(bucket, file_key)
            if ret is None and info.status_code == 200:
                print(f"已删除: {file_key}")
                deleted_count += 1
            else:
                print(f"删除失败: {file_key} - {info.error}")
        
        # 判断是否已列举完毕
        if eof:
            break
        # 设置下一次列举的起点
        marker = ret.get('marker', '')

    print(f"共删除 {deleted_count} 个文件")

def main():
    parser = argparse.ArgumentParser(description="七牛云存储文件删除工具")
    parser.add_argument("--date", help="日期（YYYY-MM-DD），操作此日期之前的文件")
    parser.add_argument("--prefix", help="文件前缀，可与日期参数一起使用或单独使用")
    parser.add_argument("--config", default="config.json", help="配置文件路径，默认为当前目录下的config.json")
    parser.add_argument("--preview", action="store_true", help="预览模式，只显示符合条件的文件，不执行删除操作")
    args = parser.parse_args()
    
    if not args.date and not args.prefix:
        parser.print_help()
        print("\n错误：必须指定--date或--prefix参数之一")
        sys.exit(1)
    
    # 加载配置
    config = load_config(args.config)
    
    # 初始化七牛客户端
    q = Auth(config['access_key'], config['secret_key'])
    bucket_manager = BucketManager(q)
    
    # 执行操作
    if args.preview:
        # 预览模式
        if args.date:
            preview_files_by_date(bucket_manager, config['bucket'], args.date, args.prefix)
        elif args.prefix:
            preview_files_by_prefix(bucket_manager, config['bucket'], args.prefix)
    else:
        # 删除模式
        if args.date:
            delete_files_by_date(bucket_manager, config['bucket'], args.date, args.prefix)
        elif args.prefix:
            delete_files_by_prefix(bucket_manager, config['bucket'], args.prefix)

if __name__ == "__main__":
    main()