Ubuntu
444 lines
16 KiB
Python
Executable File
444 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
baoyu-post-to-wechat 增强版
|
||
支持图文混排、多图上传、HTML内容
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
import requests
|
||
import yaml
|
||
import base64
|
||
from pathlib import Path
|
||
from typing import List, Dict, Optional, Union
|
||
import mimetypes
|
||
|
||
# 配置文件路径
|
||
EXTEND_PATH = "/home/ubuntu/.nanobot/workspace/skills/baoyu-post-to-wechat/EXTEND.md"
|
||
WEWRITE_CONFIG = "/home/ubuntu/wewrite/config.yaml"
|
||
|
||
class WeChatPublisher:
|
||
"""微信公众号发布器（增强版）"""
|
||
|
||
def __init__(self):
|
||
self.config = self._load_config()
|
||
self.access_token = None
|
||
|
||
def _load_config(self) -> Dict:
|
||
"""加载配置"""
|
||
config = {}
|
||
|
||
# 加载 EXTEND.md
|
||
if os.path.exists(EXTEND_PATH):
|
||
with open(EXTEND_PATH, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
if ':' in line and not line.strip().startswith('#'):
|
||
key, value = line.split(':', 1)
|
||
config[key.strip()] = value.strip()
|
||
|
||
# 回退到 WeWrite 配置
|
||
if 'app_id' not in config and os.path.exists(WEWRITE_CONFIG):
|
||
with open(WEWRITE_CONFIG, 'r', encoding='utf-8') as f:
|
||
wewrite_config = yaml.safe_load(f)
|
||
config['app_id'] = wewrite_config['wechat']['appid']
|
||
config['app_secret'] = wewrite_config['wechat']['secret']
|
||
config['default_author'] = wewrite_config['wechat']['author']
|
||
|
||
# 设置默认值
|
||
config.setdefault('default_theme', 'default')
|
||
config.setdefault('default_color', 'blue')
|
||
config.setdefault('need_open_comment', '1')
|
||
config.setdefault('only_fans_can_comment', '0')
|
||
|
||
return config
|
||
|
||
def get_access_token(self) -> str:
|
||
"""获取微信 access_token"""
|
||
if self.access_token:
|
||
return self.access_token
|
||
|
||
url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.config['app_id']}&secret={self.config['app_secret']}"
|
||
response = requests.get(url)
|
||
data = response.json()
|
||
|
||
if 'access_token' in data:
|
||
self.access_token = data['access_token']
|
||
return self.access_token
|
||
else:
|
||
raise Exception(f"获取 access_token 失败: {data}")
|
||
|
||
def upload_image(self, image_path: str) -> str:
|
||
"""上传图片到微信（永久素材）"""
|
||
access_token = self.get_access_token()
|
||
|
||
# 检查文件是否存在
|
||
if not os.path.exists(image_path):
|
||
raise FileNotFoundError(f"图片文件不存在: {image_path}")
|
||
|
||
# 获取MIME类型
|
||
mime_type, _ = mimetypes.guess_type(image_path)
|
||
if not mime_type:
|
||
mime_type = 'image/jpeg'
|
||
|
||
# 上传图片
|
||
url = f"https://api.weixin.qq.com/cgi-bin/material/add_material?access_token={access_token}&type=image"
|
||
|
||
with open(image_path, 'rb') as f:
|
||
files = {'media': (os.path.basename(image_path), f, mime_type)}
|
||
response = requests.post(url, files=files)
|
||
|
||
data = response.json()
|
||
|
||
if 'media_id' in data:
|
||
return data['media_id']
|
||
else:
|
||
raise Exception(f"上传图片失败: {data}")
|
||
|
||
def upload_image_for_body(self, image_path: str) -> str:
|
||
"""上传图片到微信（用于正文内容，返回可直接嵌入的URL）
|
||
使用 /cgi-bin/media/uploadimg API
|
||
"""
|
||
access_token = self.get_access_token()
|
||
|
||
# 检查文件是否存在
|
||
if not os.path.exists(image_path):
|
||
raise FileNotFoundError(f"图片文件不存在: {image_path}")
|
||
|
||
# 获取MIME类型
|
||
mime_type, _ = mimetypes.guess_type(image_path)
|
||
if not mime_type:
|
||
mime_type = 'image/jpeg'
|
||
|
||
# 上传图片（用于图文内容）
|
||
url = f"https://api.weixin.qq.com/cgi-bin/media/uploadimg?access_token={access_token}"
|
||
|
||
with open(image_path, 'rb') as f:
|
||
files = {'media': (os.path.basename(image_path), f, mime_type)}
|
||
response = requests.post(url, files=files)
|
||
|
||
data = response.json()
|
||
|
||
if 'url' in data:
|
||
return data['url']
|
||
else:
|
||
raise Exception(f"上传正文图片失败: {data}")
|
||
|
||
def upload_images(self, image_paths: List[str]) -> Dict[str, str]:
|
||
"""批量上传图片（用于正文）"""
|
||
results = {}
|
||
for img_path in image_paths:
|
||
try:
|
||
img_url = self.upload_image_for_body(img_path)
|
||
results[img_path] = img_url
|
||
print(f"✅ 上传成功: {os.path.basename(img_path)} -> {img_url[:40]}...")
|
||
except Exception as e:
|
||
print(f"❌ 上传失败 {os.path.basename(img_path)}: {e}")
|
||
results[img_path] = None
|
||
|
||
return results
|
||
|
||
def extract_images_from_markdown(self, markdown_content: str) -> List[str]:
|
||
"""从Markdown中提取图片路径"""
|
||
# 匹配Markdown图片语法: 
|
||
pattern = r'!\[.*?\]\((.*?)\)'
|
||
matches = re.findall(pattern, markdown_content)
|
||
|
||
# 过滤出本地文件路径（排除URL）
|
||
local_images = []
|
||
for match in matches:
|
||
if not match.startswith(('http://', 'https://')):
|
||
# 检查文件是否存在
|
||
if os.path.exists(match):
|
||
local_images.append(match)
|
||
else:
|
||
# 尝试相对路径
|
||
abs_path = os.path.abspath(match)
|
||
if os.path.exists(abs_path):
|
||
local_images.append(abs_path)
|
||
|
||
return local_images
|
||
|
||
def replace_images_in_markdown(self, markdown_content: str, image_map: Dict[str, str]) -> str:
|
||
"""将Markdown中的本地图片替换为微信图片引用
|
||
image_map 的值可以是 URL（来自 upload_image_for_body）或 media_id
|
||
"""
|
||
content = markdown_content
|
||
|
||
for img_path, img_value in image_map.items():
|
||
if img_value:
|
||
# 判断是 URL 还是 media_id
|
||
if img_value.startswith('http://') or img_value.startswith('https://'):
|
||
# 直接使用真实 URL
|
||
wechat_img = f'<img src="{img_value}" alt="图片" style="width:100%;max-width:100%;">'
|
||
else:
|
||
# 兼容旧逻辑：使用 media_id 构造 URL
|
||
wechat_img = f'<img src="https://mmbiz.qpic.cn/mmbiz_jpg/{img_value}/0" alt="图片" style="width:100%;max-width:100%;">'
|
||
|
||
# 替换Markdown图片语法
|
||
pattern = rf'!\[.*?\]\({re.escape(img_path)}\)'
|
||
content = re.sub(pattern, wechat_img, content)
|
||
|
||
return content
|
||
|
||
def markdown_to_html(self, markdown_content: str) -> str:
|
||
"""将Markdown转换为HTML（简化版）"""
|
||
# 处理标题
|
||
html = markdown_content
|
||
|
||
# 标题
|
||
html = re.sub(r'^# (.*?)$', r'<h1>\1</h1>', html, flags=re.MULTILINE)
|
||
html = re.sub(r'^## (.*?)$', r'<h2>\1</h2>', html, flags=re.MULTILINE)
|
||
html = re.sub(r'^### (.*?)$', r'<h3>\1</h3>', html, flags=re.MULTILINE)
|
||
|
||
# 粗体和斜体
|
||
html = re.sub(r'\*\*(.*?)\*\*', r'<strong>\1</strong>', html)
|
||
html = re.sub(r'\*(.*?)\*', r'<em>\1</em>', html)
|
||
|
||
# 列表
|
||
html = re.sub(r'^- (.*?)$', r'<li>\1</li>', html, flags=re.MULTILINE)
|
||
|
||
# 段落（处理空行）
|
||
lines = html.split('\n')
|
||
in_paragraph = False
|
||
result_lines = []
|
||
|
||
for line in lines:
|
||
if line.strip() == '':
|
||
if in_paragraph:
|
||
result_lines.append('</p>')
|
||
in_paragraph = False
|
||
result_lines.append('')
|
||
else:
|
||
if not in_paragraph and not line.startswith(('<h', '<li', '<img', '<blockquote')):
|
||
result_lines.append('<p>')
|
||
in_paragraph = True
|
||
result_lines.append(line)
|
||
|
||
if in_paragraph:
|
||
result_lines.append('</p>')
|
||
|
||
html = '\n'.join(result_lines)
|
||
|
||
# 引用
|
||
html = re.sub(r'^> (.*?)$', r'<blockquote>\1</blockquote>', html, flags=re.MULTILINE)
|
||
|
||
# 代码块（简化处理）
|
||
html = re.sub(r'```.*?\n(.*?)\n```', r'<pre><code>\1</code></pre>', html, flags=re.DOTALL)
|
||
|
||
return html
|
||
|
||
def create_draft(self, title: str, content: str, author: str = None,
|
||
cover_image_path: str = None) -> str:
|
||
"""创建公众号草稿（支持HTML内容）"""
|
||
access_token = self.get_access_token()
|
||
url = f"https://api.weixin.qq.com/cgi-bin/draft/add?access_token={access_token}"
|
||
|
||
# 上传封面图
|
||
thumb_media_id = None
|
||
if cover_image_path and os.path.exists(cover_image_path):
|
||
thumb_media_id = self.upload_image(cover_image_path)
|
||
|
||
# 如果没有提供作者，使用默认作者
|
||
if not author:
|
||
author = self.config.get('default_author', '')
|
||
|
||
# 生成摘要（纯文本，不超过120字）
|
||
plain_text = re.sub(r'<[^>]+>', '', content) # 移除HTML标签
|
||
plain_text = re.sub(r'\s+', ' ', plain_text).strip()
|
||
|
||
if len(plain_text) > 120:
|
||
digest = plain_text[:117] + "..."
|
||
else:
|
||
digest = plain_text
|
||
|
||
# 构建文章数据
|
||
article = {
|
||
"title": title,
|
||
"author": author,
|
||
"digest": digest,
|
||
"content": content,
|
||
"show_cover_pic": 0,
|
||
"need_open_comment": int(self.config.get('need_open_comment', 1)),
|
||
"only_fans_can_comment": int(self.config.get('only_fans_can_comment', 0))
|
||
}
|
||
|
||
if thumb_media_id:
|
||
article["thumb_media_id"] = thumb_media_id
|
||
|
||
body = {"articles": [article]}
|
||
|
||
# 发送请求
|
||
headers = {"Content-Type": "application/json; charset=utf-8"}
|
||
response = requests.post(url, data=json.dumps(body, ensure_ascii=False).encode('utf-8'), headers=headers)
|
||
data = response.json()
|
||
|
||
if 'media_id' in data:
|
||
return data['media_id']
|
||
else:
|
||
raise Exception(f"创建草稿失败: {data}")
|
||
|
||
def publish_markdown_with_images(self, markdown_path: str, title: str = None,
|
||
author: str = None, cover_image_path: str = None) -> Dict:
|
||
"""发布Markdown文章（支持图文混排）"""
|
||
print("📝 开始发布Markdown文章...")
|
||
|
||
# 检查文件
|
||
if not os.path.exists(markdown_path):
|
||
raise FileNotFoundError(f"Markdown文件不存在: {markdown_path}")
|
||
|
||
# 读取Markdown内容
|
||
with open(markdown_path, 'r', encoding='utf-8') as f:
|
||
markdown_content = f.read()
|
||
|
||
print(f"📄 读取文件: {os.path.basename(markdown_path)} ({len(markdown_content)} 字符)")
|
||
|
||
# 提取图片
|
||
print("🖼️ 提取图片...")
|
||
image_paths = self.extract_images_from_markdown(markdown_content)
|
||
|
||
if image_paths:
|
||
print(f"📸 发现 {len(image_paths)} 张本地图片:")
|
||
for img in image_paths:
|
||
print(f" - {os.path.basename(img)}")
|
||
|
||
# 上传图片
|
||
print("⬆️ 上传图片到微信...")
|
||
image_map = self.upload_images(image_paths)
|
||
|
||
# 替换图片引用
|
||
print("🔄 替换图片引用...")
|
||
processed_content = self.replace_images_in_markdown(markdown_content, image_map)
|
||
else:
|
||
print("ℹ️ 未发现本地图片")
|
||
processed_content = markdown_content
|
||
|
||
# 转换为HTML
|
||
print("🔧 转换为HTML...")
|
||
html_content = self.markdown_to_html(processed_content)
|
||
|
||
# 如果没有标题，使用文件名
|
||
if not title:
|
||
title = Path(markdown_path).stem
|
||
|
||
# 如果没有封面图，使用默认
|
||
if not cover_image_path:
|
||
default_cover = "/home/ubuntu/default_cover.png"
|
||
if os.path.exists(default_cover):
|
||
cover_image_path = default_cover
|
||
print(f"🖼️ 使用默认封面图: {os.path.basename(default_cover)}")
|
||
|
||
# 创建草稿
|
||
print("📤 创建公众号草稿...")
|
||
media_id = self.create_draft(title, html_content, author, cover_image_path)
|
||
|
||
print("✅ 发布成功！")
|
||
|
||
return {
|
||
"success": True,
|
||
"media_id": media_id,
|
||
"title": title,
|
||
"author": author or self.config.get('default_author', ''),
|
||
"images_uploaded": len([v for v in image_map.values() if v]) if image_paths else 0,
|
||
"message": "草稿创建成功"
|
||
}
|
||
|
||
def publish_html(self, html_path: str, title: str = None, author: str = None,
|
||
cover_image_path: str = None) -> Dict:
|
||
"""发布HTML文章"""
|
||
print("📝 开始发布HTML文章...")
|
||
|
||
if not os.path.exists(html_path):
|
||
raise FileNotFoundError(f"HTML文件不存在: {html_path}")
|
||
|
||
# 读取HTML内容
|
||
with open(html_path, 'r', encoding='utf-8') as f:
|
||
html_content = f.read()
|
||
|
||
print(f"📄 读取文件: {os.path.basename(html_path)} ({len(html_content)} 字符)")
|
||
|
||
# 如果没有标题，使用文件名
|
||
if not title:
|
||
title = Path(html_path).stem
|
||
|
||
# 创建草稿
|
||
print("📤 创建公众号草稿...")
|
||
media_id = self.create_draft(title, html_content, author, cover_image_path)
|
||
|
||
print("✅ 发布成功！")
|
||
|
||
return {
|
||
"success": True,
|
||
"media_id": media_id,
|
||
"title": title,
|
||
"author": author or self.config.get('default_author', ''),
|
||
"message": "草稿创建成功"
|
||
}
|
||
|
||
def main():
|
||
"""命令行入口"""
|
||
import sys
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description='微信公众号发布工具（增强版）')
|
||
subparsers = parser.add_subparsers(dest='command', help='命令')
|
||
|
||
# markdown 命令
|
||
md_parser = subparsers.add_parser('markdown', help='发布Markdown文章')
|
||
md_parser.add_argument('file', help='Markdown文件路径')
|
||
md_parser.add_argument('--title', help='文章标题')
|
||
md_parser.add_argument('--author', help='作者')
|
||
md_parser.add_argument('--cover', help='封面图路径')
|
||
|
||
# html 命令
|
||
html_parser = subparsers.add_parser('html', help='发布HTML文章')
|
||
html_parser.add_argument('file', help='HTML文件路径')
|
||
html_parser.add_argument('--title', help='文章标题')
|
||
html_parser.add_argument('--author', help='作者')
|
||
html_parser.add_argument('--cover', help='封面图路径')
|
||
|
||
# test 命令
|
||
test_parser = subparsers.add_parser('test', help='测试配置')
|
||
|
||
args = parser.parse_args()
|
||
|
||
publisher = WeChatPublisher()
|
||
|
||
try:
|
||
if args.command == 'markdown':
|
||
result = publisher.publish_markdown_with_images(
|
||
args.file, args.title, args.author, args.cover
|
||
)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
elif args.command == 'html':
|
||
result = publisher.publish_html(
|
||
args.file, args.title, args.author, args.cover
|
||
)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
elif args.command == 'test':
|
||
print("🔧 测试配置...")
|
||
print(f"AppID: {publisher.config.get('app_id')}")
|
||
print(f"Author: {publisher.config.get('default_author')}")
|
||
|
||
# 测试access_token
|
||
try:
|
||
token = publisher.get_access_token()
|
||
print(f"✅ Access Token: {token[:20]}...")
|
||
print("✅ 配置正常")
|
||
except Exception as e:
|
||
print(f"❌ 配置错误: {e}")
|
||
|
||
else:
|
||
parser.print_help()
|
||
|
||
except Exception as e:
|
||
print(json.dumps({
|
||
"success": False,
|
||
"error": str(e)
|
||
}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main() |