60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
||
"""行动计划执行周报 — 每周一08:00推送
|
||
|
||
统计本周通过/失败/升级/待处理计划数，推送任总
|
||
"""
|
||
import sys, os
|
||
from datetime import datetime, timedelta
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.database import get_db
|
||
from app.models import ActionPlan
|
||
from app.utils.notifier import send_wecom_message
|
||
|
||
|
||
def generate_weekly_report():
|
||
db = next(get_db())
|
||
try:
|
||
week_ago = datetime.now() - timedelta(days=7)
|
||
# 本周所有计划（最近7天创建的）
|
||
plans = db.query(ActionPlan).filter(ActionPlan.created_at >= week_ago).all()
|
||
if not plans:
|
||
plans = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(10).all()
|
||
|
||
total = len(plans)
|
||
passed = sum(1 for p in plans if p.verify_status == "passed")
|
||
failed = sum(1 for p in plans if p.verify_status == "failed")
|
||
escalated = sum(1 for p in plans if p.verify_status == "escalated")
|
||
pending = sum(1 for p in plans if p.verify_status in ("pending", "retrying"))
|
||
done = sum(1 for p in plans if p.status == "done")
|
||
|
||
msg = (
|
||
f"📋 行动计划执行周报（{datetime.now().strftime('%Y-%m-%d')}）\n"
|
||
f"统计范围：近7天计划 {total} 个\n"
|
||
f"✅ 验证通过: {passed}\n"
|
||
f"❌ 验证失败: {failed}\n"
|
||
f"🚨 升级人工: {escalated}\n"
|
||
f"⏳ 待验证: {pending}\n"
|
||
f"🏁 已完成: {done}\n"
|
||
)
|
||
# 列出待验证/升级项
|
||
urgent = [p for p in plans if p.verify_status in ("escalated", "retrying")]
|
||
if urgent:
|
||
msg += "\n⚠️ 需关注:\n"
|
||
for p in urgent[:5]:
|
||
msg += f" #{p.id} {p.title} ({p.verify_status})\n"
|
||
|
||
try:
|
||
send_wecom_message(msg, title="行动计划周报")
|
||
except Exception:
|
||
pass
|
||
print(msg)
|
||
return {"total": total, "passed": passed, "failed": failed, "escalated": escalated}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
generate_weekly_report()
|