Python图床迁移实现方法与经验分享

🤦🏻‍♂️编程菜鸟只能手动迁移

本来是准备全部迁移到 GitHub，迁了一半才想起 GitHub 有 1G 容量限制

于是准备剩下的都迁到 one drive 图床。

我最近刚把一个图床从七牛云迁移到阿里云OSS，分享一下我的实现方法。

核心思路就是：读取源图床的文件列表，下载到本地，再上传到新图床。这里用七牛云和阿里云OSS的Python SDK为例：

import os
from qiniu import Auth, BucketManager
from aliyunsdkcore.client import AcsClient
from aliyunsdkoss.request.v20190517 import PutObjectRequest
import requests
from concurrent.futures import ThreadPoolExecutor

class ImageBedMigrator:
    def __init__(self, qiniu_access_key, qiniu_secret_key, qiniu_bucket,
                 oss_access_key, oss_secret_key, oss_endpoint, oss_bucket):
        # 初始化七牛云
        self.qiniu_auth = Auth(qiniu_access_key, qiniu_secret_key)
        self.qiniu_bucket = qiniu_bucket
        self.bucket_manager = BucketManager(self.qiniu_auth)
        
        # 初始化阿里云OSS
        self.oss_client = AcsClient(oss_access_key, oss_secret_key, 'cn-hangzhou')
        self.oss_endpoint = oss_endpoint
        self.oss_bucket = oss_bucket
        
        # 临时下载目录
        self.temp_dir = './temp_images'
        os.makedirs(self.temp_dir, exist_ok=True)
    
    def get_qiniu_file_list(self, prefix='', limit=1000):
        """获取七牛云文件列表"""
        marker = None
        all_files = []
        
        while True:
            ret, eof, info = self.bucket_manager.list(
                self.qiniu_bucket, prefix=prefix, marker=marker, limit=limit
            )
            
            if ret is None:
                print(f"获取文件列表失败: {info}")
                break
                
            all_files.extend(ret.get('items', []))
            
            if eof:
                break
                
            marker = ret.get('marker')
            
        return all_files
    
    def download_file(self, file_info):
        """下载单个文件"""
        key = file_info['key']
        url = f"http://your-qiniu-domain/{key}"  # 替换为你的七牛云域名
        
        local_path = os.path.join(self.temp_dir, key)
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        
        try:
            response = requests.get(url, stream=True)
            if response.status_code == 200:
                with open(local_path, 'wb') as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)
                print(f"下载成功: {key}")
                return local_path, key
        except Exception as e:
            print(f"下载失败 {key}: {e}")
        return None, key
    
    def upload_to_oss(self, local_path, oss_key):
        """上传到阿里云OSS"""
        try:
            with open(local_path, 'rb') as f:
                content = f.read()
            
            request = PutObjectRequest()
            request.set_Endpoint(self.oss_endpoint)
            request.set_BucketName(self.oss_bucket)
            request.set_Key(oss_key)
            request.set_Body(content)
            
            response = self.oss_client.do_action_with_exception(request)
            print(f"上传成功: {oss_key}")
            return True
        except Exception as e:
            print(f"上传失败 {oss_key}: {e}")
            return False
    
    def migrate(self, max_workers=5):
        """执行迁移"""
        print("开始获取七牛云文件列表...")
        files = self.get_qiniu_file_list()
        print(f"共找到 {len(files)} 个文件")
        
        # 使用线程池并发下载
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            download_results = list(executor.map(self.download_file, files))
        
        # 上传到OSS
        success_count = 0
        for local_path, key in download_results:
            if local_path and os.path.exists(local_path):
                if self.upload_to_oss(local_path, key):
                    success_count += 1
                # 清理临时文件
                os.remove(local_path)
        
        print(f"迁移完成，成功 {success_count}/{len(files)} 个文件")
        
        # 清理空目录
        os.rmdir(self.temp_dir)

# 使用示例
if __name__ == "__main__":
    migrator = ImageBedMigrator(
        qiniu_access_key='your_qiniu_ak',
        qiniu_secret_key='your_qiniu_sk',
        qiniu_bucket='your_qiniu_bucket',
        oss_access_key='your_oss_ak',
        oss_secret_key='your_oss_sk',
        oss_endpoint='oss-cn-hangzhou.aliyuncs.com',
        oss_bucket='your_oss_bucket'
    )
    
    migrator.migrate(max_workers=10)

几个关键点：

1. 先获取源图床所有文件，注意处理分页
2. 用线程池并发下载提高速度
3. 保持文件路径结构不变
4. 记得清理临时文件

迁移后要更新数据库里的图片URL，这个根据你的存储方式来处理。

迁移前先小批量测试，确认无误再全量迁移。