Ubuntu
386 lines
14 KiB
Python
386 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
博客同步模块
|
|
将 Markdown 文章同步到个人博客系统
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
import requests
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional, Tuple
|
|
from datetime import datetime
|
|
|
|
# 博客 API 配置
|
|
BLOG_API_BASE = "http://129.211.173.87:8889/api"
|
|
BLOG_UPLOADS_DIR = "/home/ubuntu/nanocode/personal-blog/backend/uploads"
|
|
|
|
|
|
class BlogSync:
|
|
"""博客同步器"""
|
|
|
|
def __init__(self):
|
|
self.api_base = BLOG_API_BASE
|
|
self.uploads_dir = BLOG_UPLOADS_DIR
|
|
|
|
def _get_api_url(self, endpoint: str) -> str:
|
|
"""获取完整 API URL"""
|
|
return f"{self.api_base}{endpoint}"
|
|
|
|
def _generate_slug(self, title: str) -> str:
|
|
"""根据标题生成 URL slug"""
|
|
# 简单处理：取前30个字符，替换特殊字符
|
|
slug = re.sub(r'[^\w\u4e00-\u9fff]+', '-', title)
|
|
slug = slug.strip('-').lower()
|
|
if len(slug) > 50:
|
|
slug = slug[:50]
|
|
# 添加时间戳避免重复
|
|
timestamp = datetime.now().strftime("%m%d%H%M")
|
|
return f"{slug}-{timestamp}"
|
|
|
|
def _copy_image_to_uploads(self, image_path: str) -> Optional[str]:
|
|
"""
|
|
将图片复制到博客 uploads 目录，并插入数据库
|
|
返回相对于 uploads 的 URL 路径
|
|
"""
|
|
if not os.path.exists(image_path):
|
|
print(f" ⚠️ 图片不存在: {image_path}")
|
|
return None
|
|
|
|
try:
|
|
# 生成目标文件名：时间戳_原文件名
|
|
filename = os.path.basename(image_path)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
new_filename = f"{timestamp}_{filename}"
|
|
|
|
# 目标路径
|
|
dest_path = os.path.join(self.uploads_dir, new_filename)
|
|
|
|
# 复制文件
|
|
shutil.copy2(image_path, dest_path)
|
|
|
|
# 获取图片信息
|
|
file_size = os.path.getsize(dest_path)
|
|
width, height = self._get_image_size(dest_path)
|
|
|
|
# 生成缩略图
|
|
thumb_path = self._create_thumbnail(dest_path, new_filename)
|
|
|
|
# 插入数据库
|
|
self._save_image_to_db(
|
|
filename=new_filename,
|
|
original_name=filename,
|
|
file_path=f"/uploads/{new_filename}",
|
|
thumb_path=thumb_path,
|
|
file_size=file_size,
|
|
width=width,
|
|
height=height
|
|
)
|
|
|
|
print(f" ✅ 图片已复制并入库: {new_filename}")
|
|
return f"/uploads/{new_filename}"
|
|
|
|
except Exception as e:
|
|
print(f" ❌ 复制图片失败: {e}")
|
|
return None
|
|
|
|
def _get_image_size(self, image_path: str) -> tuple:
|
|
"""获取图片尺寸"""
|
|
try:
|
|
from PIL import Image as PILImage
|
|
with PILImage.open(image_path) as img:
|
|
return img.size
|
|
except Exception as e:
|
|
print(f" ⚠️ 获取图片尺寸失败: {e}")
|
|
return (None, None)
|
|
|
|
def _save_image_to_db(self, filename: str, original_name: str, file_path: str,
|
|
thumb_path: str, file_size: int, width: int, height: int):
|
|
"""将图片信息保存到数据库"""
|
|
import sys
|
|
blog_backend = "/home/ubuntu/nanocode/personal-blog/backend"
|
|
original_sys_path = sys.path.copy()
|
|
|
|
try:
|
|
if blog_backend not in sys.path:
|
|
sys.path.insert(0, blog_backend)
|
|
|
|
from database import SessionLocal
|
|
from models import Image as ImageModel
|
|
|
|
db = SessionLocal()
|
|
|
|
# 检查是否已存在
|
|
existing = db.query(ImageModel).filter(ImageModel.filename == filename).first()
|
|
if existing:
|
|
print(f" ⚠️ 图片已存在于数据库: {filename}")
|
|
db.close()
|
|
return
|
|
|
|
# 创建新记录
|
|
new_image = ImageModel(
|
|
filename=filename,
|
|
original_name=original_name,
|
|
file_path=file_path,
|
|
thumb_path=thumb_path,
|
|
file_size=file_size,
|
|
width=width,
|
|
height=height,
|
|
created_at=datetime.now()
|
|
)
|
|
|
|
db.add(new_image)
|
|
db.commit()
|
|
db.close()
|
|
print(f" ✅ 图片已入库: {filename}")
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ 图片入库失败: {e}")
|
|
finally:
|
|
sys.path = original_sys_path
|
|
|
|
def _create_thumbnail(self, image_path: str, filename: str) -> Optional[str]:
|
|
"""创建缩略图（使用 ImageMagick），返回缩略图URL路径"""
|
|
try:
|
|
thumb_dir = os.path.join(self.uploads_dir, "thumbnails")
|
|
os.makedirs(thumb_dir, exist_ok=True)
|
|
|
|
thumb_filename = f"thumb_{filename}"
|
|
thumb_path = os.path.join(thumb_dir, thumb_filename)
|
|
|
|
# 使用 ImageMagick convert 命令
|
|
cmd = f"convert '{image_path}' -resize 320x -quality 85 '{thumb_path}'"
|
|
result = os.system(cmd)
|
|
|
|
if result == 0 and os.path.exists(thumb_path):
|
|
return f"/uploads/thumbnails/{thumb_filename}"
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ 缩略图生成失败: {e}")
|
|
return None
|
|
|
|
def _extract_images_from_markdown(self, markdown_content: str, base_dir: str) -> List[Tuple[str, str]]:
|
|
"""
|
|
从 Markdown 中提取图片
|
|
返回 [(原始路径, 图片说明), ...]
|
|
"""
|
|
images = []
|
|
# 匹配  格式
|
|
pattern = r'!\[([^\]]*)\]\(([^)]+)\)'
|
|
|
|
for match in re.finditer(pattern, markdown_content):
|
|
alt_text = match.group(1)
|
|
image_path = match.group(2)
|
|
|
|
# 如果是相对路径，转换为绝对路径
|
|
if not image_path.startswith('/') and not image_path.startswith('http'):
|
|
image_path = os.path.join(base_dir, image_path)
|
|
|
|
images.append((image_path, alt_text))
|
|
|
|
return images
|
|
|
|
def _process_images(self, markdown_content: str, base_dir: str) -> Tuple[str, Dict[str, str]]:
|
|
"""
|
|
处理 Markdown 中的图片
|
|
返回 (处理后的 markdown, 图片映射 {原路径: 新URL})
|
|
"""
|
|
images = self._extract_images_from_markdown(markdown_content, base_dir)
|
|
image_map = {}
|
|
|
|
print(f"🖼️ 处理 {len(images)} 张图片...")
|
|
|
|
for original_path, alt_text in images:
|
|
# 复制到 uploads 目录
|
|
new_url = self._copy_image_to_uploads(original_path)
|
|
if new_url:
|
|
image_map[original_path] = new_url
|
|
|
|
# 替换 Markdown 中的图片路径
|
|
processed_markdown = markdown_content
|
|
for original_path, new_url in image_map.items():
|
|
# 使用正则替换，保留 alt 文本
|
|
escaped_path = re.escape(original_path)
|
|
pattern = rf'!\[([^\]]*)\]\({escaped_path}\)'
|
|
replacement = rf''
|
|
processed_markdown = re.sub(pattern, replacement, processed_markdown)
|
|
|
|
return processed_markdown, image_map
|
|
|
|
def _strip_frontmatter(self, markdown_content: str) -> str:
|
|
"""移除 frontmatter，只保留正文"""
|
|
if markdown_content.startswith('---'):
|
|
parts = markdown_content.split('---', 2)
|
|
if len(parts) >= 3:
|
|
return parts[2].strip()
|
|
return markdown_content
|
|
|
|
def sync_to_blog(self, markdown_path: str, title: str = None, author: str = None) -> Dict:
|
|
"""
|
|
同步文章到博客
|
|
|
|
Args:
|
|
markdown_path: Markdown 文件路径
|
|
title: 文章标题（可选，优先使用 frontmatter 中的）
|
|
author: 作者（可选）
|
|
|
|
Returns:
|
|
{"success": True/False, "post_id": id, "message": "..."}
|
|
"""
|
|
print("\n" + "=" * 60)
|
|
print("📝 同步到个人博客...")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
# 1. 读取 Markdown 文件
|
|
if not os.path.exists(markdown_path):
|
|
return {"success": False, "error": f"文件不存在: {markdown_path}"}
|
|
|
|
with open(markdown_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
base_dir = os.path.dirname(os.path.abspath(markdown_path))
|
|
|
|
# 2. 解析 frontmatter
|
|
frontmatter = {}
|
|
if content.startswith('---'):
|
|
parts = content.split('---', 2)
|
|
if len(parts) >= 3:
|
|
try:
|
|
import yaml
|
|
frontmatter = yaml.safe_load(parts[1]) or {}
|
|
except Exception as e:
|
|
print(f" ⚠️ frontmatter 解析失败: {e}")
|
|
|
|
# 3. 确定标题
|
|
if not title:
|
|
title = frontmatter.get('title', '')
|
|
if not title:
|
|
# 从正文第一行提取
|
|
lines = content.split('\n')
|
|
for line in lines:
|
|
line = line.strip()
|
|
if line.startswith('# '):
|
|
title = line[2:].strip()
|
|
break
|
|
if not title:
|
|
title = Path(markdown_path).stem
|
|
|
|
print(f"📄 文章标题: {title}")
|
|
|
|
# 4. 处理图片
|
|
processed_content, image_map = self._process_images(content, base_dir)
|
|
print(f" ✅ 成功处理 {len(image_map)} 张图片")
|
|
|
|
# 5. 移除 frontmatter，只保留正文
|
|
body_content = self._strip_frontmatter(processed_content)
|
|
|
|
# 6. 生成 slug
|
|
slug = self._generate_slug(title)
|
|
|
|
# 7. 准备文章数据
|
|
post_data = {
|
|
"title": title,
|
|
"content": body_content,
|
|
"slug": slug,
|
|
"is_published": True, # 直接发布
|
|
"is_draft": False,
|
|
"category_id": None, # 不设置分类
|
|
"summary": frontmatter.get('summary', '')[:500] if frontmatter.get('summary') else ''
|
|
}
|
|
|
|
# 8. 调用博客 API 创建文章
|
|
print(f"📤 创建博客文章...")
|
|
|
|
# 博客 API 需要认证，暂时使用管理员 token
|
|
# 由于博客 JWT 验证，这里直接操作数据库更简单
|
|
result = self._create_post_via_db(post_data)
|
|
|
|
if result.get('success'):
|
|
print(f"✅ 博客同步成功！")
|
|
print(f" 文章ID: {result.get('post_id')}")
|
|
print(f" 访问: http://129.211.173.87:8888/post/{slug}")
|
|
else:
|
|
print(f"❌ 博客同步失败: {result.get('error')}")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"❌ 同步异常: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def _create_post_via_db(self, post_data: Dict) -> Dict:
|
|
"""
|
|
直接操作数据库创建文章（绕过 JWT 认证）
|
|
"""
|
|
import sys
|
|
blog_backend = "/home/ubuntu/nanocode/personal-blog/backend"
|
|
original_path = sys.path.copy()
|
|
|
|
try:
|
|
# 添加博客后端路径
|
|
if blog_backend not in sys.path:
|
|
sys.path.insert(0, blog_backend)
|
|
|
|
from database import SessionLocal
|
|
from models import Post
|
|
from datetime import datetime
|
|
|
|
db = SessionLocal()
|
|
|
|
# 创建文章
|
|
new_post = Post(
|
|
title=post_data['title'],
|
|
content=post_data['content'],
|
|
slug=post_data['slug'],
|
|
is_published=post_data['is_published'],
|
|
is_draft=post_data['is_draft'],
|
|
category_id=post_data['category_id'],
|
|
summary=post_data.get('summary', ''),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now()
|
|
)
|
|
|
|
db.add(new_post)
|
|
db.commit()
|
|
db.refresh(new_post)
|
|
|
|
post_id = new_post.id
|
|
db.close()
|
|
|
|
return {
|
|
"success": True,
|
|
"post_id": post_id,
|
|
"slug": post_data['slug'],
|
|
"message": "文章创建成功"
|
|
}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": f"数据库操作失败: {e}"}
|
|
finally:
|
|
# 恢复原始 sys.path
|
|
sys.path = original_path
|
|
|
|
|
|
def sync_markdown_to_blog(markdown_path: str, title: str = None, author: str = None) -> Dict:
|
|
"""
|
|
便捷函数：同步 Markdown 到博客
|
|
"""
|
|
syncer = BlogSync()
|
|
return syncer.sync_to_blog(markdown_path, title, author)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试
|
|
import sys
|
|
if len(sys.argv) > 1:
|
|
result = sync_markdown_to_blog(sys.argv[1])
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
else:
|
|
print("用法: python3 blog_sync.py <markdown文件路径>")
|