fix(daily-push): 补提交分企业推送 v2——daily-push-split 任务(11:45 .done)漏commit,阻塞CI deploy hook;.gitignore 忽略 *.bak-* (2026-09-03)

This commit is contained in:
Hermes CI Fix
2026-09-03 18:15:05 +08:00
parent d89a5ab777
commit 960cd15268
2 changed files with 76 additions and 21 deletions
+2
View File
@@ -14,3 +14,5 @@ backend/logs/
# 因果链验证报告(生成物)
backend/scripts/reports/
# 脚本备份(手工备份产物,不入库)
*.bak-*
+74 -21
View File
@@ -1,11 +1,16 @@
"""每日数据找人推送 — 路线图R2 (2026-08-30)
"""每日数据找人推送 — 路线图R2 (2026-08-30) / 分企业推送 (2026-09-03)
北极星③:主动推送扩大 —— 异常 + 机会两类。
- 异常类:待处理预警(kpi_alerts pending)
- 机会类:KPI向好 / 预算余量 / 预测上行(opportunity_detector)
- 异常类:待处理预警(kpi_alerts pending,按 entity_id 隔离)
- 机会类:KPI向好 / 预算余量 / 预测上行(opportunity_detector,按 entity_id 隔离)
复用企微通道 8800/send(公司群中继服务)。
用法: /root/cma-management/backend/venv/bin/python3 scripts/daily_push.py [--dry-run]
v2 分企业版(2026-09-03,任总指示:陕西酣客与陕西博海分开汇报避免混淆):
- 默认遍历 entities 表 status='active' 的全部企业,每个企业独立组装+推送
- 消息标题带企业标识:【陕西酣客】/【陕西博海】…,一眼分清
- --entity-id 传入时只推该企业(向后兼容)
用法: /root/cma-management/backend/venv/bin/python3 scripts/daily_push.py [--dry-run] [--entity-id N]
cron: 15 9 * * * (alert_generator 9:00 之后)
"""
import sys
@@ -18,7 +23,7 @@ from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.database import get_session_local
from app.models import KPIDefinition, KPIAlert
from app.models import Entity, KPIDefinition, KPIAlert
from scripts.opportunity_detector import detect_all, flatten
logger = logging.getLogger("cma.daily_push")
@@ -27,12 +32,20 @@ RELAY_URL = "http://127.0.0.1:8800/send"
SOURCE = "管理会计OS"
def collect_exceptions(db, limit: int = 10) -> list:
"""异常类:待处理预警(red/yellow)"""
def entity_display(e) -> str:
"""企业展示名:优先 short_name(酣客/博海),无则用 name"""
if getattr(e, "short_name", None):
return e.short_name
return e.name
def collect_exceptions(db, entity_id: int, limit: int = 10) -> list:
"""异常类:待处理预警(red/yellow),限定企业"""
out = []
alerts = db.query(KPIAlert).filter(
KPIAlert.status == "pending",
KPIAlert.alert_level.in_(["red", "yellow"]),
KPIAlert.entity_id == entity_id,
).order_by(KPIAlert.created_at.desc()).limit(limit).all()
for a in alerts:
k = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first()
@@ -49,10 +62,11 @@ def collect_exceptions(db, limit: int = 10) -> list:
return out
def build_message(exceptions: list, opportunities: list) -> str:
"""组装 markdown 推送内容"""
def build_message(exceptions: list, opportunities: list, entity_name: str = "") -> str:
"""组装 markdown 推送内容(标题带企业标识)"""
now = datetime.now().strftime("%Y-%m-%d %H:%M")
lines = [f"## 📊 管理会计OS · 每日经营播报", f"**{now}**", ""]
tag = f"【{entity_name}】" if entity_name else ""
lines = [f"## 📊 {tag}管理会计OS · 每日经营播报", f"**{now}**", ""]
lines.append("### ⚠️ 异常关注")
if exceptions:
@@ -96,26 +110,65 @@ def push_wecom(msg: str) -> dict:
return {"ok": False, "error": f"推送异常: {e}"}
def resolve_targets(db, entity_id) -> list:
"""--entity-id 传入 → 单企业;否则全部 active 企业(自动适配未来新增企业)"""
if entity_id is not None:
e = db.query(Entity).filter(Entity.id == entity_id).first()
if not e:
raise SystemExit(f"entity_id={entity_id} 不存在")
return [e]
entities = db.query(Entity).filter(Entity.status == "active").order_by(Entity.id).all()
if not entities:
# 兜底:无 active 企业时退回 entity 1(兼容旧默认行为),不静默空跑
e = db.query(Entity).filter(Entity.id == 1).first()
if not e:
raise SystemExit("entities 表无 active 企业且无 entity 1,无法推送")
logger.warning("无 active 企业,退回 entity 1")
return [e]
return entities
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--dry-run", action="store_true", help="只打印不推送")
parser.add_argument("--entity-id", type=int, default=1)
parser.add_argument("--entity-id", type=int, default=None,
help="只推指定企业;不传则遍历全部 active 企业")
args = parser.parse_args()
db = get_session_local()()
try:
exceptions = collect_exceptions(db)
opportunities = flatten(detect_all(db, args.entity_id))
msg = build_message(exceptions, opportunities)
targets = resolve_targets(db, args.entity_id)
logger.info(f"推送目标企业: {[(e.id, entity_display(e)) for e in targets]}")
if args.dry_run:
print(msg)
print(f"\n[DRY-RUN] 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
return
failures = 0
for e in targets:
# 异常隔离:单个企业失败不中断其他企业
try:
entity_name = entity_display(e)
exceptions = collect_exceptions(db, e.id)
opportunities = flatten(detect_all(db, e.id))
msg = build_message(exceptions, opportunities, entity_name)
result = push_wecom(msg)
print(f"推送结果: {json.dumps(result, ensure_ascii=False)}")
print(f"统计: 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
if args.dry_run:
print("=" * 60)
print(f"企业: {e.name} (entity_id={e.id})")
print("=" * 60)
print(msg)
print(f"\n[DRY-RUN][{entity_name}] 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
continue
result = push_wecom(msg)
ok = result.get("ok", False)
print(f"[{entity_name}] 推送结果: {json.dumps(result, ensure_ascii=False)}")
print(f"[{entity_name}] 统计: 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
if not ok:
failures += 1
except Exception as exc:
failures += 1
logger.exception(f"企业 {getattr(e, 'id', '?')} 推送失败: {exc}")
if failures and not args.dry_run:
raise SystemExit(f"{failures} 个企业推送失败")
finally:
db.close()