378 lines
15 KiB
Python
378 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
文章自动分配作者脚本
|
||
监控数据库中status为pending_review且author_id=0的文章，自动分配作者
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import json
|
||
import logging
|
||
import requests
|
||
from typing import Dict, List
|
||
import traceback
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from database_config import get_db_manager
|
||
from log_config import setup_logger
|
||
|
||
# 配置日志记录器，支持按日期切割和控制台输出
|
||
logger = setup_logger(
|
||
name='assign_authors_to_articles',
|
||
log_file='logs/assign_authors_to_articles.log',
|
||
error_log_file='logs/assign_authors_to_articles_error.log',
|
||
level=logging.INFO,
|
||
console_output=True
|
||
)
|
||
|
||
# 配置常量
|
||
BASE_URL = "http://127.0.0.1:8216"
|
||
SLEEP_INTERVAL = 5 # 监控间隔（秒），可配置
|
||
|
||
class AssignAuthorsToArticles:
|
||
def __init__(self):
|
||
# API配置
|
||
self.base_url = BASE_URL
|
||
|
||
# 认证信息
|
||
self.jwt_token = None
|
||
|
||
# 使用统一的数据库管理器
|
||
self.db_manager = get_db_manager()
|
||
|
||
# 登录配置 - 使用雇员账号
|
||
self.login_credentials = {
|
||
'username': '13621242430',
|
||
'password': 'admin123'
|
||
}
|
||
|
||
# 禁用代理
|
||
self.proxies = {
|
||
'http': None,
|
||
'https': None
|
||
}
|
||
|
||
logger.info("AssignAuthorsToArticles 初始化完成")
|
||
|
||
def log_to_database(self, level: str, message: str, details: str = None):
|
||
"""记录日志到数据库ai_logs表"""
|
||
try:
|
||
with self.db_manager.get_cursor() as cursor:
|
||
# 映射日志级别到数据库状态
|
||
status_map = {
|
||
'INFO': 'success',
|
||
'WARNING': 'warning',
|
||
'ERROR': 'error'
|
||
}
|
||
status = status_map.get(level, 'success')
|
||
|
||
sql = """
|
||
INSERT INTO ai_logs (user_id, action, description, status, error_message, created_at)
|
||
VALUES (%s, %s, %s, %s, %s, NOW())
|
||
"""
|
||
cursor.execute(sql, (None, 'assign_authors', message, status, details))
|
||
logger.info(f"日志已记录到数据库: {level} - {message}")
|
||
except Exception as e:
|
||
logger.error(f"记录日志到数据库失败: {e}")
|
||
|
||
def login_and_get_jwt_token(self) -> bool:
|
||
"""登录获取JWT token"""
|
||
try:
|
||
login_url = f"{self.base_url}/api/auth/login"
|
||
login_data = {
|
||
"username": self.login_credentials['username'],
|
||
"password": self.login_credentials['password']
|
||
}
|
||
|
||
logger.info(f"尝试登录: {login_data['username']}")
|
||
self.log_to_database('INFO', f"尝试登录用户: {login_data['username']}")
|
||
|
||
response = requests.post(
|
||
login_url,
|
||
json=login_data,
|
||
headers={'Content-Type': 'application/json'},
|
||
proxies=self.proxies
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get('code') == 200:
|
||
self.jwt_token = result['data']['token']
|
||
logger.info("JWT token获取成功")
|
||
self.log_to_database('INFO', "JWT token获取成功")
|
||
return True
|
||
else:
|
||
error_msg = f"登录失败: {result.get('message', '未知错误')}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, json.dumps(result))
|
||
return False
|
||
else:
|
||
error_msg = f"登录请求失败: {response.status_code}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, response.text)
|
||
return False
|
||
|
||
except Exception as e:
|
||
error_msg = f"登录异常: {e}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, traceback.format_exc())
|
||
return False
|
||
|
||
def get_pending_articles(self) -> List[Dict]:
|
||
"""获取待分配作者的文章：enterprise_id=1, status=pending_review, author_id=0"""
|
||
try:
|
||
with self.db_manager.get_cursor() as cursor:
|
||
sql = """
|
||
SELECT id, title, batch_id, enterprise_id, product_id, status,
|
||
author_id, created_at, updated_at
|
||
FROM ai_articles
|
||
WHERE enterprise_id = 1
|
||
AND status = 'pending_review'
|
||
AND author_id = 0
|
||
ORDER BY created_at ASC
|
||
LIMIT 1000
|
||
"""
|
||
cursor.execute(sql)
|
||
results = cursor.fetchall()
|
||
|
||
if results:
|
||
logger.info(f"查询到 {len(results)} 篇待分配作者的文章")
|
||
for result in results:
|
||
logger.info(f"待分配文章 - ID: {result['id']}, 标题: {result['title']}, 状态: {result['status']}")
|
||
else:
|
||
logger.info("未查询到待分配作者的文章")
|
||
|
||
return results
|
||
except Exception as e:
|
||
error_msg = f"查询待分配文章异常: {e}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, traceback.format_exc())
|
||
return []
|
||
|
||
def get_active_authors(self) -> List[Dict]:
|
||
"""获取活跃作者：enterprise_id=1, status=active, phone!=空, created_user_id>0"""
|
||
try:
|
||
with self.db_manager.get_cursor() as cursor:
|
||
sql = """
|
||
SELECT id, author_name, phone, app_id, app_token,
|
||
department_id, department_name, status, created_at
|
||
FROM ai_authors
|
||
WHERE enterprise_id = 1
|
||
AND status = 'active'
|
||
AND phone IS NOT NULL AND phone != ''
|
||
AND created_user_id > 0
|
||
ORDER BY created_at ASC
|
||
LIMIT 1000
|
||
"""
|
||
cursor.execute(sql)
|
||
results = cursor.fetchall()
|
||
|
||
if results:
|
||
logger.info(f"查询到 {len(results)} 个活跃作者")
|
||
for result in results:
|
||
logger.info(f"活跃作者 - ID: {result['id']}, 姓名: {result['author_name']}, 手机: {result['phone']}")
|
||
else:
|
||
logger.info("未查询到活跃作者")
|
||
|
||
return results
|
||
except Exception as e:
|
||
error_msg = f"查询活跃作者异常: {e}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, traceback.format_exc())
|
||
return []
|
||
|
||
def assign_authors_to_articles_batch(self, article_ids: List[int], author_ids: List[int]) -> bool:
|
||
"""调用批量分配接口：/api/articles/batch-published-account-cycle"""
|
||
try:
|
||
if not article_ids or not author_ids:
|
||
logger.warning("文章ID或作者ID列表为空，跳过分配")
|
||
return False
|
||
|
||
# 确保有JWT token
|
||
if not self.jwt_token:
|
||
logger.warning("JWT token缺失，尝试重新登录")
|
||
if not self.login_and_get_jwt_token():
|
||
logger.error("重新登录失败")
|
||
return False
|
||
|
||
assign_url = f"{self.base_url}/api/articles/batch-published-account-cycle"
|
||
headers = {
|
||
'Authorization': f'Bearer {self.jwt_token}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
data = {
|
||
'article_ids': article_ids,
|
||
'author_ids': author_ids
|
||
}
|
||
|
||
logger.info(f"开始批量分配作者 - 文章数: {len(article_ids)}, 作者数: {len(author_ids)}")
|
||
logger.info(f"文章ID: {article_ids}")
|
||
logger.info(f"作者ID: {author_ids}")
|
||
self.log_to_database('INFO', f"开始批量分配作者",
|
||
f"文章数: {len(article_ids)}, 作者数: {len(author_ids)}")
|
||
|
||
response = requests.post(
|
||
assign_url,
|
||
json=data,
|
||
headers=headers,
|
||
timeout=60,
|
||
proxies=self.proxies
|
||
)
|
||
|
||
logger.info(f"批量分配响应状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
logger.info(f"批量分配响应: {result}")
|
||
|
||
if result.get('code') == 200:
|
||
data = result.get('data', {})
|
||
success_msg = f"批量分配成功 - 成功: {data.get('published_count')}, 失败: {data.get('failed_count')}"
|
||
logger.info(success_msg)
|
||
self.log_to_database('INFO', success_msg, json.dumps(result, ensure_ascii=False))
|
||
return True
|
||
else:
|
||
error_msg = f"批量分配失败: {result.get('message', '未知错误')}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, json.dumps(result, ensure_ascii=False))
|
||
return False
|
||
elif response.status_code == 401:
|
||
# Token过期，重新登录后重试
|
||
logger.warning("JWT token过期(401)，尝试重新登录")
|
||
if self.login_and_get_jwt_token():
|
||
logger.info("重新登录成功，重试批量分配")
|
||
headers['Authorization'] = f'Bearer {self.jwt_token}'
|
||
|
||
retry_response = requests.post(
|
||
assign_url,
|
||
json=data,
|
||
headers=headers,
|
||
timeout=60,
|
||
proxies=self.proxies
|
||
)
|
||
|
||
if retry_response.status_code == 200:
|
||
retry_result = retry_response.json()
|
||
if retry_result.get('code') == 200:
|
||
data = retry_result.get('data', {})
|
||
success_msg = f"重试批量分配成功 - 成功: {data.get('published_count')}, 失败: {data.get('failed_count')}"
|
||
logger.info(success_msg)
|
||
self.log_to_database('INFO', success_msg, json.dumps(retry_result, ensure_ascii=False))
|
||
return True
|
||
|
||
error_msg = f"重试批量分配失败: {retry_response.status_code}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, retry_response.text)
|
||
return False
|
||
else:
|
||
logger.error("重新登录失败")
|
||
return False
|
||
else:
|
||
error_msg = f"批量分配请求失败: {response.status_code}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, response.text)
|
||
return False
|
||
|
||
except Exception as e:
|
||
error_msg = f"批量分配异常: {e}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, traceback.format_exc())
|
||
return False
|
||
|
||
def run(self):
|
||
"""主运行逻辑"""
|
||
try:
|
||
logger.info("="*60)
|
||
logger.info("开始执行文章自动分配作者任务")
|
||
logger.info("="*60)
|
||
self.log_to_database('INFO', "开始执行文章自动分配作者任务")
|
||
|
||
# 1. 获取待分配作者的文章
|
||
logger.info("步骤1: 获取待分配作者的文章")
|
||
articles = self.get_pending_articles()
|
||
if not articles:
|
||
logger.info("没有待分配作者的文章，任务结束")
|
||
self.log_to_database('INFO', "没有待分配作者的文章，任务结束")
|
||
return True
|
||
|
||
# 2. 获取活跃作者
|
||
logger.info("步骤2: 获取活跃作者")
|
||
authors = self.get_active_authors()
|
||
if not authors:
|
||
logger.error("没有活跃作者，无法分配")
|
||
self.log_to_database('ERROR', "没有活跃作者，无法分配")
|
||
return False
|
||
|
||
# 3. 批量分配作者
|
||
logger.info("步骤3: 批量分配作者")
|
||
article_ids = [article['id'] for article in articles]
|
||
author_ids = [author['id'] for author in authors]
|
||
|
||
logger.info(f"准备分配 {len(article_ids)} 篇文章给 {len(author_ids)} 个作者")
|
||
self.log_to_database('INFO', f"准备分配文章",
|
||
f"文章数: {len(article_ids)}, 作者数: {len(author_ids)}")
|
||
|
||
success = self.assign_authors_to_articles_batch(article_ids, author_ids)
|
||
|
||
if success:
|
||
logger.info("="*60)
|
||
logger.info("文章自动分配作者任务完成")
|
||
logger.info("="*60)
|
||
self.log_to_database('INFO', "文章自动分配作者任务完成")
|
||
return True
|
||
else:
|
||
logger.error("文章自动分配作者任务失败")
|
||
self.log_to_database('ERROR', "文章自动分配作者任务失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
error_msg = f"运行任务异常: {e}"
|
||
logger.error(error_msg)
|
||
self.log_to_database('ERROR', error_msg, traceback.format_exc())
|
||
return False
|
||
|
||
def main():
|
||
"""主函数 - 持续监控模式"""
|
||
generator = AssignAuthorsToArticles()
|
||
|
||
logger.info("="*60)
|
||
logger.info(f"启动文章自动分配作者监控服务")
|
||
logger.info(f"监控间隔: {SLEEP_INTERVAL}秒")
|
||
logger.info("="*60)
|
||
|
||
# 首次登录获取JWT token
|
||
logger.info("首次登录获取JWT token")
|
||
if not generator.login_and_get_jwt_token():
|
||
logger.error("首次登录失败，程序退出")
|
||
generator.log_to_database('ERROR', "首次登录失败，程序退出")
|
||
return
|
||
|
||
while True:
|
||
try:
|
||
# 执行文章自动分配作者任务
|
||
generator.run()
|
||
|
||
# 等待下一次检查
|
||
logger.info(f"等待 {SLEEP_INTERVAL} 秒后进行下一次检查...")
|
||
time.sleep(SLEEP_INTERVAL)
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("="*60)
|
||
logger.info("收到中断信号，停止监控服务")
|
||
logger.info("="*60)
|
||
generator.log_to_database('INFO', '监控服务手动停止', 'KeyboardInterrupt')
|
||
break
|
||
except Exception as e:
|
||
logger.error(f"程序运行异常: {e}")
|
||
generator.log_to_database('ERROR', f'程序运行异常: {e}', traceback.format_exc())
|
||
logger.info(f"等待 {SLEEP_INTERVAL} 秒后重试...")
|
||
time.sleep(SLEEP_INTERVAL)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|