feat(ai): ai_reports 扩字段 + 预生成 worker (#121)
#121 第二阶段基础工单, 包含两部分: ## 1. ai_reports 表扩字段(Alembic migration d2e3f4a5b6c7) 新增 5 字段 + 2 索引: - template_id FK ai_report_templates ON DELETE SET NULL(模板被删时报告保留) - snapshot_date: 报告数据时点(区别于 created_at 入库时点) - snapshot_period: daily/weekly/monthly/range(与模板 period 对应) - is_public default=false: true=公共报告(预生成, 所有人可看); false=私有(实时, 仅本人) - matched_count default=0: 被命中复用次数(match_template_report 命中时 +1) 索引: - ix_ai_reports_template_snapshot (template_id, snapshot_date) - 匹配查询用 - ix_ai_reports_public_created (is_public, created_at) - 公共报告列表用 ## 2. ai_report_worker.py 预生成 worker 核心函数: - schedule_due_templates(db): 扫描 ai_report_templates, 找当前时刻匹配的 daily/weekly/monthly 模板 (trigger_at=HH:MM / wday:HH:MM / day:HH:MM, ±1 分钟容差) - generate_for_template(db, template): 用模板 prompt 调 LLM(timeout_s=180, max_tokens=template.max_tokens) 落库 is_public=True + template_id + snapshot_date; 失败写 system_event(category=ai) - warm_on_sync_complete: 同步完成后触发 sync-triggered 模板(待 #123 接入 sync_task_manager) - warm_on_startup(force=False): 启动延迟触发 active 模板(待 #123 接入 start_warm_loop) - warm_idle: 空闲期每分钟扫描(待 #123 接入 bms_scheduler) 设计要点: - 进程级单飞锁 _worker_lock 防并发(单 worker 串行) - 失败仅 log + system_event, 不阻塞主流程 - 白天 8-23 才预热(夜间无人访问省计算) - chat_completion 是 async, worker 用 asyncio.run/iscoroutine 桥接 ## 3. save_report 扩参数(向后兼容) ai_report_service.save_report 增加 template_id/snapshot_date/snapshot_period/is_public 可选参数(默认 None/None/None/False), 实时调用零侵入。 本地校验: - ruff All checks passed - alembic upgrade head 成功(d2e3f4a5b6c7) - 空表 schedule_due_templates/warm_idle 测试通过 #123 将在 llm_scheduler.py 接入 sync_task_manager + bms_scheduler + 优先级队列。
This commit is contained in:
77
backend/alembic/versions/d2e3f4a5b6c7_ai_reports_ext.py
Normal file
77
backend/alembic/versions/d2e3f4a5b6c7_ai_reports_ext.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""ai_reports 扩字段: template_id/snapshot_date/snapshot_period/is_public/matched_count (工单 #121)
|
||||
|
||||
Revision ID: d2e3f4a5b6c7
|
||||
Revises: c1d2e3f4a5b6
|
||||
Create Date: 2026-09-16 13:30:00.000000
|
||||
|
||||
设计:
|
||||
- ai_reports 表加 5 字段, 支撑预生成报告与匹配命中复用统计
|
||||
- template_id FK ai_report_templates ON DELETE SET NULL(模板被删时报告保留)
|
||||
- snapshot_date/snapshot_period 报告数据时点(区别于 created_at 入库时点)
|
||||
- is_public 区分公共报告(预生成, 所有人可看)与私有报告(实时, 仅本人)
|
||||
- matched_count 命中 +1(用 UPDATE ... RETURNING 保证并发安全, 业务层负责)
|
||||
- 2 索引: (template_id, snapshot_date) 匹配查询用; (is_public, created_at) 公共报告列表用
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = 'd2e3f4a5b6c7'
|
||||
down_revision: Union[str, None] = 'c1d2e3f4a5b6'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# 5 新增字段
|
||||
op.add_column('ai_reports', sa.Column(
|
||||
'template_id', sa.Integer(), nullable=True,
|
||||
comment='#121: 来自预生成则为模板 id, 实时生成则 NULL',
|
||||
))
|
||||
op.add_column('ai_reports', sa.Column(
|
||||
'snapshot_date', sa.Date(), nullable=True,
|
||||
comment='#121: 报告数据时点(窗口起算 date, 区别于 created_at 的入库时点)',
|
||||
))
|
||||
op.add_column('ai_reports', sa.Column(
|
||||
'snapshot_period', sa.String(length=16), nullable=True,
|
||||
comment='#121: daily/weekly/monthly/range(与模板 period 对应)',
|
||||
))
|
||||
op.add_column('ai_reports', sa.Column(
|
||||
'is_public', sa.Boolean(), nullable=False, server_default=sa.false(),
|
||||
comment='#121: true=公共报告(所有人可看, 来自预生成); false=私有(仅本人, 来自实时)',
|
||||
))
|
||||
op.add_column('ai_reports', sa.Column(
|
||||
'matched_count', sa.Integer(), nullable=False, server_default='0',
|
||||
comment='#121: 被命中复用次数(match_template_report 命中时 +1)',
|
||||
))
|
||||
|
||||
# FK: template_id -> ai_report_templates.id ON DELETE SET NULL
|
||||
op.create_foreign_key(
|
||||
'fk_ai_reports_template_id',
|
||||
'ai_reports', 'ai_report_templates',
|
||||
['template_id'], ['id'],
|
||||
ondelete='SET NULL',
|
||||
)
|
||||
|
||||
# 索引
|
||||
op.create_index(
|
||||
'ix_ai_reports_template_snapshot',
|
||||
'ai_reports', ['template_id', 'snapshot_date'],
|
||||
)
|
||||
op.create_index(
|
||||
'ix_ai_reports_public_created',
|
||||
'ai_reports', ['is_public', 'created_at'],
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index('ix_ai_reports_public_created', table_name='ai_reports')
|
||||
op.drop_index('ix_ai_reports_template_snapshot', table_name='ai_reports')
|
||||
op.drop_constraint('fk_ai_reports_template_id', 'ai_reports', type_='foreignkey')
|
||||
op.drop_column('ai_reports', 'matched_count')
|
||||
op.drop_column('ai_reports', 'is_public')
|
||||
op.drop_column('ai_reports', 'snapshot_period')
|
||||
op.drop_column('ai_reports', 'snapshot_date')
|
||||
op.drop_column('ai_reports', 'template_id')
|
||||
@@ -15,7 +15,7 @@
|
||||
- 规则层秒回路径(#98 层3)落库时 llm_model="rule-engine", tokens=NULL
|
||||
"""
|
||||
|
||||
from sqlalchemy import BigInteger, Date, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy import BigInteger, Boolean, Date, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.core.database import Base
|
||||
@@ -23,7 +23,14 @@ from app.models.base import IDMixin, TimestampMixin
|
||||
|
||||
|
||||
class AiReport(IDMixin, TimestampMixin, Base):
|
||||
"""AI 分析报告历史(仅本人可见; 管理员可查全部)。"""
|
||||
"""AI 分析报告历史(仅本人可见; 管理员可查全部)。
|
||||
|
||||
#121 扩字段:
|
||||
- template_id: 来自预生成则为模板 id, 实时生成则 NULL
|
||||
- snapshot_date / snapshot_period: 数据时点(区别于 created_at)
|
||||
- is_public: true=公共报告(所有人可看, 来自预生成), false=私有(仅本人, 来自实时)
|
||||
- matched_count: 被命中复用次数
|
||||
"""
|
||||
|
||||
__tablename__ = "ai_reports"
|
||||
|
||||
@@ -33,7 +40,28 @@ class AiReport(IDMixin, TimestampMixin, Base):
|
||||
)
|
||||
user_id: Mapped[int] = mapped_column(
|
||||
BigInteger, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True,
|
||||
comment="归属用户 id(仅本人可见, admin 可查全部)",
|
||||
comment="归属用户 id(仅本人可见, admin 可查全部); 公共报告(is_public=True)所有人可看",
|
||||
)
|
||||
template_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("ai_report_templates.id", ondelete="SET NULL"),
|
||||
nullable=True, index=True,
|
||||
comment="#121: 来自预生成则为模板 id, 实时生成则 NULL",
|
||||
)
|
||||
snapshot_date: Mapped[str | None] = mapped_column(
|
||||
Date, nullable=True, index=True,
|
||||
comment="#121: 报告数据时点(窗口起算 date, 区别于 created_at 的入库时点)",
|
||||
)
|
||||
snapshot_period: Mapped[str | None] = mapped_column(
|
||||
String(16), nullable=True,
|
||||
comment="#121: daily/weekly/monthly/range(与模板 period 对应)",
|
||||
)
|
||||
is_public: Mapped[bool] = mapped_column(
|
||||
Boolean, nullable=False, default=False,
|
||||
comment="#121: true=公共报告(所有人可看, 来自预生成); false=私有(仅本人, 来自实时)",
|
||||
)
|
||||
matched_count: Mapped[int] = mapped_column(
|
||||
Integer, nullable=False, default=0,
|
||||
comment="#121: 被命中复用次数(match_template_report 命中时 +1)",
|
||||
)
|
||||
question: Mapped[str] = mapped_column(Text, nullable=False, comment="原问句")
|
||||
period: Mapped[str | None] = mapped_column(
|
||||
|
||||
@@ -38,11 +38,19 @@ def save_report(
|
||||
llm_ms: int,
|
||||
total_ms: int,
|
||||
tokens: dict | None,
|
||||
# #121 新增预生成报告相关字段(向后兼容, 默认值让既有实时调用零侵入)
|
||||
template_id: int | None = None,
|
||||
snapshot_date: str | None = None,
|
||||
snapshot_period: str | None = None,
|
||||
is_public: bool = False,
|
||||
) -> AiReport | None:
|
||||
"""落库一条 AI 报告。
|
||||
|
||||
tokens: OpenAI usage dict {prompt_tokens, completion_tokens, total_tokens}; None 表示服务端不返
|
||||
返回 AiReport 实例; 失败返回 None(调用方决定是否影响主流程, 异步调用时不影响)。
|
||||
|
||||
#121: template_id/snapshot_date/snapshot_period/is_public 来自预生成 worker;
|
||||
实时调用默认 None/None/None/False(私有报告), 零侵入兼容。
|
||||
"""
|
||||
try:
|
||||
prompt_t = tokens.get("prompt_tokens") if tokens else None
|
||||
@@ -65,12 +73,17 @@ def save_report(
|
||||
prompt_tokens=prompt_t,
|
||||
completion_tokens=completion_t,
|
||||
total_tokens=total_t,
|
||||
template_id=template_id,
|
||||
snapshot_date=snapshot_date,
|
||||
snapshot_period=snapshot_period,
|
||||
is_public=is_public,
|
||||
)
|
||||
db.add(report)
|
||||
db.commit()
|
||||
db.refresh(report)
|
||||
logger.info("ai_report saved: id=%s task_id=%s user_id=%s llm_model=%s total_ms=%s",
|
||||
report.id, task_id, user_id, llm_model, total_ms)
|
||||
logger.info("ai_report saved: id=%s task_id=%s user_id=%s llm_model=%s total_ms=%s "
|
||||
"template_id=%s is_public=%s",
|
||||
report.id, task_id, user_id, llm_model, total_ms, template_id, is_public)
|
||||
return report
|
||||
except Exception as e: # noqa: BLE001 落库失败不阻塞主响应
|
||||
db.rollback()
|
||||
|
||||
335
backend/app/services/ai_report_worker.py
Normal file
335
backend/app/services/ai_report_worker.py
Normal file
@@ -0,0 +1,335 @@
|
||||
"""AI 报告预生成 worker (工单 #121)
|
||||
|
||||
设计:
|
||||
- 复用 cache_warmer 三时机设计, 但 LLM 调用与缓存路由独立
|
||||
- 三时机:
|
||||
① 同步完成后(sync_task_manager._run_background finally) 触发 sync-triggered 模板
|
||||
② 服务启动后延迟 8s(start_warm_loop) 触发 active 模板
|
||||
③ 空闲期(bms_scheduler._run_loop 每分钟) 扫描 daily/weekly/monthly + trigger_at 匹配
|
||||
- 进程级单飞锁 _worker_lock 防并发(单 worker, 保证互斥锁有效)
|
||||
- 失败处理: 仅 log + 通知中心写 system_event(category=ai, status=failed), 不阻塞
|
||||
- 参数分级: 后台预生成 timeout_s=180, max_tokens=template.max_tokens(2000)
|
||||
- 落库到 ai_reports + is_public=True + template_id + snapshot_date
|
||||
|
||||
#121 不接 sync_task_manager / bms_scheduler(留待 #123 调度仲裁层);
|
||||
本工单只实现 worker 函数本身, 暴露 schedule_due_templates / generate_for_template / warm_*
|
||||
供后续工单调用。
|
||||
|
||||
#123 将在 llm_scheduler.py 实现优先级队列, 与本 worker 解耦。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import datetime as _dt
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.database import SessionLocal
|
||||
from app.models.ai_report_template import AiReportTemplate
|
||||
from app.services import llm_client
|
||||
from app.services.ai_feed import build_messages
|
||||
from app.services.ai_report_service import save_report
|
||||
from app.services.system_event_service import record_system_event
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 时区(AGENTS.md 铁律: 时区换算走 Asia/Shanghai)
|
||||
_TZ_CN = _dt.timezone(_dt.timedelta(hours=8))
|
||||
|
||||
# 白天 8-23 才预热(与 cache_warmer.DAYTIME_HOURS 同步, 夜间无人访问省计算)
|
||||
_DAYTIME_HOURS = set(range(8, 24))
|
||||
|
||||
# 进程级单飞锁(防并发: 同步完成 + 启动 + 空闲期可能同时触发)
|
||||
_worker_lock = threading.Lock()
|
||||
|
||||
|
||||
def _now_cn() -> _dt.datetime:
|
||||
"""当前北京时间(AGENTS.md 铁律: 时区换算)。"""
|
||||
return _dt.datetime.now(_TZ_CN)
|
||||
|
||||
|
||||
def schedule_due_templates(db: Session) -> list[AiReportTemplate]:
|
||||
"""扫描 ai_report_templates, 返回当前时刻应当触发的模板列表。
|
||||
|
||||
判定逻辑:
|
||||
- frequency=sync-triggered: 不在此扫描(由 warm_on_sync_complete 主动触发)
|
||||
- frequency=daily + trigger_at=HH:MM: 当前北京时间 HH:MM 在 ±1 分钟内
|
||||
- frequency=weekly + trigger_at=wday:H:MM: 当前北京时间 wday:H:MM 在 ±1 分钟内
|
||||
- frequency=monthly + trigger_at=day:H:MM: 当前北京时间 day:H:MM 在 ±1 分钟内
|
||||
|
||||
返回按 priority 升序的模板列表(小先执行)。
|
||||
"""
|
||||
now = _now_cn()
|
||||
now_wday = now.isoweekday() # 1=周一, 7=周日
|
||||
now_day = now.day
|
||||
now_hm = now.strftime("%H:%M")
|
||||
|
||||
rows = db.query(AiReportTemplate).filter(
|
||||
AiReportTemplate.is_active.is_(True),
|
||||
AiReportTemplate.frequency.in_(["daily", "weekly", "monthly"]),
|
||||
).all()
|
||||
|
||||
due: list[AiReportTemplate] = []
|
||||
for t in rows:
|
||||
if not t.trigger_at:
|
||||
continue
|
||||
try:
|
||||
if t.frequency == "daily":
|
||||
# trigger_at = "HH:MM"
|
||||
if _hm_match(t.trigger_at, now_hm):
|
||||
due.append(t)
|
||||
elif t.frequency == "weekly":
|
||||
# trigger_at = "wday:HH:MM" (如 "1:09:00" = 周一9点)
|
||||
parts = t.trigger_at.split(":", 2)
|
||||
if len(parts) == 3:
|
||||
wday = int(parts[0])
|
||||
hm = f"{parts[1]}:{parts[2]}"
|
||||
if wday == now_wday and _hm_match(hm, now_hm):
|
||||
due.append(t)
|
||||
elif t.frequency == "monthly":
|
||||
# trigger_at = "day:HH:MM" (如 "1:09:00" = 每月1号9点)
|
||||
parts = t.trigger_at.split(":", 2)
|
||||
if len(parts) == 3:
|
||||
day = int(parts[0])
|
||||
hm = f"{parts[1]}:{parts[2]}"
|
||||
if day == now_day and _hm_match(hm, now_hm):
|
||||
due.append(t)
|
||||
except (ValueError, IndexError):
|
||||
logger.warning("ai_report_template(id=%s code=%s) trigger_at=%s 解析失败, 跳过",
|
||||
t.id, t.code, t.trigger_at)
|
||||
continue
|
||||
|
||||
# 按 priority 升序(小先执行)
|
||||
due.sort(key=lambda x: x.priority)
|
||||
return due
|
||||
|
||||
|
||||
def _hm_match(trigger_hm: str, now_hm: str) -> bool:
|
||||
"""HH:MM 匹配(±1 分钟容差)。
|
||||
|
||||
例: trigger="09:00", now="09:00" / "08:59" / "09:01" 都视为匹配
|
||||
"""
|
||||
try:
|
||||
th, tm = map(int, trigger_hm.split(":"))
|
||||
nh, nm = map(int, now_hm.split(":"))
|
||||
except (ValueError, AttributeError):
|
||||
return False
|
||||
trigger_min = th * 60 + tm
|
||||
now_min = nh * 60 + nm
|
||||
return abs(now_min - trigger_min) <= 1
|
||||
|
||||
|
||||
def generate_for_template(db: Session, template: AiReportTemplate) -> dict | None:
|
||||
"""为指定模板生成一份 AI 报告, 落库 is_public=True。
|
||||
|
||||
返回 {report_id, total_ms, llm_ms, total_tokens} 或 None(失败)。
|
||||
失败时: 仅 log + 写 system_event(category=ai, status=failed), 不抛异常。
|
||||
"""
|
||||
t0 = _dt.datetime.now()
|
||||
now_cn = _now_cn()
|
||||
snapshot_date = now_cn.strftime("%Y-%m-%d")
|
||||
|
||||
# 窗口变量(供 prompt 模板 {{date}} {{period}} 替换; 当前实现: period 直接用模板 period)
|
||||
period = template.period
|
||||
# question 用模板的 question_template(可含变量; 当前实现: 直接作为问句传入 build_messages)
|
||||
question = template.question_template
|
||||
|
||||
# 构造 messages(三层 Prompt: System/口径卡/Data+Question)
|
||||
try:
|
||||
messages = build_messages(
|
||||
db, period=period, date=None, question=question,
|
||||
start=None, end=None,
|
||||
window_label=f"预生成(template={template.code}, snapshot={snapshot_date})",
|
||||
window_source="template",
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 build_messages 失败不阻塞 worker
|
||||
logger.error("ai_report_worker build_messages failed (template=%s): %s", template.code, e)
|
||||
record_system_event(
|
||||
db, category="ai", action="pregen_build_messages_failed",
|
||||
title=f"AI 报告预生成 build_messages 失败: {template.name_cn}",
|
||||
summary=f"template={template.code} err={e!s}",
|
||||
status="failed", level="error",
|
||||
detail=json.dumps({"template_id": template.id, "code": template.code, "error": str(e)},
|
||||
ensure_ascii=False),
|
||||
)
|
||||
return None
|
||||
|
||||
# 调用 LLM(后台预生成: timeout_s=180, max_tokens=template.max_tokens)
|
||||
llm_model_str = template.llm_provider_code or "auto"
|
||||
try:
|
||||
llm_result = llm_client.chat_completion(
|
||||
messages,
|
||||
max_tokens=template.max_tokens,
|
||||
timeout_s=180.0, # 后台预生成允许长任务
|
||||
db=db,
|
||||
source=f"ai_pregen:{template.code}",
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 LLM 调用失败不阻塞 worker
|
||||
logger.error("ai_report_worker LLM call failed (template=%s): %s", template.code, e)
|
||||
record_system_event(
|
||||
db, category="ai", action="pregen_llm_failed",
|
||||
title=f"AI 报告预生成 LLM 失败: {template.name_cn}",
|
||||
summary=f"template={template.code} err={e!s}",
|
||||
status="failed", level="error",
|
||||
detail=json.dumps({"template_id": template.id, "code": template.code, "error": str(e)},
|
||||
ensure_ascii=False),
|
||||
)
|
||||
return None
|
||||
|
||||
# asyncio.run 包装(chat_completion 是 async, worker 同步入口)
|
||||
# 注意: 这里假设 worker 在非 asyncio 上下文调用(如 bms_scheduler to_thread);
|
||||
# 若在 asyncio 上下文, 调用方应直接 await chat_completion。
|
||||
if asyncio.iscoroutine(llm_result):
|
||||
try:
|
||||
llm_result = asyncio.run(llm_result)
|
||||
except RuntimeError as e:
|
||||
# 已在 event loop 中: 创建新 loop 跑(单 worker 串行, 不会有并发问题)
|
||||
logger.warning("ai_report_worker: already in event loop, create new loop: %s", e)
|
||||
llm_result = asyncio.new_event_loop().run_until_complete(llm_result)
|
||||
|
||||
total_ms = int((_dt.datetime.now() - t0).total_seconds() * 1000)
|
||||
usage = llm_result.usage if hasattr(llm_result, "usage") else None
|
||||
answer_md = str(llm_result)
|
||||
|
||||
# 落库(is_public=True, template_id, snapshot_date)
|
||||
report = save_report(
|
||||
db,
|
||||
task_id=None, # 预生成无 ai_task_manager 任务 id
|
||||
user_id=0, # 系统生成, user_id=0; 通过 is_public=True 让所有人可见
|
||||
question=question,
|
||||
period=period,
|
||||
date=snapshot_date,
|
||||
start=None, end=None,
|
||||
window_source="template",
|
||||
window_label=f"预生成({template.code})",
|
||||
answer_md=answer_md,
|
||||
llm_model=llm_model_str,
|
||||
llm_ms=total_ms, # 简化: 全程算 llm_ms(后续可细分 build_ms + llm_ms)
|
||||
total_ms=total_ms,
|
||||
tokens=usage,
|
||||
template_id=template.id,
|
||||
snapshot_date=snapshot_date,
|
||||
snapshot_period=template.period,
|
||||
is_public=True,
|
||||
)
|
||||
|
||||
if report is None:
|
||||
logger.error("ai_report_worker: save_report returned None (template=%s)", template.code)
|
||||
record_system_event(
|
||||
db, category="ai", action="pregen_save_failed",
|
||||
title=f"AI 报告预生成落库失败: {template.name_cn}",
|
||||
summary=f"template={template.code} save_report=None",
|
||||
status="failed", level="error",
|
||||
detail=json.dumps({"template_id": template.id, "code": template.code},
|
||||
ensure_ascii=False),
|
||||
)
|
||||
return None
|
||||
|
||||
logger.info("ai_report_worker: pregen success template=%s report_id=%s total_ms=%s",
|
||||
template.code, report.id, total_ms)
|
||||
record_system_event(
|
||||
db, category="ai", action="pregen_success",
|
||||
title=f"AI 报告预生成完成: {template.name_cn}",
|
||||
summary=f"template={template.code} report_id={report.id} total_ms={total_ms}ms",
|
||||
status="success", level="info",
|
||||
detail=json.dumps({"template_id": template.id, "report_id": report.id, "total_ms": total_ms},
|
||||
ensure_ascii=False),
|
||||
)
|
||||
return {
|
||||
"report_id": report.id,
|
||||
"total_ms": total_ms,
|
||||
"llm_ms": total_ms,
|
||||
"total_tokens": usage.get("total_tokens") if usage else None,
|
||||
}
|
||||
|
||||
|
||||
def warm_on_sync_complete() -> None:
|
||||
"""同步完成触发入口: 扫描 sync-triggered 模板, 依次生成。
|
||||
|
||||
由 sync_task_manager._run_background finally 调用(待 #123 接入)。
|
||||
失败仅 log + 通知中心, 不抛异常(不影响同步流程)。
|
||||
"""
|
||||
if not _worker_lock.acquire(blocking=False):
|
||||
logger.info("ai_report_worker warm_on_sync_complete: 已有 worker 在跑, 跳过")
|
||||
return
|
||||
db = SessionLocal()
|
||||
try:
|
||||
rows = db.query(AiReportTemplate).filter(
|
||||
AiReportTemplate.is_active.is_(True),
|
||||
AiReportTemplate.frequency == "sync-triggered",
|
||||
).order_by(AiReportTemplate.priority).all()
|
||||
if not rows:
|
||||
return
|
||||
logger.info("ai_report_worker warm_on_sync_complete: %s 个 sync-triggered 模板待生成", len(rows))
|
||||
for t in rows:
|
||||
try:
|
||||
generate_for_template(db, t)
|
||||
except Exception as e: # noqa: BLE001 单个模板失败不影响其他
|
||||
logger.error("ai_report_worker: template=%s 生成失败: %s", t.code, e)
|
||||
except Exception as e: # noqa: BLE001 worker 失败不影响业务
|
||||
logger.error("ai_report_worker warm_on_sync_complete 异常(不影响业务): %s", e)
|
||||
finally:
|
||||
db.close()
|
||||
_worker_lock.release()
|
||||
|
||||
|
||||
def warm_on_startup(force: bool = False) -> None:
|
||||
"""服务启动后预热: 触发 active 模板(无 trigger_at 匹配, 全部 active)。
|
||||
|
||||
force=True 时无视白天限制(启动时强制); 否则只在白天 8-23 触发。
|
||||
"""
|
||||
if not force and _now_cn().hour not in _DAYTIME_HOURS:
|
||||
return
|
||||
if not _worker_lock.acquire(blocking=False):
|
||||
logger.info("ai_report_worker warm_on_startup: 已有 worker 在跑, 跳过")
|
||||
return
|
||||
db = SessionLocal()
|
||||
try:
|
||||
rows = db.query(AiReportTemplate).filter(
|
||||
AiReportTemplate.is_active.is_(True),
|
||||
).order_by(AiReportTemplate.priority).all()
|
||||
if not rows:
|
||||
return
|
||||
logger.info("ai_report_worker warm_on_startup(force=%s): %s 个 active 模板待生成",
|
||||
force, len(rows))
|
||||
for t in rows:
|
||||
try:
|
||||
generate_for_template(db, t)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.error("ai_report_worker: template=%s 生成失败: %s", t.code, e)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.error("ai_report_worker warm_on_startup 异常(不影响业务): %s", e)
|
||||
finally:
|
||||
db.close()
|
||||
_worker_lock.release()
|
||||
|
||||
|
||||
def warm_idle() -> None:
|
||||
"""空闲期每分钟扫描: 触发当前时刻匹配的 daily/weekly/monthly 模板。
|
||||
|
||||
由 bms_scheduler._run_loop 每分钟调用(待 #123 接入)。
|
||||
白天 8-23 才扫描, 夜间无人访问省计算。
|
||||
"""
|
||||
if _now_cn().hour not in _DAYTIME_HOURS:
|
||||
return
|
||||
if not _worker_lock.acquire(blocking=False):
|
||||
logger.info("ai_report_worker warm_idle: 已有 worker 在跑, 跳过")
|
||||
return
|
||||
db = SessionLocal()
|
||||
try:
|
||||
due = schedule_due_templates(db)
|
||||
if not due:
|
||||
return
|
||||
logger.info("ai_report_worker warm_idle: %s 个模板到期", len(due))
|
||||
for t in due:
|
||||
try:
|
||||
generate_for_template(db, t)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.error("ai_report_worker: template=%s 生成失败: %s", t.code, e)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.error("ai_report_worker warm_idle 异常(不影响业务): %s", e)
|
||||
finally:
|
||||
db.close()
|
||||
_worker_lock.release()
|
||||
Reference in New Issue
Block a user