用 Python + 华为云 OBS SDK 写一个文件自动归档工具

为什么需要这个工具

做运维或后端开发的同学大概都遇到过这个场景：服务器上某个日志目录或数据目录越来越大，磁盘告警隔三差五弹一次。你手动清理怕删错，不清理又怕磁盘满。

其实很多旧文件你短期内根本不会再看，但万一哪天要用又不能直接删。归档到对象存储是最合适的方案——把冷数据搬到 OBS，本地只留近期文件，既省磁盘又留后路。

华为云 OBS 提供三种存储类别，价格差距很大：

如果能把 90 天前的日志自动放到归档存储，存储成本能降一个数量级。

这篇文章就带你用 Python + 华为云 OBS SDK 写一个文件自动归档工具，核心逻辑不到 120 行，拿来就能用。

工具设计思路

先想清楚要做什么，再动手写代码：

1. 扫描：递归扫描指定本地目录，拿到所有文件及其修改时间。

2. 分类：按文件修改时间算"年龄"，根据预设规则选择存储类别。比如：

• 30 天内 → 标准存储（可能还要看）
• 30～90 天 → 低频访问存储（偶尔翻一下）
• 90 天以上 → 归档存储（基本不看了）

3. 上传：把文件传到 OBS 桶，使用对应的存储类别。OBS 的 objectKey 保持相对路径，方便后续查找。

4. 记录：每上传完一个文件，把信息写入本地记录文件。下次运行时如果文件没被修改过就跳过——这就是断点续传，中途断了重跑不会重复上传。

5. 日志：所有操作写日志文件，出问题能追溯。

准备工作

安装 OBS Python SDK

pip install esdk-obs-python

这是华为云官方的 OBS Python SDK，包名是 esdk-obs-python，导入时用 from obs import ObsClient。

获取 AK/SK

在华为云控制台进入"我的凭证"页面，创建访问密钥（AK/SK）。建议通过环境变量传入，不要硬编码到脚本里：

export OBS_AK=你的AccessKey
export OBS_SK=你的SecretKey

创建 OBS 桶

在 OBS 控制台创建一个桶，记下桶名和 endpoint。比如桶名 my-archive-bucket，区域 cn-north-4，对应的 endpoint 是 obs.cn-north-4.myhuaweicloud.com。

完整代码

下面是完整脚本，保存为 obs_archiver.py：

#!/usr/bin/env python3
"""
文件自动归档工具 - 基于华为云 OBS
按文件修改时间自动选择存储类别上传到 OBS 桶
支持断点续传、日志记录、试运行模式
"""

import os
import json
import time
import logging
from datetime import datetime
from obs import ObsClient

# ============ 配置区 ============
AK = os.environ.get('OBS_AK', 'your-ak')
SK = os.environ.get('OBS_SK', 'your-sk')
ENDPOINT = 'obs.cn-north-4.myhuaweicloud.com'
BUCKET = 'my-archive-bucket'
LOCAL_DIR = '/data/logs'           # 要归档的本地目录
DRY_RUN = False                     # True = 只打印不实际上传

# 归档策略：(超过N天用此类别, 存储类别名)
# 从上往下匹配，命中第一条即停止
ARCHIVE_RULES = [
    (30,  'STANDARD'),    # 30天内：标准存储
    (90,  'WARM'),        # 30~90天：低频访问存储
    (999, 'COLD'),        # 90天以上：归档存储
]

RECORD_FILE = '.archive_record.json'   # 断点续传记录文件
LOG_FILE = 'archive.log'               # 日志文件

# ============ 初始化 ============
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE, encoding='utf-8'),
        logging.StreamHandler(),
    ]
)
logger = logging.getLogger('archiver')

obs_client = ObsClient(access_key_id=AK, secret_access_key=SK, server=ENDPOINT)


# ============ 核心逻辑 ============
def load_record():
    """加载已上传记录，用于断点续传"""
    if os.path.exists(RECORD_FILE):
        with open(RECORD_FILE, 'r') as f:
            return json.load(f)
    return {}


def save_record(record):
    """保存上传记录到本地文件"""
    with open(RECORD_FILE, 'w') as f:
        json.dump(record, f, indent=2, ensure_ascii=False)


def pick_storage_class(days_old):
    """根据文件年龄（天）选择存储类别"""
    for threshold, storage_class in ARCHIVE_RULES:
        if days_old < threshold:
            return storage_class
    return ARCHIVE_RULES[-1][1]


def scan_files(local_dir):
    """递归扫描目录，返回 (相对路径, 绝对路径, 修改时间) 列表"""
    results = []
    for root, _, files in os.walk(local_dir):
        for fname in files:
            abs_path = os.path.join(root, fname)
            rel_path = os.path.relpath(abs_path, local_dir)
            mtime = os.path.getmtime(abs_path)
            results.append((rel_path, abs_path, mtime))
    return results


def upload_file(local_path, obs_key, storage_class):
    """上传单个文件到 OBS，返回是否成功"""
    if DRY_RUN:
        logger.info(f"[DRY-RUN] {obs_key} -> {storage_class}")
        return True

    resp = obs_client.putFile(
        bucketName=BUCKET,
        objectKey=obs_key,
        filePath=local_path,
        storageClass=storage_class,
    )
    if resp.status < 300:
        logger.info(f"上传成功: {obs_key} ({storage_class})")
        return True
    else:
        logger.error(f"上传失败: {obs_key}, status={resp.status}, reason={resp.reason}")
        return False


def run():
    """主流程：扫描 -> 分类 -> 上传 -> 记录"""
    record = load_record()
    now = time.time()
    files = scan_files(LOCAL_DIR)
    logger.info(f"扫描完成: 共 {len(files)} 个文件")

    uploaded, skipped, failed = 0, 0, 0

    for rel_path, abs_path, mtime in files:
        # 断点续传：已上传且文件未修改则跳过
        if rel_path in record and record[rel_path].get('mtime') == mtime:
            skipped += 1
            continue

        days_old = int((now - mtime) / 86400)
        storage_class = pick_storage_class(days_old)
        obs_key = rel_path.replace(os.sep, '/')  # OBS 用 / 作路径分隔符

        if upload_file(abs_path, obs_key, storage_class):
            record[rel_path] = {
                'mtime': mtime,
                'storage_class': storage_class,
                'uploaded_at': datetime.now().isoformat(),
            }
            uploaded += 1
            save_record(record)  # 每次上传后保存，防中途崩溃丢记录
        else:
            failed += 1

    logger.info(f"全部完成: 上传 {uploaded}, 跳过 {skipped}, 失败 {failed}")


if __name__ == '__main__':
    run()

代码逐段解读

配置区

开头一堆大写变量是配置项，你只需要改三个地方：

• ENDPOINT：你的桶所在区域的 endpoint，在 OBS 控制台的桶概览页能看到
• BUCKET：桶名
• LOCAL_DIR：你要归档的本地目录路径

ARCHIVE_RULES 是归档策略，格式是 (天数阈值, 存储类别)。从上往下匹配，命中第一条就停。想调整策略只改这里就行，比如想 7 天就转低频，改第一个数字为 7 即可。

DRY_RUN 是试运行开关，设为 True 时只打印不上传，方便先看看会怎么处理再真正跑。

断点续传

load_record() 和 save_record() 管理一个本地 JSON 文件，记录每个文件的上传状态。关键逻辑在 run() 里：

if rel_path in record and record[rel_path].get('mtime') == mtime:
    skipped += 1
    continue

如果文件之前传过，而且修改时间没变，就跳过。这样你中途 Ctrl+C 断掉，下次重跑只会处理没传完的文件。注意每传完一个文件就 save_record(record) 保存一次，虽然 IO 多了一点，但保证记录不丢。

存储类别选择

pick_storage_class() 就是简单的规则匹配。OBS SDK 识别三个字符串：

• 'STANDARD'：标准存储
• 'WARM'：低频访问存储
• 'COLD'：归档存储

上传

核心就一行 obs_client.putFile()，它直接传本地文件路径，SDK 内部处理分片上传，大文件也不用担心。objectKey 用的是相对路径（把 os.sep 替换成 /），这样在 OBS 里能保持目录结构，下载或浏览时一目了然。

运行效果

配好环境变量后直接跑：

python obs_archiver.py

输出大概长这样：

2026-10-29 14:00:01 [INFO] 扫描完成: 共 1280 个文件
2026-10-29 14:00:02 [INFO] 上传成功: app/2026/access.log.20260101 (COLD)
2026-10-29 14:00:03 [INFO] 上传成功: app/2026/access.log.20260102 (COLD)
2026-10-29 14:00:04 [INFO] 上传成功: app/2026/access.log.20260901 (WARM)
2026-10-29 14:00:05 [INFO] 上传成功: app/2026/access.log.20261020 (STANDARD)
...
2026-10-29 14:15:30 [INFO] 全部完成: 上传 1024, 跳过 256, 失败 0

再跑一次，已上传且未修改的文件会被跳过：

2026-10-29 15:00:00 [INFO] 扫描完成: 共 1280 个文件
2026-10-29 15:00:01 [INFO] 全部完成: 上传 0, 跳过 1280, 失败 0

搭配 crontab 定时运行

手动跑一次不够，真正用起来要定时执行。加一条 crontab 每天凌晨 2 点自动归档：

# 编辑 crontab
crontab -e

# 添加：每天凌晨2点执行
0 2 * * * /usr/bin/python3 /path/to/obs_archiver.py >> /path/to/archive_cron.log 2>&1

这样每天自动扫描一次，新产生的文件到了年龄阈值就会被归档，完全不用人管。

几个注意事项

1. 归档存储取回有延迟。 归档存储（COLD）的对象不能直接下载，需要先做"恢复"操作，通常需要几分钟到几小时。所以只把"几乎不会看"的文件放归档，不确定的就放低频。

2. 上传带宽。 如果本地文件很多、很大，首次全量上传可能跑很久。可以在配置区加一个 MAX_FILE_SIZE 过滤，先传小文件，大文件单独处理。或者分目录多次运行。

3. AK/SK 安全。 文里用的是环境变量传入，不要把密钥硬编码到脚本里再提交到 Git。如果必须在脚本里写，至少加个 .gitignore 把脚本排除掉。

4. 桶的区域。 endpoint 要和桶的实际区域一致，不然会报 404。不确定的话在 OBS 控制台桶概览页看"终端节点"那一栏。

扩展思路

这个工具的核心逻辑很简单，很容易往上加功能：

• 上传后删除本地文件：加一个 CLEAN_LOCAL = True 开关，上传成功后删掉本地文件，真正释放磁盘。建议先跑一段时间确认上传没问题再开。
• 按文件大小过滤：只归档大于某个阈值的文件，小文件留着不折腾。
• 多桶策略：不同目录的文件传到不同的桶，配置区改成字典就行。
• 加密上传：OBS 支持 SSE-KMS 服务端加密，putFile 加一个 sseHeader 参数即可。
• 进度条：装个 tqdm 库，扫描完文件总数后套一层进度条，大目录跑起来心里有数。

总结

这篇文章用不到 120 行 Python 代码实现了一个实用的文件归档工具，核心就三步：扫描本地文件、按年龄选存储类别、传到 OBS。加上断点续传和日志记录，已经能当一个小工具正经用了。

OBS 的三种存储类别价格差距很大，把冷数据自动放到归档存储是省钱的有效手段。配合 crontab 定时运行，基本实现"设好策略就不用管"的效果。

代码拿去改改配置就能跑，希望对你有用。