Ubuntu
724 lines
27 KiB
Python
724 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
baoyu-post-to-wechat 编码修复版
|
||
修复中文内容在微信后台显示为Unicode转义的问题
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
import requests
|
||
import yaml
|
||
import base64
|
||
import tempfile
|
||
import mimetypes
|
||
from pathlib import Path
|
||
from typing import List, Dict, Optional, Union, Tuple
|
||
from datetime import datetime
|
||
|
||
# 尝试导入Pillow（图片处理）
|
||
try:
|
||
from PIL import Image, ImageOps
|
||
HAS_PILLOW = True
|
||
except ImportError:
|
||
HAS_PILLOW = False
|
||
print("⚠️ Pillow未安装，图片优化功能受限")
|
||
|
||
# 配置文件路径
|
||
EXTEND_PATH = "/home/ubuntu/.nanobot/workspace/skills/baoyu-post-to-wechat/EXTEND.md"
|
||
WEWRITE_CONFIG = "/home/ubuntu/wewrite/config.yaml"
|
||
|
||
class ImageOptimizer:
|
||
"""图片优化器"""
|
||
|
||
@staticmethod
|
||
def optimize_image(image_path: str, max_size: int = 1024*1024) -> str:
|
||
"""
|
||
优化图片以适应微信要求
|
||
返回优化后的图片路径
|
||
"""
|
||
if not HAS_PILLOW:
|
||
print(f"⚠️ Pillow未安装，跳过图片优化: {os.path.basename(image_path)}")
|
||
return image_path
|
||
|
||
try:
|
||
with Image.open(image_path) as img:
|
||
original_format = img.format
|
||
original_size = os.path.getsize(image_path)
|
||
|
||
print(f"🖼️ 优化图片: {os.path.basename(image_path)} "
|
||
f"({original_format}, {original_size/1024:.1f}KB)")
|
||
|
||
# 转换模式
|
||
if img.mode not in ['RGB', 'L']:
|
||
img = img.convert('RGB')
|
||
|
||
# 检查是否需要调整尺寸（微信建议宽度≤2560）
|
||
max_width = 2560
|
||
if img.width > max_width:
|
||
ratio = max_width / img.width
|
||
new_height = int(img.height * ratio)
|
||
img = img.resize((max_width, new_height), Image.Resampling.LANCZOS)
|
||
print(f" ↳ 调整尺寸: {img.width}x{img.height}")
|
||
|
||
# 检查文件大小
|
||
if original_size > max_size:
|
||
# 创建临时文件
|
||
temp_file = tempfile.NamedTemporaryFile(
|
||
suffix='.jpg', delete=False, prefix='optimized_'
|
||
)
|
||
temp_path = temp_file.name
|
||
temp_file.close()
|
||
|
||
# 渐进式压缩
|
||
quality = 85
|
||
while quality >= 50:
|
||
img.save(temp_path, 'JPEG', quality=quality, optimize=True)
|
||
optimized_size = os.path.getsize(temp_path)
|
||
|
||
if optimized_size <= max_size or quality <= 50:
|
||
print(f" ↳ 压缩质量: {quality}% → {optimized_size/1024:.1f}KB")
|
||
return temp_path
|
||
|
||
quality -= 10
|
||
|
||
return temp_path
|
||
else:
|
||
# 如果图片已经符合要求，直接返回原路径
|
||
return image_path
|
||
|
||
except Exception as e:
|
||
print(f"❌ 图片优化失败 {os.path.basename(image_path)}: {e}")
|
||
return image_path
|
||
|
||
@staticmethod
|
||
def get_image_info(image_path: str) -> Dict:
|
||
"""获取图片信息"""
|
||
info = {
|
||
'path': image_path,
|
||
'filename': os.path.basename(image_path),
|
||
'exists': os.path.exists(image_path),
|
||
'size_kb': os.path.getsize(image_path) / 1024 if os.path.exists(image_path) else 0
|
||
}
|
||
|
||
if HAS_PILLOW and os.path.exists(image_path):
|
||
try:
|
||
with Image.open(image_path) as img:
|
||
info['width'] = img.width
|
||
info['height'] = img.height
|
||
info['format'] = img.format
|
||
info['mode'] = img.mode
|
||
except:
|
||
pass
|
||
|
||
return info
|
||
|
||
@staticmethod
|
||
def format_needs_conversion(image_path: str) -> bool:
|
||
"""检查图片是否需要格式转换"""
|
||
if not HAS_PILLOW or not os.path.exists(image_path):
|
||
return False
|
||
|
||
try:
|
||
with Image.open(image_path) as img:
|
||
# 微信推荐使用JPEG格式
|
||
if img.format not in ['JPEG', 'JPG', 'PNG']:
|
||
return True
|
||
|
||
# 检查模式
|
||
if img.mode not in ['RGB', 'L']:
|
||
return True
|
||
|
||
return False
|
||
except:
|
||
return False
|
||
|
||
class WeChatPublisher:
|
||
"""微信发布器（编码修复版）"""
|
||
|
||
def __init__(self):
|
||
self.config = self.load_config()
|
||
self.image_optimizer = ImageOptimizer()
|
||
|
||
def load_config(self) -> Dict:
|
||
"""加载配置"""
|
||
config = {}
|
||
|
||
# 从WeWrite配置加载
|
||
if os.path.exists(WEWRITE_CONFIG):
|
||
try:
|
||
with open(WEWRITE_CONFIG, 'r', encoding='utf-8') as f:
|
||
wewrite_config = yaml.safe_load(f)
|
||
if wewrite_config and 'wechat' in wewrite_config:
|
||
config.update(wewrite_config['wechat'])
|
||
except Exception as e:
|
||
print(f"⚠️ 读取WeWrite配置失败: {e}")
|
||
|
||
# 从EXTEND.md加载
|
||
if os.path.exists(EXTEND_PATH):
|
||
try:
|
||
with open(EXTEND_PATH, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
# 提取配置信息
|
||
appid_match = re.search(r'appid:\s*(\w+)', content)
|
||
secret_match = re.search(r'secret:\s*(\w+)', content)
|
||
|
||
if appid_match:
|
||
config['appid'] = appid_match.group(1)
|
||
if secret_match:
|
||
config['secret'] = secret_match.group(1)
|
||
except Exception as e:
|
||
print(f"⚠️ 读取EXTEND配置失败: {e}")
|
||
|
||
# 检查必要配置
|
||
required = ['appid', 'secret']
|
||
missing = [key for key in required if key not in config]
|
||
if missing:
|
||
print(f"❌ 缺少必要配置: {missing}")
|
||
print("请确保配置文件中包含以下信息:")
|
||
print(" appid: 微信公众平台AppID")
|
||
print(" secret: 微信公众平台AppSecret")
|
||
raise ValueError(f"缺少配置: {missing}")
|
||
|
||
return config
|
||
|
||
def get_access_token(self) -> str:
|
||
"""获取微信Access Token"""
|
||
url = f"https://api.weixin.qq.com/cgi-bin/token"
|
||
params = {
|
||
'grant_type': 'client_credential',
|
||
'appid': self.config['appid'],
|
||
'secret': self.config['secret']
|
||
}
|
||
|
||
response = requests.get(url, params=params)
|
||
data = response.json()
|
||
|
||
if 'access_token' in data:
|
||
return data['access_token']
|
||
else:
|
||
error_msg = data.get('errmsg', '未知错误')
|
||
raise Exception(f"获取Access Token失败: {error_msg}")
|
||
|
||
def upload_image(self, image_path: str, access_token: str) -> Dict:
|
||
"""
|
||
上传图片到微信
|
||
使用media/uploadimg接口，返回完整的图片URL
|
||
"""
|
||
# 优化图片
|
||
optimized_path = self.image_optimizer.optimize_image(image_path)
|
||
|
||
url = f"https://api.weixin.qq.com/cgi-bin/media/uploadimg"
|
||
params = {'access_token': access_token}
|
||
|
||
with open(optimized_path, 'rb') as f:
|
||
files = {'media': (os.path.basename(image_path), f, 'image/jpeg')}
|
||
response = requests.post(url, params=params, files=files)
|
||
|
||
data = response.json()
|
||
|
||
# 清理临时文件
|
||
if optimized_path != image_path and os.path.exists(optimized_path):
|
||
try:
|
||
os.unlink(optimized_path)
|
||
except:
|
||
pass
|
||
|
||
if 'url' in data:
|
||
return {
|
||
'media_id': data.get('url', '').split('/')[-1].split('.')[0] if 'url' in data else '',
|
||
'url': data['url']
|
||
}
|
||
else:
|
||
error_msg = data.get('errmsg', '未知错误')
|
||
raise Exception(f"上传图片失败: {error_msg}")
|
||
|
||
def upload_cover_image(self, image_path: str, access_token: str) -> str:
|
||
"""
|
||
上传封面图（使用add_material接口）
|
||
"""
|
||
# 优化图片
|
||
optimized_path = self.image_optimizer.optimize_image(image_path)
|
||
|
||
url = f"https://api.weixin.qq.com/cgi-bin/material/add_material"
|
||
params = {
|
||
'access_token': access_token,
|
||
'type': 'image'
|
||
}
|
||
|
||
with open(optimized_path, 'rb') as f:
|
||
files = {'media': (os.path.basename(image_path), f, 'image/jpeg')}
|
||
response = requests.post(url, params=params, files=files)
|
||
|
||
data = response.json()
|
||
|
||
# 清理临时文件
|
||
if optimized_path != image_path and os.path.exists(optimized_path):
|
||
try:
|
||
os.unlink(optimized_path)
|
||
except:
|
||
pass
|
||
|
||
if 'media_id' in data:
|
||
return data['media_id']
|
||
else:
|
||
error_msg = data.get('errmsg', '未知错误')
|
||
raise Exception(f"上传封面图失败: {error_msg}")
|
||
|
||
def process_images(self, markdown_content: str) -> Tuple[str, Dict]:
|
||
"""
|
||
处理图片：上传并替换为正确的微信图片引用
|
||
返回处理后的内容和图片映射
|
||
"""
|
||
# 提取图片
|
||
image_paths = self.extract_images_from_markdown(markdown_content)
|
||
|
||
if not image_paths:
|
||
print("ℹ️ 未发现本地图片")
|
||
return markdown_content, {}
|
||
|
||
print(f"📸 发现 {len(image_paths)} 张图片:")
|
||
image_map = {}
|
||
|
||
# 获取access token
|
||
try:
|
||
access_token = self.get_access_token()
|
||
except Exception as e:
|
||
print(f"❌ 获取Access Token失败: {e}")
|
||
return markdown_content, {}
|
||
|
||
# 显示图片信息
|
||
for i, img_path in enumerate(image_paths, 1):
|
||
img_info = self.image_optimizer.get_image_info(img_path)
|
||
size_info = f"{img_info.get('size_kb', 0):.1f}KB"
|
||
if 'width' in img_info:
|
||
size_info = f"{img_info['width']}x{img_info['height']}, {size_info}"
|
||
|
||
format_info = img_info.get('format', '未知')
|
||
needs_convert = self.image_optimizer.format_needs_conversion(img_path)
|
||
|
||
status = "🔄 需要转换" if needs_convert else "✅ 格式正常"
|
||
print(f" {i}. {status} {img_info['filename']} ({format_info}, {size_info})")
|
||
|
||
print("⬆️ 上传图片到微信...")
|
||
for img_path in image_paths:
|
||
if not os.path.exists(img_path):
|
||
image_map[img_path] = None
|
||
continue
|
||
|
||
try:
|
||
result = self.upload_image(img_path, access_token)
|
||
image_map[img_path] = result
|
||
print(f" ✅ 上传成功: {os.path.basename(img_path)} -> {result['url'][:50]}...")
|
||
except Exception as e:
|
||
print(f" ❌ 上传失败 {os.path.basename(img_path)}: {e}")
|
||
image_map[img_path] = None
|
||
|
||
# 替换图片引用
|
||
processed_content = self.replace_images_in_markdown(markdown_content, image_map)
|
||
|
||
return processed_content, image_map
|
||
|
||
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, Dict]) -> str:
|
||
"""
|
||
将Markdown中的本地图片替换为正确的微信图片引用
|
||
使用完整的URL，而不是media_id
|
||
"""
|
||
content = markdown_content
|
||
|
||
for img_path, result in image_map.items():
|
||
if result and 'url' in result:
|
||
# 构建正确的微信图片引用格式
|
||
wechat_img = f'<img src="{result["url"]}" style="display: block; width: 100%; margin: 1.5em auto;">'
|
||
|
||
# 替换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'<h4>\1</h4>', 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'<del>\1</del>', html)
|
||
|
||
# 内联代码
|
||
html = re.sub(r'`([^`]+)`', r'<code>\1</code>', html)
|
||
|
||
# 列表
|
||
lines = html.split('\n')
|
||
in_list = False
|
||
result_lines = []
|
||
|
||
for line in lines:
|
||
if re.match(r'^[*-] ', line):
|
||
if not in_list:
|
||
result_lines.append('<ul>')
|
||
in_list = True
|
||
content = line[2:].strip()
|
||
result_lines.append(f' <li>{content}</li>')
|
||
elif re.match(r'^\d+\. ', line):
|
||
if not in_list:
|
||
result_lines.append('<ol>')
|
||
in_list = True
|
||
content = line[line.find('.')+1:].strip()
|
||
result_lines.append(f' <li>{content}</li>')
|
||
else:
|
||
if in_list:
|
||
result_lines.append('</ul>' if line.startswith('* ') else '</ol>')
|
||
in_list = False
|
||
result_lines.append(line)
|
||
|
||
if in_list:
|
||
result_lines.append('</ul>')
|
||
|
||
html = '\n'.join(result_lines)
|
||
|
||
# 段落
|
||
html = re.sub(r'\n\n+', r'</p>\n<p>', html)
|
||
html = '<p>' + html + '</p>'
|
||
|
||
# 链接
|
||
html = re.sub(r'\[(.*?)\]\((.*?)\)', r'<a href="\2">\1</a>', html)
|
||
|
||
# 代码块（简化处理）
|
||
html = re.sub(r'```.*?\n(.*?)\n```', r'<pre><code>\1</code></pre>', html, flags=re.DOTALL)
|
||
|
||
# 表格（简化处理）
|
||
html = re.sub(r'\|(.*?)\|', r'<td>\1</td>', html)
|
||
html = re.sub(r'<td>(.*?)</td>\s*<td>(.*?)</td>', r'<tr><td>\1</td><td>\2</td></tr>', html)
|
||
|
||
return html
|
||
|
||
def parse_frontmatter(self, content: str) -> Tuple[Dict, str]:
|
||
"""解析Frontmatter"""
|
||
frontmatter = {}
|
||
body = content
|
||
|
||
# 检查是否有Frontmatter
|
||
if content.startswith('---'):
|
||
parts = content.split('---', 2)
|
||
if len(parts) >= 3:
|
||
frontmatter_str = parts[1].strip()
|
||
body = parts[2].strip()
|
||
|
||
try:
|
||
frontmatter = yaml.safe_load(frontmatter_str) or {}
|
||
except:
|
||
print("⚠️ Frontmatter解析失败，跳过")
|
||
|
||
return frontmatter, body
|
||
|
||
def generate_summary(self, content: str, max_length: int = 120) -> str:
|
||
"""生成摘要"""
|
||
# 提取第一段有意义的文本
|
||
lines = content.split('\n')
|
||
for line in lines:
|
||
line = line.strip()
|
||
if line and not line.startswith(('#', '!', '[', '|', '-', '*', '`')):
|
||
# 清理Markdown标记
|
||
clean_line = re.sub(r'[#*`\[\]]', '', line)
|
||
if len(clean_line) > 20:
|
||
summary = clean_line[:max_length]
|
||
if len(clean_line) > max_length:
|
||
summary += '...'
|
||
return summary
|
||
|
||
# 如果没有找到合适的文本，使用默认摘要
|
||
return "本文探讨了当前热点话题，提供深度分析和见解。"
|
||
|
||
def publish_markdown(self, markdown_file: str, title: str = None, author: str = None) -> Dict:
|
||
"""发布Markdown文章（编码修复版）"""
|
||
print("=" * 60)
|
||
print("📝 baoyu-post-to-wechat 编码修复版 v2.2")
|
||
print("=" * 60)
|
||
|
||
# 读取文件
|
||
if not os.path.exists(markdown_file):
|
||
raise FileNotFoundError(f"文件不存在: {markdown_file}")
|
||
|
||
with open(markdown_file, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
print(f"📄 读取文件: {os.path.basename(markdown_file)} ({len(content)} 字符)")
|
||
|
||
# 解析Frontmatter
|
||
print("🔍 解析Frontmatter...")
|
||
frontmatter, body = self.parse_frontmatter(content)
|
||
|
||
if frontmatter:
|
||
print(" ✅ 发现Frontmatter:")
|
||
for key, value in frontmatter.items():
|
||
print(f" {key}: {value}")
|
||
|
||
# 处理图片
|
||
processed_body, image_map = self.process_images(body)
|
||
|
||
# 合并Frontmatter和正文
|
||
if frontmatter:
|
||
# 如果有Frontmatter，保持原格式
|
||
final_content = f"---\n{yaml.dump(frontmatter, allow_unicode=True, default_flow_style=False)}---\n\n{processed_body}"
|
||
else:
|
||
final_content = processed_body
|
||
|
||
# 转换为HTML
|
||
print("🔧 转换为HTML...")
|
||
html_content = self.markdown_to_html(processed_body)
|
||
|
||
# 生成摘要
|
||
print("📋 生成摘要...")
|
||
summary = frontmatter.get('summary') or self.generate_summary(body)
|
||
print(f" 摘要: {summary[:80]}...")
|
||
|
||
# 获取标题和作者
|
||
final_title = title or frontmatter.get('title') or os.path.basename(markdown_file).replace('.md', '')
|
||
final_author = author or frontmatter.get('author') or "蔡贤明"
|
||
|
||
# 获取Access Token
|
||
print("🔑 获取Access Token...")
|
||
access_token = self.get_access_token()
|
||
|
||
# 上传封面图
|
||
print("🖼️ 上传封面图...")
|
||
cover_path = "/home/ubuntu/default_cover.png"
|
||
if not os.path.exists(cover_path):
|
||
# 创建默认封面图
|
||
cover_path = self.create_default_cover(final_title)
|
||
|
||
thumb_media_id = self.upload_cover_image(cover_path, access_token)
|
||
print(f" ✅ 封面图上传成功: {thumb_media_id[:20]}...")
|
||
|
||
# 创建草稿 - 修复编码问题
|
||
print("📤 创建公众号草稿（编码修复）...")
|
||
|
||
# 确保数据是UTF-8编码的字符串
|
||
draft_data = {
|
||
'articles': [{
|
||
'title': final_title,
|
||
'author': final_author,
|
||
'digest': summary,
|
||
'content': html_content,
|
||
'thumb_media_id': thumb_media_id,
|
||
'need_open_comment': 1,
|
||
'only_fans_can_comment': 0
|
||
}]
|
||
}
|
||
|
||
# 调试：查看数据格式
|
||
print("🔍 调试数据格式...")
|
||
print(f" 标题长度: {len(final_title)} 字符")
|
||
print(f" 作者长度: {len(final_author)} 字符")
|
||
print(f" 摘要长度: {len(summary)} 字符")
|
||
print(f" 内容长度: {len(html_content)} 字符")
|
||
|
||
# 检查是否有Unicode转义字符
|
||
if '\\u' in html_content:
|
||
print("⚠️ 警告：内容中包含Unicode转义字符")
|
||
# 尝试解码Unicode转义
|
||
try:
|
||
html_content = html_content.encode('utf-8').decode('unicode_escape')
|
||
print(" ✅ 已解码Unicode转义字符")
|
||
except:
|
||
print(" ❌ Unicode转义解码失败")
|
||
|
||
url = f"https://api.weixin.qq.com/cgi-bin/draft/add"
|
||
params = {'access_token': access_token}
|
||
|
||
# 使用json.dumps确保正确的编码
|
||
json_data = json.dumps(draft_data, ensure_ascii=False)
|
||
|
||
headers = {
|
||
'Content-Type': 'application/json; charset=utf-8'
|
||
}
|
||
|
||
response = requests.post(url, params=params, data=json_data.encode('utf-8'), headers=headers)
|
||
result = response.json()
|
||
|
||
print("=" * 60)
|
||
|
||
if 'media_id' in result:
|
||
print("✅ 发布成功！")
|
||
print("=" * 60)
|
||
|
||
# 显示预览信息
|
||
print("📱 预览信息:")
|
||
print(f" 标题: {final_title}")
|
||
print(f" 作者: {final_author}")
|
||
print(f" 摘要: {summary}")
|
||
print(f" Media ID: {result['media_id']}")
|
||
print(f" 图片上传: {len([v for v in image_map.values() if v])} 张")
|
||
print("=" * 60)
|
||
|
||
return {
|
||
'success': True,
|
||
'media_id': result['media_id'],
|
||
'title': final_title,
|
||
'author': final_author,
|
||
'digest': summary,
|
||
'images_uploaded': len([v for v in image_map.values() if v]),
|
||
'has_frontmatter': bool(frontmatter),
|
||
'message': '草稿创建成功'
|
||
}
|
||
else:
|
||
error_msg = result.get('errmsg', '未知错误')
|
||
print(f"❌ 发布失败: {error_msg}")
|
||
print("=" * 60)
|
||
|
||
return {
|
||
'success': False,
|
||
'error': error_msg,
|
||
'message': '草稿创建失败'
|
||
}
|
||
|
||
def create_default_cover(self, title: str) -> str:
|
||
"""创建默认封面图"""
|
||
if not HAS_PILLOW:
|
||
# 如果没有Pillow，创建一个简单的文本文件作为占位符
|
||
temp_file = tempfile.NamedTemporaryFile(suffix='.txt', delete=False, prefix='cover_')
|
||
temp_path = temp_file.name
|
||
temp_file.close()
|
||
|
||
with open(temp_path, 'w', encoding='utf-8') as f:
|
||
f.write(f"封面图: {title}")
|
||
|
||
return temp_path
|
||
|
||
# 使用Pillow创建简单的封面图
|
||
try:
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
# 创建图片
|
||
width, height = 800, 450
|
||
image = Image.new('RGB', (width, height), color=(41, 128, 185))
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
# 添加标题
|
||
try:
|
||
# 尝试使用系统字体
|
||
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 40)
|
||
except:
|
||
font = ImageFont.load_default()
|
||
|
||
# 计算文本位置
|
||
text = title[:30] + "..." if len(title) > 30 else title
|
||
bbox = draw.textbbox((0, 0), text, font=font)
|
||
text_width = bbox[2] - bbox[0]
|
||
text_height = bbox[3] - bbox[1]
|
||
|
||
x = (width - text_width) // 2
|
||
y = (height - text_height) // 2
|
||
|
||
# 绘制文本
|
||
draw.text((x, y), text, font=font, fill=(255, 255, 255))
|
||
|
||
# 保存图片
|
||
temp_file = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False, prefix='cover_')
|
||
temp_path = temp_file.name
|
||
temp_file.close()
|
||
|
||
image.save(temp_path, 'JPEG', quality=90)
|
||
|
||
return temp_path
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ 创建封面图失败: {e}")
|
||
|
||
# 创建文本文件作为后备
|
||
temp_file = tempfile.NamedTemporaryFile(suffix='.txt', delete=False, prefix='cover_')
|
||
temp_path = temp_file.name
|
||
temp_file.close()
|
||
|
||
with open(temp_path, 'w', encoding='utf-8') as f:
|
||
f.write(f"封面图: {title}")
|
||
|
||
return temp_path
|
||
|
||
def test_connection(self):
|
||
"""测试连接"""
|
||
print("🔧 测试微信API连接...")
|
||
|
||
try:
|
||
access_token = self.get_access_token()
|
||
print(f"✅ Access Token获取成功: {access_token[:20]}...")
|
||
|
||
# 测试图片上传
|
||
test_image = self.create_default_cover("测试封面")
|
||
if os.path.exists(test_image):
|
||
try:
|
||
result = self.upload_image(test_image, access_token)
|
||
print(f"✅ 图片上传测试成功: {result['url'][:50]}...")
|
||
except Exception as e:
|
||
print(f"⚠️ 图片上传测试失败: {e}")
|
||
|
||
# 清理测试文件
|
||
if test_image.startswith('/tmp/'):
|
||
try:
|
||
os.unlink(test_image)
|
||
except:
|
||
pass
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 连接测试失败: {e}")
|
||
return False
|
||
|
||
def main():
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description='baoyu-post-to-wechat 编码修复版')
|
||
parser.add_argument('action', choices=['markdown', 'test'], help='执行动作')
|
||
parser.add_argument('file', nargs='?', help='Markdown文件路径')
|
||
parser.add_argument('--title', help='文章标题')
|
||
parser.add_argument('--author', help='文章作者')
|
||
|
||
args = parser.parse_args()
|
||
|
||
publisher = WeChatPublisher()
|
||
|
||
if args.action == 'test':
|
||
publisher.test_connection()
|
||
elif args.action == 'markdown':
|
||
if not args.file:
|
||
print("❌ 请指定Markdown文件路径")
|
||
return
|
||
|
||
result = publisher.publish_markdown(args.file, args.title, args.author)
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
|
||
if __name__ == '__main__':
|
||
main() |