用 Python 写一个 OBS 批量上传与过期文件清理脚本

一、为什么需要这个脚本

日常把本地备份、日志归档往 OBS（对象存储服务）上搬时，我遇到三个痛点：

1. 控制台上传只适合零散文件，几百个文件点到手酸；
2. OBS 的生命周期规则按前缀+天数配置，遇到“只删 backup/tmp/ 下 30 天前的文件、但保留 backup/db/”这种需求时，控制台配起来很别扭；
3. 删除前想先看一遍“会被删掉哪些文件”，控制台没有试运行模式。

于是写了一个 100 行左右的 Python 脚本：批量上传（自动跳过未变更文件）+ 按前缀和天数清理过期文件（带 dry-run 试运行模式）。下面是完整过程，代码可直接运行。

二、环境准备

1. Python 3.8 及以上版本；
2. 安装官方 SDK：pip install esdk-obs-python；
3. 在 IAM 控制台为当前用户创建一个访问密钥（AK/SK），并记下桶所在区域的终端节点，例如华北-北京四是 obs.cn-north-4.myhuaweicloud.com；
4. 提前创建一个桶，例如 my-backup-bucket。

安全提醒：AK/SK 不要写死在代码里，下面的脚本统一从环境变量 OBS_AK / OBS_SK 读取。

三、完整代码

import os
import logging
from datetime import datetime, timezone, timedelta
from obs import ObsClient

SERVER = "https://obs.cn-north-4.myhuaweicloud.com"  # 按桶所在区域修改
BUCKET = "my-backup-bucket"

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("obs-batch")


def get_client():
    ak = os.environ.get("OBS_AK")
    sk = os.environ.get("OBS_SK")
    if not ak or not sk:
        raise SystemExit("请先 export OBS_AK / OBS_SK")
    return ObsClient(access_key_id=ak, secret_access_key=sk, server=SERVER)


def upload_dir(client, local_dir, prefix=""):
    """递归上传本地目录；远端已存在且大小一致的文件自动跳过。"""
    for root, _, files in os.walk(local_dir):
        for name in files:
            local_path = os.path.join(root, name)
            key = prefix + os.path.relpath(local_path, local_dir).replace(os.sep, "/")
            try:
                meta = client.getObjectMetadata(BUCKET, key)
                if int(meta.body.get("contentLength", -1)) == os.path.getsize(local_path):
                    log.info("跳过未变更: %s", key)
                    continue
            except Exception:
                pass  # 对象不存在，继续上传
            resp = client.putFile(BUCKET, key, local_path)
            if resp.status < 300:
                log.info("上传成功: %s", key)
            else:
                log.error("上传失败 %s: %s", key, resp.errorCode)


def delete_expired(client, prefix, days, dry_run=True):
    """删除指定前缀下 lastModified 早于 days 天的对象；dry_run 只打印不真删。"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    marker, expired = "", []
    while True:
        resp = client.listObjects(BUCKET, prefix=prefix, marker=marker, max_keys=100)
        if resp.status >= 300:
            raise RuntimeError("列举失败: %s" % resp.errorCode)
        for obj in resp.body.contents or []:
            lm = datetime.fromisoformat(obj["lastModified"].replace("Z", "+00:00"))
            if lm < cutoff:
                expired.append(obj["key"])
        if not resp.body.isTruncated:
            break
        marker = resp.body.nextMarker
    if not expired:
        log.info("没有过期对象")
        return
    log.info("发现 %d 个过期对象（早于 %s）", len(expired), cutoff.date())
    if dry_run:
        for key in expired[:20]:
            log.info("[DRY-RUN] 将删除: %s", key)
        if len(expired) > 20:
            log.info("[DRY-RUN] …还有 %d 个未列出", len(expired) - 20)
        return
    for i in range(0, len(expired), 1000):  # 单次批量删除上限 1000 个
        client.deleteObjects(BUCKET, [{"key": k} for k in expired[i:i + 1000]])
        log.info("已删除 %d 个", min(1000, len(expired) - i))


if __name__ == "__main__":
    c = get_client()
    upload_dir(c, "./data", prefix="backup/")
    delete_expired(c, prefix="backup/tmp/", days=30, dry_run=True)
    c.close()

四、运行与验证

export OBS_AK=你的AK
export OBS_SK=你的SK
python obs_batch.py

第一次建议保持 dry_run=True，先确认日志里“[DRY-RUN] 将删除”列出的文件确实都是可删的，再改成 dry_run=False 跑一次真删。我自己的目录 300 多个日志文件，上传只用了约 2 分钟；清理时 dry-run 发现 41 个过期文件，逐条核对无误后才执行删除。

五、几个实测踩过的坑

1. server 域名必须带上区域：例如 obs.cn-north-4.myhuaweicloud.com，区域写错会报 403 签名错误。遇到 403 先检查域名，别急着怀疑 AK/SK。
2. Windows 路径分隔符：os.path.relpath 在 Windows 下产出反斜杠，对象名里出现反斜杠会很丑，代码里用 .replace(os.sep, "/") 统一成正斜杠。
3. 时间比较用 UTC：lastModified 是 UTC 时间，datetime.now() 不带时区得到的是本地时间，直接比较会差 8 小时，统一转成 timezone.utc 再比较。
4. 大文件用分段上传：单个文件超过 100MB 时建议改用分段上传接口，断点续传更稳；上面的 putFile 适合中小文件批量场景。

六、小结

这个脚本我放在定时任务里每周跑一次：先增量上传本周备份，再清理 30 天前的临时文件，全程不用打开控制台。如果你也有“上传 + 定期清理”的固定动作，不妨照着改一改前缀和天数直接用。欢迎在评论区交流你在 OBS 自动化上的做法。