1fb93e34c6
- 飞书消息接收与处理（文字、图片、Word 文档） - WordPress REST API 文章发布 - 图片自动上传到媒体库 - Word 文档解析与发布 - HTML 格式化与分类自动匹配 - Python CLI 工具（避免 shell 引号冲突） - Webhook 服务器（8080 端口） - 完整日志系统
174 lines
5.3 KiB
Python
174 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
WordPress 发布系统 - 日志模块
|
|
提供发布日志和调试日志功能
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
from datetime import datetime
|
|
|
|
# 基础目录
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
LOG_DIR = os.path.join(BASE_DIR, 'logs')
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
|
|
# 日志格式
|
|
LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
DEBUG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
|
|
PUBLISH_FORMAT = '%(asctime)s | %(message)s'
|
|
|
|
# 日志文件大小限制 (10MB)
|
|
MAX_BYTES = 10 * 1024 * 1024
|
|
BACKUP_COUNT = 5
|
|
|
|
|
|
class PublishLogger:
|
|
"""发布日志记录器"""
|
|
|
|
def __init__(self, log_file='publish.log'):
|
|
self.logger = logging.getLogger('wp_publish')
|
|
self.logger.setLevel(logging.INFO)
|
|
|
|
# 避免重复添加处理器
|
|
if self.logger.handlers:
|
|
return
|
|
|
|
# 发布日志文件处理器
|
|
log_path = os.path.join(LOG_DIR, log_file)
|
|
file_handler = RotatingFileHandler(
|
|
log_path,
|
|
maxBytes=MAX_BYTES,
|
|
backupCount=BACKUP_COUNT,
|
|
encoding='utf-8'
|
|
)
|
|
file_handler.setLevel(logging.INFO)
|
|
file_handler.setFormatter(logging.Formatter(PUBLISH_FORMAT))
|
|
self.logger.addHandler(file_handler)
|
|
|
|
# 控制台输出
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setLevel(logging.INFO)
|
|
console_handler.setFormatter(logging.Formatter(PUBLISH_FORMAT))
|
|
self.logger.addHandler(console_handler)
|
|
|
|
def info(self, message):
|
|
self.logger.info(message)
|
|
|
|
def success(self, message):
|
|
self.logger.info(f"✅ SUCCESS: {message}")
|
|
|
|
def warning(self, message):
|
|
self.logger.warning(f"⚠️ WARNING: {message}")
|
|
|
|
def error(self, message):
|
|
self.logger.error(f"❌ ERROR: {message}")
|
|
|
|
def start_publish(self, source_type, filename=None):
|
|
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
self.logger.info("=" * 60)
|
|
self.logger.info(f"🚀 开始发布 - {timestamp}")
|
|
self.logger.info(f"📋 发布类型：{source_type}")
|
|
if filename:
|
|
self.logger.info(f"📁 文件名：{filename}")
|
|
self.logger.info("=" * 60)
|
|
|
|
def end_publish(self, success, post_id=None, post_url=None, error_msg=None):
|
|
if success:
|
|
self.success(f"发布成功！文章 ID: {post_id}")
|
|
if post_url:
|
|
self.info(f"🔗 文章链接：{post_url}")
|
|
else:
|
|
self.error(f"发布失败：{error_msg}")
|
|
self.info("=" * 60)
|
|
|
|
|
|
class DebugLogger:
|
|
"""调试日志记录器"""
|
|
|
|
def __init__(self, log_file='debug.log'):
|
|
self.logger = logging.getLogger('wp_debug')
|
|
self.logger.setLevel(logging.DEBUG)
|
|
|
|
# 避免重复添加处理器
|
|
if self.logger.handlers:
|
|
return
|
|
|
|
# 调试日志文件处理器
|
|
log_path = os.path.join(LOG_DIR, log_file)
|
|
file_handler = RotatingFileHandler(
|
|
log_path,
|
|
maxBytes=MAX_BYTES,
|
|
backupCount=BACKUP_COUNT,
|
|
encoding='utf-8'
|
|
)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
file_handler.setFormatter(logging.Formatter(DEBUG_FORMAT))
|
|
self.logger.addHandler(file_handler)
|
|
|
|
# 错误日志单独记录
|
|
error_log_path = os.path.join(LOG_DIR, 'error.log')
|
|
error_handler = RotatingFileHandler(
|
|
error_log_path,
|
|
maxBytes=MAX_BYTES,
|
|
backupCount=BACKUP_COUNT,
|
|
encoding='utf-8'
|
|
)
|
|
error_handler.setLevel(logging.ERROR)
|
|
error_handler.setFormatter(logging.Formatter(DEBUG_FORMAT))
|
|
self.logger.addHandler(error_handler)
|
|
|
|
def debug(self, message):
|
|
self.logger.debug(message)
|
|
|
|
def info(self, message):
|
|
self.logger.info(message)
|
|
|
|
def warning(self, message):
|
|
self.logger.warning(message)
|
|
|
|
def error(self, message, exc_info=None):
|
|
self.logger.error(message, exc_info=exc_info)
|
|
|
|
def log_step(self, step_name, details=None):
|
|
"""记录步骤"""
|
|
self.logger.info(f"📌 步骤：{step_name}")
|
|
if details:
|
|
self.logger.info(f" 详情：{details}")
|
|
|
|
def log_result(self, result_type, result_data):
|
|
"""记录结果"""
|
|
self.logger.info(f"📊 结果：{result_type}")
|
|
self.logger.debug(f" 数据：{result_data}")
|
|
|
|
|
|
# 创建全局日志实例
|
|
publish_logger = PublishLogger()
|
|
debug_logger = DebugLogger()
|
|
|
|
|
|
def get_publish_logger():
|
|
"""获取发布日志实例"""
|
|
return publish_logger
|
|
|
|
|
|
def get_debug_logger():
|
|
"""获取调试日志实例"""
|
|
return debug_logger
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# 测试日志
|
|
pl = get_publish_logger()
|
|
dl = get_debug_logger()
|
|
|
|
pl.start_publish('测试', 'test.docx')
|
|
pl.info('正在解析文档...')
|
|
pl.success('文档解析完成')
|
|
dl.log_step('解析测试', '提取标题和正文')
|
|
dl.debug('调试信息：文档结构正常')
|
|
pl.end_publish(True, post_id=1, post_url='https://www.nanlou.net/test')
|