137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
导出审核通过的文章内容和标签到CSV文件
|
|||
|
|
此脚本将从ai_articles表中导出status为approved的文章内容和标签
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import csv
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
from datetime import datetime
|
|||
|
|
from log_config import setup_logger
|
|||
|
|
|
|||
|
|
|
|||
|
|
def export_approved_articles_to_csv(output_file='approved_articles_export.csv'):
|
|||
|
|
"""
|
|||
|
|
导出审核通过的文章内容和标签到CSV文件
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
output_file: 输出的CSV文件名
|
|||
|
|
"""
|
|||
|
|
# 设置日志记录器
|
|||
|
|
logger = setup_logger('article_export', 'logs/article_export.log', 'logs/article_export_error.log')
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# 从数据库获取真实数据
|
|||
|
|
from database_config import db_manager
|
|||
|
|
|
|||
|
|
# 查询审核通过的文章，包含内容和标签
|
|||
|
|
sql = """
|
|||
|
|
SELECT id, title, content, coze_tag, created_at, updated_at
|
|||
|
|
FROM ai_articles
|
|||
|
|
WHERE status = 'approved'
|
|||
|
|
ORDER BY id
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
logger.info("开始查询审核通过的文章数据...")
|
|||
|
|
results = db_manager.execute_query(sql)
|
|||
|
|
|
|||
|
|
if not results:
|
|||
|
|
logger.warning("没有找到状态为approved的文章")
|
|||
|
|
print("没有找到状态为approved的文章")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
logger.info(f"查询到 {len(results)} 条审核通过的文章")
|
|||
|
|
print(f"查询到 {len(results)} 条审核通过的文章")
|
|||
|
|
|
|||
|
|
# 准备输出目录
|
|||
|
|
output_dir = os.path.dirname(output_file)
|
|||
|
|
if output_dir and not os.path.exists(output_dir):
|
|||
|
|
os.makedirs(output_dir)
|
|||
|
|
|
|||
|
|
# 写入CSV文件
|
|||
|
|
with open(output_file, 'w', newline='', encoding='utf-8-sig') as csvfile:
|
|||
|
|
fieldnames = ['ID', '标题', '内容', '标签', '创建时间', '更新时间']
|
|||
|
|
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
|||
|
|
|
|||
|
|
# 写入表头
|
|||
|
|
writer.writeheader()
|
|||
|
|
|
|||
|
|
# 写入数据
|
|||
|
|
for row in results:
|
|||
|
|
id_val, title, content, coze_tag, created_at, updated_at = row
|
|||
|
|
|
|||
|
|
# 尝试解析标签，如果是JSON格式则转换为字符串
|
|||
|
|
parsed_tags = coze_tag
|
|||
|
|
if coze_tag:
|
|||
|
|
try:
|
|||
|
|
# 尝试解析JSON格式的标签
|
|||
|
|
tags_data = json.loads(coze_tag)
|
|||
|
|
if isinstance(tags_data, list):
|
|||
|
|
parsed_tags = ', '.join(tags_data)
|
|||
|
|
elif isinstance(tags_data, dict):
|
|||
|
|
# 如果是字典格式，提取值
|
|||
|
|
parsed_tags = ', '.join(str(v) for v in tags_data.values())
|
|||
|
|
except json.JSONDecodeError:
|
|||
|
|
# 如果不是JSON格式，保持原样
|
|||
|
|
parsed_tags = coze_tag
|
|||
|
|
|
|||
|
|
writer.writerow({
|
|||
|
|
'ID': id_val,
|
|||
|
|
'标题': title,
|
|||
|
|
'内容': content,
|
|||
|
|
'标签': parsed_tags or '',
|
|||
|
|
'创建时间': created_at.strftime('%Y-%m-%d %H:%M:%S') if created_at else '',
|
|||
|
|
'更新时间': updated_at.strftime('%Y-%m-%d %H:%M:%S') if updated_at else ''
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
logger.info(f"成功导出 {len(results)} 条文章到 {output_file}")
|
|||
|
|
print(f"成功导出 {len(results)} 条文章到 {output_file}")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"导出文章数据时发生错误: {e}", exc_info=True)
|
|||
|
|
print(f"导出文章数据时发生错误: {e}")
|
|||
|
|
raise
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_db_connection():
|
|||
|
|
"""
|
|||
|
|
测试数据库连接
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
from database_config import db_manager
|
|||
|
|
# 尝试执行一个简单的查询来测试连接
|
|||
|
|
test_sql = "SELECT 1 as test"
|
|||
|
|
result = db_manager.execute_query(test_sql)
|
|||
|
|
print("数据库连接测试成功:", result)
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"数据库连接测试失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# 创建logs目录
|
|||
|
|
if not os.path.exists('logs'):
|
|||
|
|
os.makedirs('logs')
|
|||
|
|
|
|||
|
|
# 检查命令行参数
|
|||
|
|
import sys
|
|||
|
|
if len(sys.argv) > 1:
|
|||
|
|
output_filename = sys.argv[1]
|
|||
|
|
else:
|
|||
|
|
output_filename = 'approved_articles_export.csv'
|
|||
|
|
|
|||
|
|
# 测试数据库连接
|
|||
|
|
print("正在测试数据库连接...")
|
|||
|
|
if not test_db_connection():
|
|||
|
|
print("数据库连接失败，请检查数据库配置。")
|
|||
|
|
print("请确认以下信息：")
|
|||
|
|
print("- 数据库服务器是否正常运行")
|
|||
|
|
print("- 数据库地址、用户名、密码是否正确")
|
|||
|
|
print("- 网络连接是否正常")
|
|||
|
|
print("- 用户是否有查询ai_articles表的权限")
|
|||
|
|
exit(1)
|
|||
|
|
|
|||
|
|
export_approved_articles_to_csv(output_filename)
|