Python脚本实现批量发布Markdown文章

import os
import frontmatter

def parse_markdown_file(file_path):
    """解析Markdown文件，提取元数据和纯正文。"""
    with open(file_path, 'r', encoding='utf-8') as f:
        post = frontmatter.load(f)
    
    title = post.get('title', '')
    tags = post.get('tags', [])
    date = post.get('date', None)
    category = post.get('category', '')
    content = post.content
    
    # 无Front Matter时的备用提取逻辑
    if not title:
        lines = content.split('\n')
        for line in lines:
            if line.startswith('# '):
                title = line.lstrip('# ').strip()
                break
        if not title:
            title = os.path.splitext(os.path.basename(file_path))[0]
    
    return {
        'title': title,
        'content': content,
        'tags': tags,
        'date': date,
        'category': category,
        'source_path': file_path
    }

import re
import os
import requests

def find_local_images(content, md_file_path):
    """在Markdown内容中查找所有本地图片的引用。"""
    pattern = r'!\[(.*?)\]\((.*?)\)'
    matches = re.findall(pattern, content)
    
    local_images = []
    md_dir = os.path.dirname(md_file_path)
    
    for alt_text, img_path in matches:
        if img_path.startswith(('http://', 'https://')):
            continue
            
        abs_path = img_path if os.path.isabs(img_path) else os.path.join(md_dir, img_path)
        
        if os.path.exists(abs_path):
            local_images.append({
                'alt': alt_text,
                'local_path': abs_path,
                'markdown_ref': img_path
            })
        else:
            print(f"警告：图片文件不存在 {abs_path}")
    
    return local_images

def upload_image_to_blog(local_image_path, alt_text=''):
    """
    调用博客/图床API上传单张图片。
    【重要】请将 upload_url 和 headers 替换为实际平台凭证。
    """
    upload_url = "https://api.your-blog.com/v1/upload"  # 示例占位符
    headers = {"Authorization": "Bearer YOUR_ACCESS_TOKEN"}
    
    with open(local_image_path, 'rb') as f:
        files = {'file': (os.path.basename(local_image_path), f, 'image/png')}
        try:
            resp = requests.post(upload_url, headers=headers, files=files, timeout=30)
            resp.raise_for_status()
            result = resp.json()
            image_url = result.get('url')
            if not image_url:
                print(f"警告：上传成功但未返回URL")
                return None
            print(f"图片上传成功：{local_image_path}")
            return image_url
        except Exception as e:
            print(f"图片上传失败 {local_image_path}: {e}")
            return None

def process_and_replace_images(parsed_post):
    """处理文章中的本地图片：上传并替换链接。"""
    content = parsed_post['content']
    local_images = find_local_images(content, parsed_post['source_path'])
    
    if not local_images:
        return parsed_post
    
    for img_info in local_images:
        online_url = upload_image_to_blog(img_info['local_path'], img_info['alt'])
        if online_url:
            old_syntax = f'![{img_info["alt"]}]({img_info["markdown_ref"]})'
            new_syntax = f'![{img_info["alt"]}]({online_url})'
            content = content.replace(old_syntax, new_syntax)
        else:
            print(f"图片上传失败，链接未替换。")
    
    parsed_post['content'] = content
    return parsed_post

def publish_post_to_blog(post_data):
    """
    调用博客平台的API发布文章。
    【重要】请将 api_url 和 headers 替换为实际平台凭证。
    """
    api_url = "https://api.your-blog.com/v1/posts"  # 示例占位符
    headers = {
        "Authorization": "Bearer YOUR_ACCESS_TOKEN",
        "Content-Type": "application/json"
    }
    
    # 文本标签转ID映射表
    tag_name_to_id = {"Python": 1, "踩坑记录": 2, "自动化": 3}
    tag_ids = [tag_name_to_id[t] for t in post_data.get('tags', []) if t in tag_name_to_id]
    
    payload = {
        "title": post_data['title'],
        "content": post_data['content'],
        "status": "publish",  # 调试阶段建议改为 "draft"
        "tags": tag_ids,
        "category": post_data.get('category', ''),
    }
    
    try:
        resp = requests.post(api_url, json=payload, headers=headers, timeout=30)
        resp.raise_for_status()
        print(f"发布成功！标题：{post_data['title']}")
        return True
    except Exception as e:
        print(f"发布失败 '{post_data['title']}': {e}")
        return False

#!/usr/bin/env python3
import os
import argparse
from time import sleep

def main(markdown_dir):
    if not os.path.isdir(markdown_dir):
        print(f"错误：目录不存在 {markdown_dir}")
        return
    
    md_files = [
        os.path.join(root, file)
        for root, _, files in os.walk(markdown_dir)
        for file in files if file.lower().endswith('.md')
    ]
    
    if not md_files:
        print(f"在目录 {markdown_dir} 中未找到.md文件。")
        return
    
    print(f"找到 {len(md_files)} 篇Markdown文章，开始处理...")
    success_count = 0
    
    for i, md_file in enumerate(md_files, 1):
        print(f"\n--- 正在处理第 {i}/{len(md_files)} 篇: {os.path.basename(md_file)} ---")
        parsed = parse_markdown_file(md_file)
        parsed = process_and_replace_images(parsed)
        
        if publish_post_to_blog(parsed):
            success_count += 1
        sleep(1)
    
    print(f"\n处理完成！成功发布 {success_count}/{len(md_files)} 篇文章。")

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='批量发布Markdown文章到博客')
    parser.add_argument('dir', help='包含.md文件的目录路径')
    args = parser.parse_args()
    main(args.dir)