工程实践：天猫结算凭证RPA自动下载与COS存储落地，从踩坑到稳定运行的完整记录

# -*- coding: utf-8 -*-
"""
天猫结算凭证自动下载 -> 本地校验 -> 上传腾讯云COS
依赖：pip install cos-python-sdk-v5 requests
注意：替换 DOWNLOAD 接口地址、SecretId/SecretKey 为你自己的配置
"""
import hashlib
import logging
import os
import random
import time

import requests
from qcloud_cos import CosConfig, CosS3Client

logging.basicConfig(
    filename="voucher_sync.log",
    filemode="a",  # 追加模式，保留历史日志
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

BUCKET = "voucher-1234567890"   # 替换为你的存储桶名
REGION = "ap-guangzhou"         # 替换为存储桶所在地域
LOCAL_DIR = "./vouchers"
UPLOADED_LIST = "./uploaded.txt"  # 断点续跑台账

def md5_of(path):
    h = hashlib.md5()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

def download_one(voucher_id):
    """按凭证ID下载PDF，伪接口需替换为实际导出地址"""
    url = f"https://你的结算导出接口/vouchers/{voucher_id}.pdf"  # 替换
    for attempt in range(3):
        try:
            resp = requests.get(url, timeout=30)
            if resp.status_code == 200 and resp.content[:4] == b"%PDF":
                path = os.path.join(LOCAL_DIR, f"{voucher_id}.pdf")
                with open(path, "wb") as f:
                    f.write(resp.content)
                return path
            logging.warning("下载异常 %s status=%s", voucher_id, resp.status_code)
        except requests.RequestException as e:
            logging.warning("下载重试 %s 第%d次: %s", voucher_id, attempt + 1, e)
        time.sleep(0.5 + random.random() * 0.3)  # 500~800ms随机间隔
    else:
        logging.error("下载失败，记入待处理清单: %s", voucher_id)
        return None

def main(voucher_ids):
    client = CosS3Client(CosConfig(Region=REGION, SecretId="你的SecretId", SecretKey="你的SecretKey"))
    done = set()
    if os.path.exists(UPLOADED_LIST):
        with open(UPLOADED_LIST) as f:
            done = {line.strip() for line in f}
    for vid in voucher_ids:
        if f"{vid}.pdf" in done:
            continue  # 断点续跑：跳过已上传文件
        path = download_one(vid)
        if not path:
            continue
        checksum = md5_of(path)  # 本地完整性校验，哈希随台账一并留存
        for attempt in range(3):
            try:
                key = "tmall-vouchers/" + os.path.basename(path)
                client.upload_file(BUCKET, key, path)
                with open(UPLOADED_LIST, "a") as f:
                    f.write(f"{os.path.basename(path)} {checksum}\n")
                logging.info("已上传 %s md5=%s -> cos://%s/%s", path, checksum, BUCKET, key)
                break
            except Exception as e:
                logging.warning("上传重试 %s 第%d次: %s", vid, attempt + 1, e)
                time.sleep(2 ** attempt)  # 指数退避
        else:
            logging.error("上传失败待人工处理: %s", vid)

if __name__ == "__main__":
    os.makedirs(LOCAL_DIR, exist_ok=True)
    main(["20261100001", "20261100002"])  # 替换为实际凭证ID列表