2025-11-24-Mon-T-博客文章同步脚本

背景：几年前我通过hexo博客框架自己搭了一套自己的博客，托管在github io服务器上。最近我在博客园也开通了博客，用作backup。

如果手动地同步github io上的文章到博客园，工作量有点大，并且github io文章中引用的图片大部分也在自定义的github图床上，如果github图床出问题，博客园文章也会受影响。

因此我打算将github io上的文章全量同步至博客园，并且将文章中的图片也上传至博客园，同时替换图片引用。

博客园账号

开通博客园博客高级功能后， 在博客园-->设置-->博客设置-->页脚HTML代码-->其他设置 开通允许MetaWeblog博客客户端访问。复制登录名和访问令牌。将其保存在config.json文件中, 例如:

{
	"url":  "https://rpc.cnblogs.com/metaweblog/fei" ,  // metaweblog访问地址，需要修改为自己账号对应的地址
	"username": "fei",
	"password": "访问令牌",
	"hexo_root_path": "D:/Users/fei/gitrepo/myblog/myblog", // hexo博客根目录
  	"bash_path": "D:\\Program Files\\Git\\bin\\bash.exe", // windows githash 程序目录
	"hexo_path": "/d/Users/fei/gitrepo/myblog/node_modules/.bin/hexo" // hexo命令目录
} 

图片上传与文章同步脚本 - python

将hexo中的文章同步至博客园，再将hexo 重新部署。这样再通过typora进行脚本执行时，就可以完成hexo博客和博客园两处的博客文章上传或更新了。

import xmlrpc.client
import ssl
import os
import json
import sys
import re # 导入 re 模块用于正则表达式操作
import requests # 需要安装 requests 库：pip install requests
import subprocess # 【新增】导入 subprocess 模块用于执行系统命令
import shlex # 【新增】用于安全地引用字符串
import time # 【新增】导入 time 模块用于休眠

# 设置标准输出编码为 UTF-8，以支持 Emoji 和 Unicode 字符
if sys.platform.startswith('win'):
    sys.stdout.reconfigure(encoding='utf-8')


ssl._create_default_https_context = ssl._create_unverified_context

rootPath = os.path.abspath(os.path.dirname(__file__))
with open(os.path.join(rootPath,"config.json"),"rb") as f:
    config = json.loads(f.read())

# 用于匹配所有图片引用的正则表达式
# Group 1: Alt text
# Group 2: 图片路径或URL
ALL_IMAGE_PATTERN = re.compile(r'!\[(.*?)\]\((.*?)\)')


POST_DELAY_SECONDS = 3.5 # 推荐 3 到 4 秒，确保每分钟发文少于 20 篇


# ----------------------------------------------------------------------------------
# 【新增】获取最新文章列表并查找目标文章 ID 的函数
# ----------------------------------------------------------------------------------
def getExistingPostId(title):
    """
    通过检查最新文章列表，查找给定标题的文章 ID。
    
    注意：MetaWeblog API 只能获取最近的 N 篇文章 (例如 20 篇)。
    如果文章非常旧，可能无法找到。N 的值通常由博客平台决定。
    """
    print(f"🔄 尝试在最近的文章中查找标题为 '{title}' 的文章...")
    
    # 博客园 MetaWeblog API URL
    url = config["url"]
    username = config['username']
    password = config['password']
    
    # 获取最近文章的篇数，例如获取最近 50 篇
    NUMBER_OF_POSTS_TO_CHECK = 50 

    proxy = xmlrpc.client.ServerProxy(url)
    try:
        # 调用 getRecentPosts 方法
        recent_posts = proxy.metaWeblog.getRecentPosts(
            '', username, password, NUMBER_OF_POSTS_TO_CHECK
        )
        
        for post in recent_posts:
            # post 字典中的 'title' 键存储文章标题
            if post['title'] == title:
                print(f"✅ 找到已存在的文章！Post ID: {post['postid']}")
                # post['postid'] 存储文章的唯一标识符
                return post['postid'] 
                
        print("➡️ 未找到同名文章，将创建新文章。")
        return None
        
    except Exception as e:
        print(f"❌ 查找现有文章时出错: {e}")
        return None

# ----------------------------------------------------------------------------------
# 【修改】uploadArticle 函数，实现创建或更新逻辑
# ----------------------------------------------------------------------------------
def uploadArticle(articles):
    for article in articles:
        # ... (读取文件内容和图片处理逻辑保持不变) ...
        with open(article,"r",encoding="utf8") as f:
            original_data = f.read()
        
        # data_to_post = original_data
        data_to_post = original_data
        
        # 1. 获取当前MD文件所在目录，用于解析本地图片的相对路径
        md_dir = os.path.dirname(article)
        
        # 2. 查找所有的图片引用
        # 使用 finditer 可以找到所有匹配项及其在字符串中的位置
        matches = list(ALL_IMAGE_PATTERN.finditer(data_to_post))
        
        # 使用字典来存储已上传图片的映射，避免重复上传相同的图片
        uploaded_map = {} 
        
        # 3. 逐个处理图片
        for match in matches:
            alt_text = match.group(1)
            source_path = match.group(2) # 原始的路径/URL
            original_full_match = match.group(0) # 完整的 ![alt](path) 字符串
            
            # 如果图片路径已经在映射中，则直接使用已上传的URL
            if source_path in uploaded_map:
                new_url = uploaded_map[source_path]
                print(f"➡️ 图片已处理: {source_path}，使用缓存URL: {new_url}")
            else:
                # 确定图片在本地文件系统中的绝对路径或保持网络URL
                if source_path.startswith('http'):
                    upload_target = source_path
                else:
                    # 相对路径转绝对路径
                    upload_target = os.path.normpath(os.path.join(md_dir, source_path))
                
                # 4. 调用上传函数
                new_url = uploadImage(upload_target)
                
                if new_url:
                    uploaded_map[source_path] = new_url # 存入缓存
                else:
                    print(f"⚠️ 图片 {source_path} 上传失败，将保留原地址。")
                    continue
            
            # 5. 替换文章内容中的链接
            # 构建新的引用字符串
            new_image_reference = f"![{alt_text}]({new_url})"
            # 替换内容，注意我们是在 data_to_post 中替换
            data_to_post = data_to_post.replace(original_full_match, new_image_reference, 1)
        
        # 假设图片处理成功完成， data_to_post 现在包含博客园的图片 URL
        
        # 获取标题 (作为查找和发布的依据)
        title = os.path.basename(article)[:-3]

        
        
        # 1. 检查文章是否已存在
        post_id = getExistingPostId(title)
        
        # 2. 构造文章数据
        post = dict(
            # dateCreated 仅对 newPost 有意义，editPost 不需要
            description = data_to_post, 
            title = title,                      
            categories = ['[Markdown]'],         
        )
        
        proxy = xmlrpc.client.ServerProxy(config["url"])
        userName = config["url"].split("/")[-1]
        
        if post_id:
            # 3. 文章已存在：执行更新 (editPost)
            try:
                # metaWeblog.editPost(postid, username, password, struct post, bool publish)
                success = proxy.metaWeblog.editPost(
                    post_id, config['username'], config['password'], post, True
                )
                if success:
                    print(f"🎉 文章更新成功 (Post ID: {post_id})")
                else:
                    raise Exception("API 返回 False")
                    
                # 构造博客园文章链接
                article_url = f"https://www.cnblogs.com/{userName}/p/{post_id}.html"

            except Exception as e:
                print(f"❌ 文章更新失败 (editPost): {e}")
                continue # 处理下一个文件
        else:
            # 4. 文章不存在：执行创建 (newPost)
            try:
                post['dateCreated'] = xmlrpc.client.DateTime() # 确保新文章有日期
                # metaWeblog.newPost(blogid, username, password, struct post, bool publish)
                new_post_id = proxy.metaWeblog.newPost(
                    '', config['username'], config['password'], post, True
                )
                print(f"🎉 文章创建成功！新 Post ID: {new_post_id}")
                
                # 构造博客园文章链接
                article_url = f"https://www.cnblogs.com/{userName}/p/{new_post_id}.html"

            except Exception as e:
                print(f"❌ 文章创建失败 (newPost): {e}")
                # 如果创建失败，可能需要更长的等待时间再尝试下一个，或者直接跳过
                print(f"⚠️ 遇到频率限制，暂停 {POST_DELAY_SECONDS * 2} 秒...")
                time.sleep(POST_DELAY_SECONDS * 2) 
                continue # 处理下一个文件

        print(f"✅ 文章发布/更新成功: {article_url}")
        # 5. 【关键】在处理完一篇文章后，添加强制延迟
        print(f"⏸️ 暂停 {POST_DELAY_SECONDS} 秒，以避免频率限制...")
        time.sleep(POST_DELAY_SECONDS)



    # 确保在函数定义外部定义了 deployHexo 或将其注释掉，以防报错

# 将 Windows 绝对路径 hexo_root_path 转换为 Git Bash 兼容的路径
def to_bash_path(win_path):
    """
    将 Windows 绝对路径转换为 Git Bash/MinGW 路径。
    例如: C:/Users/name/ -> /c/Users/name/
    """
    if not win_path:
        return win_path
        
    # 统一斜杠方向
    normalized_path = win_path.replace('\\', '/')
    
    # 检查是否以驱动器字母开头 (如 D:/ 或 C:/)
    if re.match(r'^[A-Za-z]:/', normalized_path):
        # 转换为 /d/Users/fei...
        drive = normalized_path[0].lower()
        rest_of_path = normalized_path[2:]
        return f'/{drive}{rest_of_path}'
        
    return normalized_path # 如果不是驱动器路径，则保持不变

# 定义 Hexo 命令的执行函数
def deployHexo():
    """
    执行 Hexo 部署命令 (hexo clean, hexo g, hexo d)，使用 Bash -c 选项进行安全封装。
    """
    hexo_root_path_win = config.get("hexo_root_path")
    bash_executable_path = config.get("bash_path")
    HEXO_EXE_PATH = config.get("hexo_path")
    
    # ... (路径检查代码保持不变) ...

    # 1. 将 Windows 路径转换为 Bash 可识别的路径
    hexo_root_path_bash = to_bash_path(hexo_root_path_win) 


    # 2. 构造所有命令，使用 Bash 风格的 && 串联
    # /d/Users/fei/gitrepo/myblog/node_modules/.bin/hexo
    commands_to_run = " && ".join([
        f"{HEXO_EXE_PATH} clean",
        f"{HEXO_EXE_PATH} generate",
        f"{HEXO_EXE_PATH} deploy"
    ])
    
    # 3. 构造要传给 Bash -c 的内部命令。
    # 我们使用 Bash 语法来封装整个命令。
    internal_command = f'cd {shlex.quote(hexo_root_path_bash)} && {commands_to_run}'
    
    # 4. 构造最终的执行命令：调用 Bash，并用 -c 传递 internal_command
    # 这里的关键是：我们不再依赖 shell=True 和 executable 的复杂结合。
    # 我们直接构造一个命令列表 (list of strings) 让 subprocess 执行 Bash。
    
    # 确保 Bash 路径被安全引用 (Windows路径中的空格问题)
    quoted_bash_path = shlex.quote(bash_executable_path)
    
    # 构造命令列表： [Bash可执行文件, -c, "内部命令"]
    cmd_list = [
        quoted_bash_path.strip("'\""), # 移除shlex可能添加的引号，确保Windows能找到文件
        "-c",
        internal_command # 内部命令包含 cd 和 hexo 串联
    ]

    print("\n--- 🚀 开始执行 Hexo 部署流程 (使用 Bash 列表模式) ---")
    print(f"Bash Command List: {cmd_list}")
    
    try:
        # 关键修改：
        # 1. 传入命令列表 (cmd_list)，而不是单个命令字符串
        # 2. 移除 shell=True 和 executable=... 参数
        # 这样做能最大限度地避免 Windows Shell 的解析干扰
        result = subprocess.run(
            cmd_list,
            check=True,         
            capture_output=True,
            text=True,
            encoding='utf-8'
        )
        print("✅ Hexo 部署命令全部成功完成。")

    except subprocess.CalledProcessError as e:
        print(f"❌ Bash 命令执行失败，请检查 Hexo 错误。")
        print(f"Stdout:\n{e.stdout}")
        print(f"Stderr:\n{e.stderr}")
        print("--- 终止 Hexo 部署流程 ---")
        return
    except FileNotFoundError:
        print(f"❌ 错误: 找不到 Bash 可执行文件: {quoted_bash_path}")
        print("--- 终止 Hexo 部署流程 ---")
        return
    except Exception as e:
        print(f"❌ 发生未知错误：{e}")
        print("--- 终止 Hexo 部署流程 ---")
        return

    print("--- 🎉 Hexo 部署流程全部完成！ ---")
def uploadImage(image_path_or_url):
    """
    上传单个图片，无论是本地路径还是网络URL。
    
    参数:
        image_path_or_url: 本地图片路径或网络图片的URL。
        
    返回:
        成功上传后返回的博客园图片URL字符串；失败则返回 None。
    """
    # 确定要上传的文件名和数据
    if image_path_or_url.startswith('http'):
        # **处理网络图片**：需要先下载图片
        try:

            response = requests.get(image_path_or_url, stream=True)
            response.raise_for_status() # 检查请求是否成功
            
            # 尝试从 URL 提取文件名和后缀
            baseName = image_path_or_url.split('/')[-1].split('?')[0]
            # 如果 URL 结尾没有明显文件名，则给一个默认名
            if not baseName or '.' not in baseName:
                 baseName = f"remote_image_{os.urandom(4).hex()}.png"

            imageData = response.content
            suffix = baseName.split(".")[-1]
            print(f"🌍 正在下载并上传网络图片: {image_path_or_url}")

        except Exception as e:
            print(f"❌ 错误：无法下载或处理网络图片 {image_path_or_url}: {e}")
            return None
    else:
        # **处理本地图片**：使用原来的逻辑
        if not os.path.exists(image_path_or_url):
            print(f"⚠️ 警告：本地图片文件不存在，跳过上传：{image_path_or_url}")
            return None
            
        with open(image_path_or_url,"rb") as f:
            imageData = f.read()

        baseName = os.path.basename(image_path_or_url)
        suffix = baseName.split(".")[-1]
        print(f"🖼️ 正在上传本地图片: {image_path_or_url}")
    
    
    file = dict(
        bits = imageData,
        name = baseName,
        type = f"image/{suffix}"
    )
    try:
        proxy = xmlrpc.client.ServerProxy(config["url"])
        s = proxy.metaWeblog.newMediaObject('', config['username'], config['password'],file)
        print(f"✨ 上传成功！新URL: {s['url']}")
        return s["url"]
    except Exception as e:
        print(f"❌ 图片上传到博客园失败: {e}")
        return None




if __name__ == '__main__':
    # 请确保在运行前安装 requests 库：pip install requests
    if len(sys.argv) <= 1:
        print("请提供要上传的 Markdown 文件路径作为命令行参数。")
    else:
        uploadArticle(sys.argv[1:])
        # 5. 自动执行 Hexo 部署 (如果已添加此功能)
        deployHexo() 

Typora设置导出命令

在Typora中设置自定义的文件导出命令：

`python D:\Users\fei\gitrepo\myblog\cnblogs_uploader\uploader.py ${currentPath}`

这样通过这个自定义导出，就可以完成两处的博客文章上传。如果有扩展，再修改脚本逻辑，即可实现多平台文章同步。

批量同步

由于前期积累了大量的文章未同步至博客园中，因此可以编写脚本批量将这些文章同步至博客园：

from uploader import uploadArticle # 导入刚刚编写上传文章方法。
import os
import glob
import sys
import time # 【新增】导入 time 模块用于休眠
# ... (其他 imports) ...

# -----------------------------------------------------------
# 配置 Markdown 文件所在的根文件夹
MARKDOWN_FOLDER_PATH = os.path.abspath("D:\\Users\\fei\\gitrepo\\myblog\\myblog\\source\\_posts") 
# -----------------------------------------------------------


def run_batch_upload(folder_path):
    """
    扫描指定文件夹，批量上传所有找到的 Markdown 文件。
    """
    
    # 解决 Windows 编码输出问题（如果适用）
    if sys.platform.startswith('win'):
        try:
            sys.stdout.reconfigure(encoding='utf-8')
        except Exception:
            pass 

    print(f"🔍 正在扫描文件夹: {folder_path} 查找 Markdown 文件...")
    
    # 使用 glob 递归查找所有 .md 文件
    # recursive=True 允许 glob 查找子文件夹
    md_files = glob.glob(os.path.join(folder_path, '**', '*.md'), recursive=True)
    
    if not md_files:
        print("❌ 未找到任何 .md 文件，请检查 MARKDOWN_FOLDER_PATH 配置是否正确。")
    else:
        print(f"✅ 找到 {len(md_files)} 个 Markdown 文件，开始上传...")
        
        # 调用导入的 uploadArticle 函数批量处理这些文件
        uploadArticle(md_files) 
        
    print("\n\n🎉 所有任务处理完成！")


if __name__ == '__main__':
    run_batch_upload(MARKDOWN_FOLDER_PATH)