哈喽，大家好啊！
在平时大家记录点东西、写笔记、做博客之类，经常需要插入图片。
以前我一直用七牛云做个人图床，个人有免费10G存储空间用着还不错，但最近流量费用是在负担不起，所以就只能从七牛云迁移到 Cloudflare R2。
这出口流量费用还是有点肉疼。
所以在综合对比之后，我最终还是决定将图床迁移到 Cloudflare R2。
并且相比七牛云，Cloudflare R2 具有天然的优势，出入流量不限制：
- 不收取出口流量费用，图片访问再多也不用担心流量费用，不像国内的云厂商。
- 免费提供 10GB 存储额度，个人博客基本够用。
- 自带 Cloudflare CDN 网络，无需额外购买加速服务。
- 支持绑定自定义域名。
- 每月提供约 100 万次写入请求 + 1000 万次读取请求，个人博客图片加载基本无压力。
接下来我会把七牛云上的图片逐步迁移到 Cloudflare R2，并配置自定义域名。
使用七牛云作为备份存储，实现双备份，避免单个平台出现问题导致图片数据丢失。
同时教大家如何搭建自定义图床，顺便简单记录一下。
一、准备
在开始之前，需要提前准备以下内容：
-
一个 Cloudflare 账号，没有可以直接使用QQ邮箱注册一个
-
一个域名，没有的话可以购买一个，域名推荐大家可以去Spaceship 购买一个XYZ后缀的域名，输入优惠码<code>XYZ52</code>，折合人民币几块钱，支持国内支付，比国内任何X云都便宜。
https://spaceship.sjv.io/c/7338998/1794549/21274
- 七牛云图床中的图片数据（如果需要迁移）
二、Cloudflare 添加域名
打开 <code>Cloudflare</code>官网 ，如果没有账户可以注册一个。
Cloudflare 注册完成、通过邮箱验证后，会直接跳转到面板页，点击域名—>概览—>加入域。
输入刚刚购买好的域名，然后点击继续。
选择免费计划，继续选择前往激活。
复制<code>cloudflare</code>生成的DNS 填写到刚刚注册的域名对应的DNS地址里面。
回到 Cloudflare，点击底部我已更新名称服务器，接下来，就只需要等待很短的时间，一般为 3-5 分钟添加托管到 Cloudflare 的域名就会生效。
三、创建 Cloudflare R2 存储桶
接下来我们开始创建 Cloudflare R2 存储桶，并绑定自定义域名。
1. 进入 R2 控制台
在 Cloudflare 后台左侧菜单找到存储和数据库→ R2对象存储→将R2订阅添加到我的帐户。
接着绑定支付方式。
点击进入后，选择 Create bucket（创建存储桶）。
2. 创建 Bucket
这里需要填写一个 Bucket 名称，例如：<code>blog-img</code>，位置选择亚太地区，然后点击 Create bucket。
创建成功后，你会看到一个空的存储桶界面。
3. 绑定自定义域名
接下来绑定自己的域名，点击设置→自定义域名
输入你的域名，例如：
img.yourdomain.com然后 Cloudflare 会自动帮你完成解析。绑定成功后，你的图片访问地址会变成：
https://img.yourdomain.com/xxx.png当然，你也可以开启开发域名，然后Cloudflare 会提供临时一个域名提供你测试访问。
4. 设置CORS 访问策略
为了后期不必要的麻烦，这里简单设置一下CORS访问策略，点击CORS 策略添加。
复制如下内容填入后保存。
[
{
"AllowedOrigins": [
"http://img.yourdomain.com",
"https://img.yourdomain.com"
],
"AllowedMethods": [
"GET",
"POST",
"PUT",
"DELETE",
"HEAD"
],
"AllowedHeaders": ["*"],
"ExposeHeaders": ["ETag"],
"MaxAgeSeconds": 3600
}
]5.获取 API令牌
在R2 控制台右下角点击管理
创建账户 API 令牌
配置权限支持读写操作，并且设置
最后保存访问密钥ID和机密访问密钥
四、迁移七牛云数据
1. 下载迁移工具
先去官网下载 <code>qshell</code> 工具：
https://developer.qiniu.com/kodo/1302/qshell根据自己的操作系统下载对应版本，Linux安装示例如下：
wget https://github.com/qiniu/qshell/releases/download/v2.19.8/qshell-v2.19.8-linux-amd64.tar.gz
tar -xzf qshell-v2.19.8-linux-amd64.tar.gz
chmod +x qshell
mv qshell /usr/local/bin/qshell
qshell -v安装好<code>qshell</code>以后安装 <code>rclone</code>：
curl https://rclone.org/install.sh | sudo bash
rclone version创建工作目录：
mkdir -p ~/qiniu-r2-migration/data
cd ~/qiniu-r2-migration2. 环境变量配置
把下面占位符替换成你的真实值：
export QINIU_AK="你的七牛AK"
export QINIU_SK="你的七牛SK"
export QINIU_BUCKET="你的七牛空间名"
export LOCAL_DIR="$HOME/qiniu-r2-migration/data"
export R2_ACCOUNT_ID="你的Cloudflare账户ID"
export R2_ACCESS_KEY_ID="你的R2 Access Key ID"
export R2_SECRET_ACCESS_KEY="你的R2 Secret Access Key"
export R2_BUCKET="你的R2桶名"3. 七牛认证
配置七牛认证：
qshell account "${QINIU_AK}" "${QINIU_SK}" default4. 七牛批量下载
生成七牛下载配置文件，全量迁移整个空间时用：
cat > qdownload.json <<EOF
{
"dest_dir": "${LOCAL_DIR}",
"bucket": "${QINIU_BUCKET}",
"prefix": "",
"check_exists": true
}
EOF
开始从七牛下载到本地：
qshell qdownload qdownload.json5. 统计
下载完成后先看本地统计：
find "${LOCAL_DIR}" -type f | wc -l
du -sh "${LOCAL_DIR}"6. 创建rclone 配置文件。
mkdir -p ~/.config/rclone
cat > ~/.config/rclone/rclone.conf <<EOF
[qiniu]
type = s3
provider = Other
access_key_id = ${QINIU_AK}
secret_access_key = ${QINIU_SK}
endpoint = https://s3-cn-east-1.qiniucs.com
acl = private
no_check_bucket = true
[r2]
type = s3
provider = Cloudflare
access_key_id = ${R2_ACCESS_KEY_ID}
secret_access_key = ${R2_SECRET_ACCESS_KEY}
endpoint = https://${R2_ACCOUNT_ID}.r2.cloudflarestorage.com
acl = private
no_check_bucket = true
EOF
7. 创建rclone 配置文件
然后创建配置文件
mkdir -p ~/.config/rclone
cat > ~/.config/rclone/rclone.conf <<EOF
[qiniu]
type = s3
provider = Other
access_key_id = ${QINIU_AK}
secret_access_key = ${QINIU_SK}
endpoint = https://s3-cn-east-1.qiniucs.com
acl = private
no_check_bucket = true
[r2]
type = s3
provider = Cloudflare
access_key_id = ${R2_ACCESS_KEY_ID}
secret_access_key = ${R2_SECRET_ACCESS_KEY}
endpoint = https://${R2_ACCOUNT_ID}.r2.cloudflarestorage.com
acl = private
no_check_bucket = true
EOF8. 演练测试
把七牛迁到 R2，先演练测试：
rclone sync "qiniu:${QINIU_BUCKET}" "r2:${R2_BUCKET}" --dry-run -P9. 执行上传任务
确认无误后正式执行上传同步任务：
rclone copy "qiniu:${QINIU_BUCKET}" "r2:${R2_BUCKET}" -P --transfers 16 --checkers 32 --fast-list10. 确认检验
迁移完成后，确认七牛和 R2 的文件完全一致先跑 <code>check</code>检查一般。
rclone check "qiniu:${QINIU_BUCKET}" "r2:${R2_BUCKET}" --size-only -P如果结果是 <code>0 differences found</code>，就说明两边文件数量和大小一致。
五、实现 R2 和七牛云双写备份
虽然 Cloudflare R2 很香，但我个人并不习惯把所有数据放在单一平台。
万一哪天账号出现异常或者平台出现故障，图床中的所有图片都有可能受到影响。
因此我又保留了七牛云作为备份存储，实现：
上传图片
│
├── Cloudflare R2（主存储）
│
└── 七牛云（备份存储）这样即使其中一个存储出现问题，也可以快速切换到另一个存储服务。
下面是我自己配合Typora 上传图片的脚本，你改改自己的配置直接安装好环境直接用就行，或者让你的AI 帮你改也可以。
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
PicGo 风格的命令行图片上传工具：R2 主写入，七牛云备份双写。
功能特色：
- 支持从命令行上传本地图片文件或 HTTP/HTTPS 图片地址。
- 先写入 Cloudflare R2，再把同一份对象写入七牛云作为备份。
- 输出 Markdown 图片链接，格式为：
- 如果输入地址已经属于 PUBLIC_DOMAIN，会直接复用原链接，避免重复上传。
- JPG/JPEG/JFIF、PNG、WebP、静态 GIF 会先尝试无损优化。
- 如果图片仍然超过 5 MB，会在不改变宽高尺寸的前提下继续压缩，
方便满足微信公众号等平台的图片大小限制。
安装依赖：
uv pip install boto3 qiniu loguru Pillow
可选系统依赖，用于更好地无损优化 JPEG：
brew install jpeg-turbo
# Linux 可安装发行版中提供 jpegtran 的软件包
必填环境变量：
R2_ACCOUNT_ID
R2_ACCESS_KEY_ID
R2_SECRET_ACCESS_KEY
R2_BUCKET
QINIU_ACCESS_KEY
QINIU_SECRET_KEY
QINIU_BUCKET_NAME
PUBLIC_DOMAIN
可选环境变量：
R2_ENDPOINT_URL # 默认：https://<R2_ACCOUNT_ID>.r2.cloudflarestorage.com
BUSINESS_MEDIA_PATH # 默认：md
MAX_IMAGE_SIZE_BYTES # 默认：5000000
使用示例：
python picgo_public.py /path/to/image.png
python picgo_public.py https://example.com/image.jpg
"""
import datetime
import hashlib
import io
import os
import shutil
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
from loguru import logger
from qiniu import Auth, BucketManager, put_data
import boto3
from botocore.client import Config
try:
from PIL import Image
except ImportError:
Image = None
BASE_DIR = Path(__file__).resolve().parent
LOG_DIR = BASE_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)
LOG_FILE = LOG_DIR / f"{datetime.datetime.now().strftime('%Y%m')}.log"
logger.add(str(LOG_FILE), retention="10 days")
QINIU_ACCESS_KEY = os.getenv(
"QINIU_ACCESS_KEY",
"",
)
QINIU_SECRET_KEY = os.getenv(
"QINIU_SECRET_KEY",
"",
)
QINIU_BUCKET_NAME = os.getenv("QINIU_BUCKET_NAME", "")
R2_ACCOUNT_ID = os.getenv("R2_ACCOUNT_ID", "")
R2_ACCESS_KEY_ID = os.getenv("R2_ACCESS_KEY_ID", "")
R2_SECRET_ACCESS_KEY = os.getenv(
"R2_SECRET_ACCESS_KEY",
"",
)
R2_BUCKET = os.getenv("R2_BUCKET", "")
R2_ENDPOINT_URL = os.getenv(
"R2_ENDPOINT_URL",
f"https://{R2_ACCOUNT_ID}.r2.cloudflarestorage.com" if R2_ACCOUNT_ID else "",
)
PUBLIC_DOMAIN = os.getenv("PUBLIC_DOMAIN", "")
BUSINESS_MEDIA_PATH = os.getenv("BUSINESS_MEDIA_PATH", "md")
MAX_IMAGE_SIZE_BYTES = int(os.getenv("MAX_IMAGE_SIZE_BYTES", "5000000"))
@dataclass
class FilePayload:
content: bytes
original_filename: str
size: int
source_path: str
@dataclass
class UploadResult:
provider: str
key: str
url: str
class StorageBackend(ABC):
def __init__(self, base_path: str):
self.base_path = (base_path or "").strip("/")
def build_object_key(self, file_name: str) -> str:
file_name = os.path.basename(file_name)
date_folder = datetime.datetime.now().strftime("%Y/%m")
if self.base_path:
return f"{self.base_path}/{date_folder}/{file_name}"
return f"{date_folder}/{file_name}"
def generate_md5_filename(
self,
file_data: bytes,
original_filename: Optional[str] = None,
) -> str:
md5_hash = hashlib.md5(file_data).hexdigest()
extension = self.get_file_extension(original_filename or "")
return f"{md5_hash}{extension}" if extension else md5_hash
@staticmethod
def get_file_extension(file_path: str) -> str:
if file_path.startswith(("http://", "https://")):
parsed_url = urllib.parse.urlparse(file_path)
file_path = parsed_url.path
return os.path.splitext(file_path)[1].lower()
@abstractmethod
def upload(self, payload: FilePayload, object_key: str) -> Optional[UploadResult]:
raise NotImplementedError
class QiniuStorage(StorageBackend):
def __init__(
self,
access_key: str = QINIU_ACCESS_KEY,
secret_key: str = QINIU_SECRET_KEY,
bucket_name: str = QINIU_BUCKET_NAME,
domain: str = PUBLIC_DOMAIN,
base_path: str = BUSINESS_MEDIA_PATH,
):
super().__init__(base_path)
self.access_key = access_key
self.secret_key = secret_key
self.bucket_name = bucket_name
self.domain = domain.rstrip("/")
self.enabled = all(
[self.access_key, self.secret_key, self.bucket_name, self.domain]
)
self.auth = Auth(self.access_key, self.secret_key) if self.enabled else None
self.bucket_manager = BucketManager(self.auth) if self.enabled else None
if self.enabled:
logger.info(f"QiniuStorage initialized for bucket: {bucket_name}")
else:
logger.warning(
"Qiniu backup is disabled because required env vars are missing"
)
def get_file_url(self, key: str, expires: Optional[int] = None) -> str:
base_url = f"http://{self.domain}/{key}"
if expires:
return self.auth.private_download_url(base_url, expires=expires)
return base_url
def upload(self, payload: FilePayload, object_key: str) -> Optional[UploadResult]:
if not self.enabled:
return None
try:
token = self.auth.upload_token(self.bucket_name, object_key, 3600)
_, info = put_data(token, object_key, payload.content)
if info.status_code != 200:
logger.error(f"Qiniu upload failed: {info.error}")
return None
logger.info(f"Qiniu backup uploaded successfully: {object_key}")
return UploadResult(
provider="qiniu",
key=object_key,
url=self.get_file_url(object_key),
)
except Exception as exc:
logger.error(f"Error uploading file to Qiniu: {exc}")
return None
class R2Storage(StorageBackend):
def __init__(
self,
account_id: str = R2_ACCOUNT_ID,
access_key_id: str = R2_ACCESS_KEY_ID,
secret_access_key: str = R2_SECRET_ACCESS_KEY,
bucket_name: str = R2_BUCKET,
domain: str = PUBLIC_DOMAIN,
base_path: str = BUSINESS_MEDIA_PATH,
endpoint_url: str = R2_ENDPOINT_URL,
):
super().__init__(base_path)
self.account_id = account_id
self.access_key_id = access_key_id
self.secret_access_key = secret_access_key
self.bucket_name = bucket_name
self.domain = domain.rstrip("/")
self.endpoint_url = endpoint_url
if not all([
self.account_id,
self.access_key_id,
self.secret_access_key,
self.bucket_name,
]):
raise ValueError("R2 is required but related env vars are missing")
if boto3 is None or Config is None:
raise ImportError(
"boto3 is required for R2 uploads. Please install it with pip install boto3"
)
self.client = boto3.client(
"s3",
endpoint_url=self.endpoint_url,
aws_access_key_id=self.access_key_id,
aws_secret_access_key=self.secret_access_key,
region_name="auto",
config=Config(signature_version="s3v4"),
)
logger.info(f"R2Storage initialized for bucket: {bucket_name}")
def get_file_url(self, key: str) -> str:
if self.domain:
return f"https://{self.domain}/{key}"
return f"{self.endpoint_url}/{self.bucket_name}/{key}"
def upload(self, payload: FilePayload, object_key: str) -> Optional[UploadResult]:
self.client.put_object(
Bucket=self.bucket_name,
Key=object_key,
Body=payload.content,
ContentLength=payload.size,
)
logger.info(f"R2 primary upload successful: {object_key}")
return UploadResult(
provider="r2",
key=object_key,
url=self.get_file_url(object_key),
)
class FileLoader:
@staticmethod
def load(path_or_url: str) -> FilePayload:
try:
if path_or_url.startswith(("http://", "https://")):
return FileLoader._load_from_url(path_or_url)
return FileLoader._load_from_file(path_or_url)
except Exception as exc:
raise ValueError(f"Error reading file content: {exc}") from exc
@staticmethod
def _load_from_url(url: str) -> FilePayload:
logger.info(f"Downloading file from URL: {url}")
filename = os.path.basename(urllib.parse.urlparse(url).path)
if not filename:
filename = f"download_{int(time.time())}"
response = urllib.request.urlopen(url)
if response.status != 200:
raise ValueError(
f"Failed to download file from URL: HTTP {response.status}"
)
content = response.read()
return FilePayload(
content=content,
original_filename=filename,
size=len(content),
source_path=url,
)
@staticmethod
def _load_from_file(file_path: str) -> FilePayload:
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
with open(file_path, "rb") as file:
content = file.read()
return FilePayload(
content=content,
original_filename=os.path.basename(file_path),
size=len(content),
source_path=file_path,
)
class ImageOptimizer:
PILLOW_FORMATS = {".png", ".webp", ".gif"}
JPEG_FORMATS = {".jpg", ".jpeg", ".jfif"}
@classmethod
def optimize(cls, payload: FilePayload) -> FilePayload:
extension = StorageBackend.get_file_extension(payload.original_filename)
original_size = payload.size
optimized_content = None
if extension in cls.JPEG_FORMATS:
optimized_content = cls._optimize_jpeg(payload.content)
elif extension in cls.PILLOW_FORMATS:
optimized_content = cls._optimize_with_pillow(payload.content, extension)
if optimized_content and len(optimized_content) < payload.size:
payload = FilePayload(
content=optimized_content,
original_filename=payload.original_filename,
size=len(optimized_content),
source_path=payload.source_path,
)
logger.info(f"Image optimized: {original_size} -> {payload.size} bytes")
if payload.size <= MAX_IMAGE_SIZE_BYTES:
return payload
compressed_content = cls._compress_to_limit(
payload.content,
extension,
MAX_IMAGE_SIZE_BYTES,
)
if not compressed_content or len(compressed_content) >= payload.size:
logger.warning(
f"Image still exceeds {MAX_IMAGE_SIZE_BYTES} bytes after "
"compression attempts"
)
return payload
logger.info(
f"Image compressed for size limit: {payload.size} -> "
f"{len(compressed_content)} bytes"
)
return FilePayload(
content=compressed_content,
original_filename=payload.original_filename,
size=len(compressed_content),
source_path=payload.source_path,
)
@staticmethod
def _optimize_jpeg(content: bytes) -> Optional[bytes]:
jpegtran = shutil.which("jpegtran")
if not jpegtran:
logger.info("Skip JPEG lossless optimization because jpegtran is missing")
return None
result = subprocess.run(
[jpegtran, "-copy", "none", "-optimize"],
input=content,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
if result.returncode != 0:
logger.warning(f"jpegtran optimization failed: {result.stderr.decode()}")
return None
if ImageOptimizer._has_same_pixels(content, result.stdout):
return result.stdout
logger.warning("jpegtran output skipped because pixel verification failed")
return None
@classmethod
def _optimize_with_pillow(cls, content: bytes, extension: str) -> Optional[bytes]:
if Image is None:
logger.info("Skip image optimization because Pillow is not installed")
return None
try:
with Image.open(io.BytesIO(content)) as image:
if getattr(image, "is_animated", False):
logger.info("Skip animated GIF optimization")
return None
output = io.BytesIO()
if extension == ".png":
image.save(output, format="PNG", optimize=True, compress_level=9)
elif extension == ".webp":
image.save(
output,
format="WEBP",
lossless=True,
quality=100,
method=6,
)
elif extension == ".gif":
image.save(output, format="GIF", optimize=True)
else:
return None
optimized = output.getvalue()
if cls._has_same_pixels(content, optimized):
return optimized
except Exception as exc:
logger.warning(f"Lossless image optimization skipped: {exc}")
return None
@classmethod
def _compress_to_limit(
cls,
content: bytes,
extension: str,
max_bytes: int,
) -> Optional[bytes]:
if Image is None:
logger.info("Skip size limit compression because Pillow is not installed")
return None
if extension in cls.JPEG_FORMATS:
return cls._compress_jpeg_to_limit(content, max_bytes)
if extension == ".webp":
return cls._compress_webp_to_limit(content, max_bytes)
if extension == ".png":
return cls._compress_png_to_limit(content, max_bytes)
if extension == ".gif":
return cls._compress_static_gif_to_limit(content, max_bytes)
return None
@staticmethod
def _compress_jpeg_to_limit(content: bytes, max_bytes: int) -> Optional[bytes]:
with Image.open(io.BytesIO(content)) as image:
image = image.convert("RGB")
return ImageOptimizer._quality_search(
image=image,
image_format="JPEG",
max_bytes=max_bytes,
min_quality=35,
save_options={"optimize": True, "progressive": True},
)
@staticmethod
def _compress_webp_to_limit(content: bytes, max_bytes: int) -> Optional[bytes]:
with Image.open(io.BytesIO(content)) as image:
image = image.convert("RGBA" if image.mode == "RGBA" else "RGB")
return ImageOptimizer._quality_search(
image=image,
image_format="WEBP",
max_bytes=max_bytes,
min_quality=45,
save_options={"method": 6},
)
@staticmethod
def _compress_png_to_limit(content: bytes, max_bytes: int) -> Optional[bytes]:
candidates = []
with Image.open(io.BytesIO(content)) as image:
if getattr(image, "is_animated", False):
return None
for colors in (256, 192, 128, 96, 64, 32):
output = io.BytesIO()
palette_image = image.convert("RGBA").quantize(
colors=colors,
method=Image.Quantize.FASTOCTREE,
)
palette_image.save(
output,
format="PNG",
optimize=True,
compress_level=9,
)
candidate = output.getvalue()
if len(candidate) <= max_bytes:
candidates.append(candidate)
return max(candidates, key=len) if candidates else None
@staticmethod
def _compress_static_gif_to_limit(content: bytes, max_bytes: int) -> Optional[bytes]:
candidates = []
with Image.open(io.BytesIO(content)) as image:
if getattr(image, "is_animated", False):
logger.info("Skip animated GIF size limit compression")
return None
for colors in (256, 192, 128, 96, 64, 32):
output = io.BytesIO()
palette_image = image.convert("RGB").quantize(colors=colors)
palette_image.save(output, format="GIF", optimize=True)
candidate = output.getvalue()
if len(candidate) <= max_bytes:
candidates.append(candidate)
return max(candidates, key=len) if candidates else None
@staticmethod
def _quality_search(
image: "Image.Image",
image_format: str,
max_bytes: int,
min_quality: int,
save_options: dict,
) -> Optional[bytes]:
best = None
low = min_quality
high = 95
while low <= high:
quality = (low + high) // 2
output = io.BytesIO()
image.save(output, format=image_format, quality=quality, **save_options)
candidate = output.getvalue()
if len(candidate) <= max_bytes:
best = candidate
low = quality + 1
else:
high = quality - 1
return best
@staticmethod
def _has_same_pixels(original: bytes, optimized: bytes) -> bool:
if Image is None:
return False
with Image.open(io.BytesIO(original)) as original_image:
with Image.open(io.BytesIO(optimized)) as optimized_image:
if original_image.size != optimized_image.size:
return False
return (
original_image.convert("RGBA").tobytes()
== optimized_image.convert("RGBA").tobytes()
)
class DualWriterUploader:
def __init__(
self,
primary_storage: Optional[StorageBackend] = None,
backup_storage: Optional[StorageBackend] = None,
):
self.primary_storage = primary_storage or R2Storage()
self.backup_storage = backup_storage or QiniuStorage()
@staticmethod
def is_public_domain_url(path_or_url: str) -> bool:
if not path_or_url.startswith(("http://", "https://")):
return False
parsed_url = urllib.parse.urlparse(path_or_url)
return (parsed_url.hostname or "").lower() == PUBLIC_DOMAIN.lower()
def upload_payload(self, payload: FilePayload) -> Optional[UploadResult]:
object_key = self.primary_storage.build_object_key(
self.primary_storage.generate_md5_filename(
payload.content,
payload.original_filename,
)
)
try:
primary_result = self.primary_storage.upload(payload, object_key)
except Exception as exc:
logger.error(f"R2 primary upload failed: {exc}")
return None
if primary_result is None:
return None
backup_result = self.backup_storage.upload(payload, object_key)
if backup_result is None:
logger.warning("Qiniu backup upload failed, but primary R2 upload succeeded")
return primary_result
def upload_with_retry(
self,
path_or_url: str,
max_retries: int = 20,
retry_interval: int = 2,
) -> Optional[str]:
if self.is_public_domain_url(path_or_url):
logger.info(f"Skip upload for existing public URL: {path_or_url}")
return path_or_url
retries = 0
while retries <= max_retries:
try:
payload = FileLoader.load(path_or_url)
payload = ImageOptimizer.optimize(payload)
result = self.upload_payload(payload)
if result is not None:
logger.info(f"File uploaded successfully after {retries} retries")
return result.url
except Exception as exc:
logger.error(
f"Error uploading file (attempt {retries + 1}/{max_retries}): {exc}"
)
retries += 1
if retries <= max_retries:
logger.warning(f"Upload failed, retrying ({retries}/{max_retries})...")
time.sleep(retry_interval)
logger.error(f"Failed to upload file after {max_retries} attempts")
return None
def main() -> None:
if len(sys.argv) < 2:
print("调用错误, 图片格式不对")
sys.exit(1)
uploader = DualWriterUploader()
uploaded_urls = []
for file_path in sys.argv[1:]:
file_url = uploader.upload_with_retry(file_path)
if file_url:
uploaded_urls.append(file_url)
if uploaded_urls:
print("Upload Success:")
for file_url in uploaded_urls:
print(file_url)
if __name__ == "__main__":
main()
六、配置Typora 上传图片
打开Typora 设置图像那里，按照如图设置即可，输入脚本命令：
（你的env绝对环境路径）/python picgo_public.py 最后点击一下验证是否上传成功。
七、总结
好了，今天就先写到这里。
整个迁移过程其实并不复杂，如果过程中有问题，你可以把这篇文章直接复制给AI 帮你来做。
迁移完后，我所有博客图片已经全部切换到了 Cloudflare R2。
当然，我个人还是建议不要把所有数据放在单一平台，所以文章里也介绍了双写备份方案。
即使某个平台出现问题，也能快速切换，不至于影响自己的图片数据丢了，同时还能保证图片正常访问。
如果你目前也在使用七牛云、腾讯云 COS、阿里云 OSS 等对象存储，并且同样被流量费用困扰，不妨抽点时间体验一下 Cloudflare R2。
最后，如果大家在迁移过程中遇到问题，欢迎在评论区留言交流。
如果本文对你有所帮助，欢迎 点赞、在看、转发 支持一下。
我是小欣，咱们下篇文章见。
扫码访问小程序中的本文
