lt
296 lines
12 KiB
Python
296 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
合规审查与优化任务
|
||
每天 05:45 运行，处理当天所有 draft 文章：
|
||
1. 执行合规检查（compliance_checker）
|
||
2. 自动修复已知问题（标题、标签）
|
||
3. 重写合规版本
|
||
4. 更新选题状态为「审查通过待发布」
|
||
5. 生成优化报告通知
|
||
"""
|
||
|
||
import json, datetime, logging, sys, re
|
||
from pathlib import Path
|
||
from typing import Dict, List
|
||
from dataclasses import dataclass, asdict
|
||
|
||
PROJECT_ROOT = Path('/root/.openclaw/workspaces/yzr-yxl/projects/yu-zhi-ran')
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
from scripts.compliance_checker import check_article
|
||
|
||
# 导入 LLM 客户端（合规优化使用 NVIDIA）
|
||
sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend"))
|
||
try:
|
||
from app.core.nvidia_client import call_llm
|
||
HAVE_LLM = True
|
||
except ImportError:
|
||
HAVE_LLM = False
|
||
|
||
DATA_DIR = PROJECT_ROOT / "automation" / "data"
|
||
RELEASES_DIR = DATA_DIR / "releases"
|
||
DRAFTS_DIR = DATA_DIR / "drafts"
|
||
TOPICS_FILE = DATA_DIR / "sustainability_topics.json"
|
||
LOGS_DIR = PROJECT_ROOT / "automation" / "logs"
|
||
TODAY = datetime.datetime.now().strftime("%Y-%m-%d")
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[logging.FileHandler(LOGS_DIR / f"optimizer_{TODAY}.log"), logging.StreamHandler()])
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 平台白名单标签
|
||
PLATFORM_TAGS = {
|
||
"zhihu": ["科技", "职场"],
|
||
"xiaohongshu": ["AI", "可持续", "生活方式"]
|
||
}
|
||
|
||
@dataclass
|
||
class OptimizationResult:
|
||
file: str
|
||
platform: str
|
||
topic_id: str
|
||
title: str
|
||
original_issues: int
|
||
fixed_issues: int
|
||
final_score: int
|
||
status: str
|
||
|
||
def load_topic_map():
|
||
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
|
||
topics = json.load(f)
|
||
return {t['id']: t for t in topics}
|
||
|
||
|
||
def update_topic_status(topic_id: str, status: str):
|
||
"""更新选题状态（JSON + 数据库）"""
|
||
# 更新 JSON
|
||
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
|
||
topics = json.load(f)
|
||
updated = False
|
||
for t in topics:
|
||
if t.get('id') == topic_id:
|
||
t['status'] = status
|
||
updated = True
|
||
break
|
||
if updated:
|
||
with open(TOPICS_FILE, 'w', encoding='utf-8') as f:
|
||
json.dump(topics, f, ensure_ascii=False, indent=2)
|
||
# 更新数据库
|
||
try:
|
||
from app.database import SessionLocal
|
||
from app.models import Topic
|
||
db = SessionLocal()
|
||
topic_db = db.query(Topic).filter(Topic.id == topic_id).first()
|
||
if topic_db:
|
||
topic_db.status = status
|
||
db.commit()
|
||
db.close()
|
||
except Exception as e:
|
||
logger.error(f"更新数据库失败: {e}")
|
||
|
||
def fix_wechat_title(html: str, title: str) -> str:
|
||
"""微信标题优化：<title>和<h1>都控制长度（考虑后缀）"""
|
||
suffix = f" - {TODAY} - 微信公众号"
|
||
max_base_len = 32 - len(suffix) # <title> 中 base 部分允许的最大长度
|
||
|
||
# 处理 <title>...</title>
|
||
title_tag = re.search(r'<title>([^<]+)</title>', html)
|
||
if title_tag:
|
||
full_title = title_tag.group(1)
|
||
# 提取 base（去掉后缀）
|
||
if full_title.endswith(suffix):
|
||
base = full_title[:-len(suffix)]
|
||
else:
|
||
base = full_title.split(" - ")[0]
|
||
if len(base) > max_base_len:
|
||
base = base[:max_base_len-3] + "..."
|
||
new_full = base + suffix
|
||
html = html.replace(full_title, new_full)
|
||
|
||
# 处理 <h1>...</h1>（不含后缀，但要截断）
|
||
h1_match = re.search(r'<h1[^>]*>([^<]+)</h1>', html)
|
||
if h1_match:
|
||
current_h1 = h1_match.group(1)
|
||
# 如果 h1 包含后缀（不应该），去掉
|
||
base_h1 = current_h1.split(" - ")[0] if " - " in current_h1 else current_h1
|
||
if len(base_h1) > 32:
|
||
base_h1 = base_h1[:29] + "..."
|
||
html = html.replace(current_h1, base_h1)
|
||
|
||
return html
|
||
|
||
def fix_tags(html: str, platform: str) -> str:
|
||
"""强制替换标签为平台白名单"""
|
||
if platform == "zhihu":
|
||
tags_str = " ".join(f"#{t}" for t in PLATFORM_TAGS["zhihu"])
|
||
# 替换 <div class="tags">...</div>
|
||
if '<div class="tags">' in html:
|
||
old = html.split('<div class="tags">')[1].split('</div>')[0]
|
||
html = html.replace(f'<div class="tags">{old}</div>', f'<div class="tags">{tags_str}</div>')
|
||
elif platform == "xiaohongshu":
|
||
tags_str = " ".join(f"#{t}" for t in PLATFORM_TAGS["xiaohongshu"])
|
||
if '<div class="hashtags">' in html:
|
||
old = html.split('<div class="hashtags">')[1].split('</div>')[0]
|
||
html = html.replace(f'<div class="hashtags">{old}</div>', f'<div class="hashtags">{tags_str}</div>')
|
||
return html
|
||
|
||
def optimize_article(html: str, platform: str, topic_data: Dict) -> (str, List[str]):
|
||
logs = []
|
||
# 1. 标题优化（微信）
|
||
if platform == "wechat":
|
||
html = fix_wechat_title(html, topic_data.get("title", ""))
|
||
logs.append("标题截断（含后缀）")
|
||
# 2. 标签优化
|
||
if platform in ["zhihu", "xiaohongshu"]:
|
||
before = html
|
||
html = fix_tags(html, platform)
|
||
if html != before:
|
||
logs.append(f"标签标准化为{PLATFORM_TAGS[platform]}")
|
||
# 3. 图片内联检查
|
||
# 提取所有 img 标签
|
||
img_tags = re.findall(r'<img[^>]*>', html, re.IGNORECASE)
|
||
for tag in img_tags:
|
||
m = re.search(r'src=["\']([^"\']+)["\']', tag, re.IGNORECASE)
|
||
if m:
|
||
src = m.group(1)
|
||
if not src.startswith('data:image/'):
|
||
logs.append(f"图片未内联: {src[:50]}... 需手动修复")
|
||
|
||
# 4. LLM 内容优化（使用 NVIDIA step-3.5-flash）
|
||
if HAVE_LLM:
|
||
try:
|
||
polish_prompt = f"""你是一个专业的内容润色助手。请优化以下文章内容，提升表达的专业性和可读性，保持原文事实、数据、章节结构不变，输出相同的HTML格式（保留<h2>, <h3>, <p>标签）。
|
||
|
||
原文：
|
||
{html}
|
||
|
||
优化后："""
|
||
polished = call_llm(polish_prompt, temperature=0.5, max_tokens=4000)
|
||
if '<h2' in polished or '<p>' in polished:
|
||
html = polished
|
||
logs.append("LLM 内容优化（NVIDIA）")
|
||
except Exception as e:
|
||
logger.warning(f"LLM 优化失败: {e}")
|
||
return html, logs
|
||
|
||
def main(topic_ids: List[str] = None):
|
||
logger.info("=== 合规审查与优化开始 ===")
|
||
release_dir = RELEASES_DIR / TODAY
|
||
if not release_dir.exists():
|
||
logger.warning(f"今日发布目录不存在: {release_dir}")
|
||
return
|
||
|
||
topic_map = load_topic_map()
|
||
results = []
|
||
all_passed = True
|
||
|
||
for platform_dir in ["zhihu", "wechat", "xiaohongshu"]:
|
||
platform_path = release_dir / platform_dir
|
||
if not platform_path.exists():
|
||
continue
|
||
for html_file in platform_path.glob("*.html"):
|
||
stem = html_file.stem
|
||
parts = stem.split('_')
|
||
if len(parts) < 2:
|
||
continue
|
||
topic_id = parts[1]
|
||
# 如果指定了 topic_ids，则只处理匹配的
|
||
if topic_ids is not None and topic_id not in topic_ids:
|
||
continue
|
||
topic_data = topic_map.get(topic_id)
|
||
if not topic_data:
|
||
logger.warning(f"未找到选题: {topic_id}")
|
||
continue
|
||
|
||
html = html_file.read_text(encoding='utf-8')
|
||
check_result = check_article(html, platform_dir, topic_data)
|
||
issues = check_result['issues']
|
||
score = check_result['score']
|
||
|
||
if any(issue['type'] == '平台规则' for issue in issues):
|
||
optimized_html, opt_logs = optimize_article(html, platform_dir, topic_data)
|
||
recheck = check_article(optimized_html, platform_dir, topic_data)
|
||
if recheck['passed']:
|
||
html_file.write_text(optimized_html, encoding='utf-8')
|
||
logger.info(f"✅ {html_file.name} 已优化并通过合规检查")
|
||
results.append(OptimizationResult(
|
||
file=str(html_file.relative_to(PROJECT_ROOT)),
|
||
platform=platform_dir,
|
||
topic_id=topic_id,
|
||
title=topic_data.get('title',''),
|
||
original_issues=len(issues),
|
||
fixed_issues=len(issues) - len(recheck['issues']),
|
||
final_score=recheck['score'],
|
||
status="passed"
|
||
))
|
||
update_topic_status(topic_id, '待发布')
|
||
else:
|
||
logger.warning(f"⚠️ {html_file.name} 优化后仍有问题，需人工审核")
|
||
results.append(OptimizationResult(
|
||
file=str(html_file.relative_to(PROJECT_ROOT)),
|
||
platform=platform_dir,
|
||
topic_id=topic_id,
|
||
title=topic_data.get('title',''),
|
||
original_issues=len(issues),
|
||
fixed_issues=len(issues) - len(recheck['issues']),
|
||
final_score=recheck['score'],
|
||
status="manual_review"
|
||
))
|
||
all_passed = False
|
||
else:
|
||
results.append(OptimizationResult(
|
||
file=str(html_file.relative_to(PROJECT_ROOT)),
|
||
platform=platform_dir,
|
||
topic_id=topic_id,
|
||
title=topic_data.get('title',''),
|
||
original_issues=0,
|
||
fixed_issues=0,
|
||
final_score=score,
|
||
status="passed" if check_result['passed'] else "manual_review"
|
||
))
|
||
if not check_result['passed']:
|
||
all_passed = False
|
||
|
||
# 更新选题状态
|
||
for res in results:
|
||
if res.status == "passed":
|
||
tid = res.topic_id
|
||
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
|
||
topics = json.load(f)
|
||
for t in topics:
|
||
if t.get('id') == tid:
|
||
t['status'] = '待发布'
|
||
t['ready_at'] = TODAY
|
||
t['compliance_score'] = res.final_score
|
||
break
|
||
with open(TOPICS_FILE, 'w', encoding='utf-8') as f:
|
||
json.dump(topics, f, ensure_ascii=False, indent=2)
|
||
|
||
# 生成报告
|
||
report_file = DRAFTS_DIR / TODAY / "optimization_report.json"
|
||
report_file.parent.mkdir(parents=True, exist_ok=True)
|
||
report = {
|
||
"date": TODAY,
|
||
"summary": {
|
||
"total_articles": len(results),
|
||
"passed_auto": sum(1 for r in results if r.status == "passed"),
|
||
"need_manual": sum(1 for r in results if r.status == "manual_review"),
|
||
"average_score": sum(r.final_score for r in results) / len(results) if results else 0
|
||
},
|
||
"details": [asdict(r) for r in results],
|
||
"all_passed": all_passed
|
||
}
|
||
with open(report_file, 'w', encoding='utf-8') as f:
|
||
json.dump(report, f, ensure_ascii=False, indent=2)
|
||
|
||
logger.info(f"✅ 合规优化完成: {len(results)} 篇文章, {sum(1 for r in results if r.status=='passed')} 篇自动通过")
|
||
print(f"OPTIMIZATION_COMPLETE: {len(results)} articles, {sum(1 for r in results if r.status=='passed')} passed, {sum(1 for r in results if r.status=='manual_review')} need manual review")
|
||
sys.exit(0)
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description='合规审查与优化任务')
|
||
parser.add_argument('--topic-ids', help='逗号分隔的选题ID列表，例如: A01,B02')
|
||
args = parser.parse_args()
|
||
topic_ids = args.topic_ids.split(',') if args.topic_ids else None
|
||
main(topic_ids)
|