cfc/scripts/migrate-to-jdcloud.py

#!/usr/bin/env python3
"""
京东云对象存储存量迁移脚本。

遍历本地 upload.base-dir（默认 /data/cfc-uploads）下的所有文件，
上传到京东云 OSS bucket cfc，保持原有目录结构。

用法:
    # 查看会迁移哪些文件（不实际上传）
    python3 scripts/migrate-to-jdcloud.py --dry-run

    # 执行迁移（自动跳过云端已存在的文件）
    python3 scripts/migrate-to-jdcloud.py

    # 强制覆盖云端已存在的文件
    python3 scripts/migrate-to-jdcloud.py --overwrite

    # 指定本地目录
    python3 scripts/migrate-to-jdcloud.py --local-dir /data/cfc-uploads

    # 迁移后回填 file_record 表（用于去重）
    python3 scripts/migrate-to-jdcloud.py --backfill
"""

import os
import sys
import boto3
import hashlib
import pymysql
from botocore.config import Config
from pathlib import Path
import argparse
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("migrate")

# 京东云 OSS 配置
ENDPOINT = "s3.cn-north-1.jdcloud-oss.com"
ACCESS_KEY = "JDC_F351EC3CB1F8593204ABCDA2CBF2"
SECRET_KEY = "970C0C6985A6BF29D0C79AE3693B9899"
BUCKET = "cfc"
REGION = "cn-north-1"

# MySQL 配置（用于回填 file_record 表）
DB_HOST = "mysql-internet-cn-north-1-23feae22680e4dfa.rds.jdcloud.com"
DB_PORT = 3306
DB_USER = "cfc"
DB_PASS = "cfc@1314"
DB_NAME = "zxyj"

# 公网URL前缀
PUBLIC_URL_PREFIX = f"https://{BUCKET}.{ENDPOINT}/"


def get_client():
    return boto3.client(
        "s3",
        endpoint_url=f"https://{ENDPOINT}",
        aws_access_key_id=ACCESS_KEY,
        aws_secret_access_key=SECRET_KEY,
        config=Config(s3={"addressing_style": "path"}, signature_version="s3v4"),
        region_name=REGION,
    )


def get_db():
    return pymysql.connect(
        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS,
        database=DB_NAME, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor,
    )


def list_cloud_keys(client):
    """列出云端已存在的所有 key"""
    keys = set()
    marker = None
    while True:
        if marker:
            resp = client.list_objects_v2(Bucket=BUCKET, MaxKeys=1000, StartAfter=marker)
        else:
            resp = client.list_objects_v2(Bucket=BUCKET, MaxKeys=1000)
        if "Contents" not in resp:
            break
        for obj in resp["Contents"]:
            keys.add(obj["Key"])
            marker = obj["Key"]
        if not resp.get("IsTruncated"):
            break
    return keys


def upload_file(client, local_path, key, dry_run=False, overwrite=False, cloud_keys=None):
    """上传单个文件，返回 (成功与否, target_url)"""
    target_url = PUBLIC_URL_PREFIX + key

    if cloud_keys and key in cloud_keys and not overwrite:
        log.info("  [SKIP] 已在云端存在: %s", target_url)
        return False, target_url

    if dry_run:
        log.info("  [DRY-RUN] 将上传: %s -> %s", local_path, target_url)
        return False, target_url

    try:
        content_type = guess_content_type(local_path)
        extra_args = {"ContentType": content_type} if content_type else {}
        client.upload_file(str(local_path), BUCKET, key, ExtraArgs=extra_args)
        log.info("  [OK] %s", target_url)
        return True, target_url
    except Exception as e:
        log.error("  [FAIL] %s: %s", local_path, e)
        return False, target_url


def guess_content_type(path):
    """根据扩展名猜测 Content-Type"""
    ext = Path(path).suffix.lower()
    mapping = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp",
        ".bmp": "image/bmp",
        ".svg": "image/svg+xml",
        ".pdf": "application/pdf",
        ".doc": "application/msword",
        ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        ".xls": "application/vnd.ms-excel",
        ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        ".mp4": "video/mp4",
        ".mp3": "audio/mpeg",
        ".wav": "audio/wav",
        ".txt": "text/plain",
        ".json": "application/json",
        ".zip": "application/zip",
        ".html": "text/html",
        ".css": "text/css",
        ".js": "application/javascript",
    }
    return mapping.get(ext, "application/octet-stream")


def guess_category(local_path, local_dir):
    """根据目录结构猜测文件分类"""
    relative = local_path.relative_to(local_dir)
    parts = relative.parts
    if len(parts) >= 2:
        return parts[0]
    return "general"


def backfill_file_record(db, local_path, url, local_dir):
    """回填 file_record 表"""
    category = guess_category(local_path, local_dir)
    file_size = local_path.stat().st_size
    content_type = guess_content_type(local_path)
    original_name = local_path.name

    # 计算 SHA-256
    sha256_hash = hashlib.sha256()
    with open(local_path, "rb") as f:
        sha256_hash.update(f.read())
    file_hash = sha256_hash.hexdigest()

    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    with db.cursor() as cursor:
        # 检查是否已存在
        cursor.execute("SELECT id FROM file_record WHERE hash = %s", (file_hash,))
        if cursor.fetchone():
            log.debug("  [SKIP-BACKFILL] hash 已存在: %s", file_hash)
            return False
        sql = (
            "INSERT INTO file_record (hash, url, file_size, content_type, "
            "original_filename, category, created_at, updated_at) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s)"
        )
        cursor.execute(sql, (
            file_hash, url, file_size, content_type,
            original_name, category, now, now,
        ))
    db.commit()
    log.info("  [BACKFILL] %s -> %s", file_hash, url)
    return True


def main():
    parser = argparse.ArgumentParser(description="迁移本地文件到京东云对象存储")
    parser.add_argument("--local-dir", default="/data/cfc-uploads", help="本地文件目录")
    parser.add_argument("--dry-run", action="store_true", help="仅预览，不实际上传")
    parser.add_argument("--overwrite", action="store_true", help="覆盖云端已存在的文件")
    parser.add_argument("--backfill", action="store_true", help="上传后回填 file_record 表（用于去重）")
    args = parser.parse_args()

    local_dir = Path(args.local_dir)
    if not local_dir.is_dir():
        log.error("本地目录不存在: %s", local_dir)
        sys.exit(1)

    # 收集所有本地文件
    all_files = sorted(local_dir.rglob("*"))
    files = [f for f in all_files if f.is_file()]

    if not files:
        log.info("本地目录为空，无需迁移")
        return

    log.info("本地目录: %s", local_dir)
    log.info("目标 Bucket: %s", BUCKET)
    log.info("目标 Endpoint: %s", ENDPOINT)
    log.info("共发现 %d 个文件", len(files))
    if args.dry_run:
        log.info("模式: DRY-RUN（仅预览不上传）")
    elif args.overwrite:
        log.info("模式: 覆盖已存在文件")
    else:
        log.info("模式: 跳过已存在文件")
    if args.backfill:
        log.info("回填: 启用（写入 file_record 表）")

    # 连接京东云
    if not args.dry_run:
        log.info("正在连接京东云 OSS...")
        client = get_client()
        try:
            client.head_bucket(Bucket=BUCKET)
            log.info("Bucket %s 连接成功", BUCKET)
        except Exception as e:
            log.error("Bucket 连接失败: %s", e)
            sys.exit(1)

        cloud_keys = list_cloud_keys(client) if not args.overwrite else set()
        log.info("云端已有 %d 个文件", len(cloud_keys))
    else:
        client = None
        cloud_keys = set()

    # 连接 MySQL（回填模式）
    db = None
    if args.backfill and not args.dry_run:
        try:
            db = get_db()
            log.info("MySQL 连接成功")
        except Exception as e:
            log.error("MySQL 连接失败: %s", e)
            sys.exit(1)

    # 上传文件
    success = 0
    skipped = 0
    failed = 0
    backfill_ok = 0
    backfill_skip = 0
    backfill_fail = 0

    for file_path in files:
        relative = file_path.relative_to(local_dir)
        key = "uploads/" + relative.as_posix()

        if client and key in cloud_keys and not args.overwrite:
            skipped += 1
            # 即使跳过上传，也回填 file_record（如果启用）
            if db:
                try:
                    if backfill_file_record(db, file_path, PUBLIC_URL_PREFIX + key, local_dir):
                        backfill_ok += 1
                    else:
                        backfill_skip += 1
                except Exception as e:
                    log.error("  [BACKFILL-FAIL] %s: %s", file_path, e)
                    backfill_fail += 1
            continue

        ok, url = upload_file(client, file_path, key, args.dry_run, args.overwrite, cloud_keys)
        if ok:
            success += 1
            if db:
                try:
                    if backfill_file_record(db, file_path, url, local_dir):
                        backfill_ok += 1
                    else:
                        backfill_skip += 1
                except Exception as e:
                    log.error("  [BACKFILL-FAIL] %s: %s", file_path, e)
                    backfill_fail += 1
        else:
            if not args.dry_run:
                failed += 1

    # 总结
    if not args.dry_run:
        log.info("=" * 50)
        parts = [f"成功={success}", f"跳过={skipped}", f"失败={failed}"]
        if db:
            parts.append(f"回填={backfill_ok}")
            parts.append(f"回填跳过={backfill_skip}")
            if backfill_fail:
                parts.append(f"回填失败={backfill_fail}")
        log.info("迁移完成: " + ", ".join(parts))
        if failed > 0:
            log.warning("有 %d 个文件上传失败，请检查日志", failed)
        if db:
            db.close()
    else:
        log.info("=" * 50)
        log.info("DRY-RUN 完成: 将上传 %d 个文件，跳过 %d 个（已存在）", success, skipped)


if __name__ == "__main__":
    main()