242 lines
9.4 KiB
Python
242 lines
9.4 KiB
Python
"""
|
||
AI分析模块：四维度评分、智能分类、中文摘要、分类洞察、网摘生成
|
||
"""
|
||
import re
|
||
import json
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from .config import SCORE_DIMENSIONS, WECHAT_SOURCES, AI_CONCURRENCY
|
||
from .utils import call_ai
|
||
from .logger import get_logger
|
||
|
||
|
||
def score_and_classify_article(article, api_key, base_url, model, index=0, total=0):
|
||
"""
|
||
AI三合一分析：四维度评分 + 中文摘要 + 智能分类
|
||
一次API调用完成三项任务，返回包含 scores/summary/category 的字典
|
||
"""
|
||
logger = get_logger()
|
||
title = article['translated_title']
|
||
content = article['translated_content'][:1000] if article['translated_content'] else article['translated_title']
|
||
source = article['source']
|
||
|
||
dimensions_desc = '\n'.join(
|
||
f'- {d["name"]} (权重 {int(d["weight"]*100)}%): {d["description"]}'
|
||
for d in SCORE_DIMENSIONS
|
||
)
|
||
|
||
prompt = f"""请对以下军事科技文章进行分析。
|
||
|
||
【文章标题】：{title}
|
||
|
||
【文章内容摘要】：{content}
|
||
|
||
【文章来源】：{source}
|
||
|
||
【评分维度】：
|
||
{dimensions_desc}
|
||
|
||
【输出要求】：
|
||
请严格按照以下 JSON 格式返回（不要其他文字说明）：
|
||
{{
|
||
"scores": {{
|
||
"维度名称1": 分数（1-10的整数）,
|
||
"维度名称2": 分数（1-10的整数）
|
||
}},
|
||
"summary": "2-3句话的高质量中文摘要，突出技术参数、战略影响、国际竞争格局",
|
||
"category": "文章分类，必须是以下之一：装备动态/地区冲突/战略政策"
|
||
}}
|
||
|
||
只输出 JSON，不要其他内容。"""
|
||
|
||
try:
|
||
result = call_ai([
|
||
{'role': 'system', 'content': '你是专业的军事科技分析专家，擅长武器装备分析、战略评估。'},
|
||
{'role': 'user', 'content': prompt}
|
||
], api_key, base_url, model, temperature=0.2, max_tokens=800, purpose="AI评分分类")
|
||
|
||
json_match = re.search(r'\{[\s\S]*\}', result)
|
||
if not json_match:
|
||
raise ValueError('未找到JSON输出')
|
||
|
||
parsed = json.loads(json_match[0])
|
||
if index > 0 and total > 0:
|
||
logger.info("AI分析完成 %d/%d: %s...", index, total, title[:30])
|
||
return (article, parsed)
|
||
|
||
except Exception as e:
|
||
logger.warning("AI处理失败 '%s...': %s", title[:30], e)
|
||
default_scores = {d['name']: 5 for d in SCORE_DIMENSIONS}
|
||
default_result = {
|
||
'scores': default_scores,
|
||
'summary': title,
|
||
'category': '装备动态'
|
||
}
|
||
return (article, default_result)
|
||
|
||
|
||
def batch_score_and_classify_articles(articles, api_key, base_url, model, max_concurrent=None):
|
||
"""批量对文章进行AI评分与分类，并发执行
|
||
返回处理后的文章列表，文章已包含ai_result、final_score、summary、category等字段
|
||
"""
|
||
logger = get_logger()
|
||
if not articles:
|
||
return []
|
||
|
||
if max_concurrent is None:
|
||
max_concurrent = AI_CONCURRENCY
|
||
|
||
logger.info("开始批量AI分析，共%d篇文章，并发数：%d", len(articles), max_concurrent)
|
||
processed_articles = []
|
||
|
||
with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
|
||
future_to_article = {
|
||
executor.submit(score_and_classify_article, article, api_key, base_url, model, i+1, len(articles)): article
|
||
for i, article in enumerate(articles)
|
||
}
|
||
|
||
for future in as_completed(future_to_article):
|
||
article, ai_result = future.result()
|
||
article['ai_result'] = ai_result
|
||
article['final_score'] = calculate_weighted_score(ai_result)
|
||
article['summary'] = ai_result.get('summary', article['translated_title'])
|
||
article['category'] = ai_result.get('category', '装备动态')
|
||
article['scores'] = ai_result.get('scores', {})
|
||
processed_articles.append(article)
|
||
|
||
logger.info("批量AI分析完成，共处理%d篇文章", len(processed_articles))
|
||
return processed_articles
|
||
|
||
|
||
def calculate_weighted_score(ai_result):
|
||
"""根据四维度分数和权重计算加权总分"""
|
||
scores = ai_result.get('scores', {})
|
||
total = 0.0
|
||
for dim in SCORE_DIMENSIONS:
|
||
score = scores.get(dim['name'], 5)
|
||
total += score * dim['weight']
|
||
return round(total, 1)
|
||
|
||
|
||
def generate_category_overview(category_name, articles, api_key, base_url, model):
|
||
"""为某分类生成50字左右的总体情况介绍"""
|
||
logger = get_logger()
|
||
if not articles:
|
||
return ""
|
||
|
||
article_titles = "\u3001".join(a['translated_title'][:15] for a in articles[:5])
|
||
|
||
prompt = f"""请为以下军事科技新闻分类生成50字左右的总体情况介绍。
|
||
|
||
【分类名称】：{category_name}
|
||
|
||
【本分类文章】：{article_titles}
|
||
|
||
【要求】：
|
||
- 50字左右，简洁专业
|
||
- 概括本分类今日核心看点或趋势
|
||
- 军事科技专业风格，不要口语化
|
||
|
||
直接输出文字，不要其他格式。"""
|
||
|
||
try:
|
||
result = call_ai([
|
||
{'role': 'system', 'content': '你是专业的军事科技编辑，擅长撰写简洁专业的新闻综述。'},
|
||
{'role': 'user', 'content': prompt}
|
||
], api_key, base_url, model, temperature=0.3, max_tokens=150, purpose="分类介绍")
|
||
return result.strip()
|
||
except Exception as e:
|
||
logger.warning("生成分类介绍失败: %s", e)
|
||
return ""
|
||
|
||
|
||
def generate_webzine_for_article(article, api_key, base_url, model, max_retries=3):
|
||
"""
|
||
为单篇文章生成《参考消息》风格的网摘
|
||
支持字数校验重试（正文280-320字、价值点35-45字），最多重试3次
|
||
"""
|
||
logger = get_logger()
|
||
source_name = article['source']
|
||
publish_time = article['published'].strftime('%Y年%m月%d日')
|
||
|
||
is_wechat = source_name in WECHAT_SOURCES
|
||
source_prefix = f"据公众号 {source_name} 报道" if is_wechat else f"据 {source_name} 报道"
|
||
|
||
prompt = f"""文章标题：{article['translated_title']}
|
||
文章来源：{source_name}
|
||
文章发布时间：{publish_time}
|
||
文章内容：{article.get('summary', article.get('content', '')[:800])}
|
||
文章链接：{article['link']}
|
||
|
||
【重要警告：字数必须严格达标，否则视为不合格！】
|
||
|
||
请你严格按照 《参考消息》官方新闻报道格式、标题风格 ，对我提供的新闻内容进行改写，严格遵守以下所有规则：
|
||
|
||
标题要求
|
||
简洁、客观、中性、信息密度高
|
||
结构：主体 + 事件 + 核心态势
|
||
不抒情、不夸张、不用网络用语
|
||
|
||
正文格式要求
|
||
开头第一句必须加：{source_prefix}
|
||
正文风格：客观、平实、严谨、书面化，类似外电编译稿
|
||
必须充分展开：补充背景、说明意义、分析影响、展望前景，确保内容充实
|
||
只保留核心信息：时间、地点、人物、事件、内容、前景
|
||
全文一段到底，不分段
|
||
|
||
【字数强制要求 - 必须严格执行】
|
||
正文字数严格控制在 280～320 字（含标点），一个字都不能少，一个字都不能多！
|
||
300字左右的正文示例长度：大概5-6个完整的长句，涵盖事件背景、核心内容、技术细节、战略意义、未来影响五个层面。
|
||
如果内容不够，请合理补充相关背景、行业态势、同类项目对比等专业内容，确保达到字数要求。
|
||
|
||
价值点要求
|
||
正文结束后空一行，再写价值点
|
||
价值点严格 35～45 字（含标点）
|
||
句式结构：事件 - 影响 / 后果 - 值得关注
|
||
凝练、客观、不发散
|
||
|
||
输出格式
|
||
标题：新闻标题
|
||
原标题：原文章标题
|
||
发布日期：严格使用上面提供的「{publish_time}」，不要修改
|
||
正文：改写后的正文内容
|
||
价值点：价值点内容
|
||
|
||
注意：每一行都必须以「类别：内容」的格式输出，类别严格使用上面的5个名称，不要修改。"""
|
||
|
||
last_result = None
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
result = call_ai([
|
||
{'role': 'system', 'content': '你是《参考消息》资深编辑，擅长撰写客观、严谨、专业的新闻报道，对字数控制极其精准。'},
|
||
{'role': 'user', 'content': prompt}
|
||
], api_key, base_url, model, temperature=0.4, max_tokens=800, purpose="网摘生成")
|
||
|
||
result = result.strip()
|
||
last_result = result
|
||
|
||
lines = result.split('\n')
|
||
body_text = ''
|
||
for line in lines:
|
||
if line.startswith('正文：'):
|
||
body_text = line[3:].strip()
|
||
break
|
||
|
||
body_len = len(body_text)
|
||
if 280 <= body_len <= 320:
|
||
if attempt > 0:
|
||
logger.info("第%d次尝试成功，字数：%d", attempt + 1, body_len)
|
||
return result
|
||
else:
|
||
logger.warning("第%d次尝试字数不达标：%d字（要求280-320），重试中...", attempt + 1, body_len)
|
||
if body_len < 280:
|
||
prompt += f"\n\n【上次反馈：正文只有{body_len}字，太少！请大幅增加内容，补充更多背景、细节和分析，确保达到280-320字！】"
|
||
else:
|
||
prompt += f"\n\n【上次反馈：正文有{body_len}字，太多！请精简内容，删除冗余描述，控制在280-320字以内！】"
|
||
|
||
except Exception as e:
|
||
logger.warning("第%d次尝试失败: %s", attempt + 1, e)
|
||
|
||
logger.warning("重试%d次仍未达标，返回最后结果", max_retries)
|
||
return last_result if last_result else ""
|