feat: schedule periodic reports asynchronously

ai_knowledge_base_v2/apps/backend/app/services/entitlement_service.py

@@ -41,7 +41,7 @@ class EntitlementView:
 class EntitlementService:
     @staticmethod
     def list_plans(db: Session, *, include_disabled: bool = False) -> list[EntitlementPlan]:
-        query = select(EntitlementPlan)
+        query = select(EntitlementPlan).where(EntitlementPlan.plan_type != "teacher")
         if not include_disabled:
             query = query.where(EntitlementPlan.status == 1)
         return list(db.scalars(query.order_by(EntitlementPlan.sort_order.asc(), EntitlementPlan.id.asc())))
@@ -50,7 +50,11 @@ class EntitlementService:
     def default_plan(db: Session) -> EntitlementPlan | None:
         plan = db.scalar(
             select(EntitlementPlan)
-            .where(EntitlementPlan.plan_type == DEFAULT_PLAN_TYPE, EntitlementPlan.status == 1)
+            .where(
+                EntitlementPlan.plan_type == DEFAULT_PLAN_TYPE,
+                EntitlementPlan.plan_type != "teacher",
+                EntitlementPlan.status == 1,
+            )
             .order_by(EntitlementPlan.sort_order.asc(), EntitlementPlan.id.asc())
             .limit(1)
         )
@@ -58,7 +62,7 @@ class EntitlementService:
             return plan
         return db.scalar(
             select(EntitlementPlan)
-            .where(EntitlementPlan.status == 1)
+            .where(EntitlementPlan.status == 1, EntitlementPlan.plan_type != "teacher")
             .order_by(EntitlementPlan.sort_order.asc(), EntitlementPlan.id.asc())
             .limit(1)
         )
@@ -73,6 +77,7 @@ class EntitlementService:
                 UserEntitlement.user_id == user.id,
                 UserEntitlement.status == "active",
                 EntitlementPlan.status == 1,
+                EntitlementPlan.plan_type != "teacher",
             )
             .where((UserEntitlement.effective_at.is_(None)) | (UserEntitlement.effective_at <= now))
             .where((UserEntitlement.expired_at.is_(None)) | (UserEntitlement.expired_at >= now))
@@ -113,7 +118,7 @@ class EntitlementService:
         remark: str | None = None,
     ) -> UserEntitlement:
         plan = db.get(EntitlementPlan, plan_id)
-        if plan is None or plan.status != 1:
+        if plan is None or plan.status != 1 or plan.plan_type == "teacher":
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="权益版本不存在或已停用")
         now = _now()
         if effective_at is not None:

ai_knowledge_base_v2/apps/backend/app/services/periodic_report_service.py

@@ -2,12 +2,14 @@ from __future__ import annotations
 
 import json
 from datetime import UTC, datetime, timedelta
-from typing import Literal
+from typing import Iterable, Literal
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.models.chat import TopicSession
+from app.core.config import get_settings
 from app.models.growth import PeriodicReport, TopicSummary, UserGrowthProfile
 from app.models.user import User
 from app.services.model_service import ModelClientService
@@ -24,29 +26,98 @@ REPORT_TYPE_LABELS = {
 class PeriodicReportService:
     @staticmethod
     def default_period(report_type: ReportType, now: datetime | None = None) -> tuple[datetime, datetime]:
-        current = (now or datetime.now(UTC)).replace(tzinfo=None)
-        if report_type == "weekly":
-            end = current
-            start = end - timedelta(days=7)
-        elif report_type == "monthly":
-            end = current
-            start = end - timedelta(days=30)
-        else:
-            end = current
-            start = end - timedelta(days=150)
+        current = now or datetime.now(UTC)
+        if report_type in {"weekly", "monthly"}:
+            return calendar_period(
+                report_type,
+                current,
+                get_settings().periodic_report_timezone,
+                respect_schedule_delay=False,
+            )
+        end = current.astimezone(UTC).replace(tzinfo=None) if current.tzinfo else current
+        start = end - timedelta(days=150)
         return start.replace(microsecond=0), end.replace(microsecond=0)
 
     @staticmethod
-    def list_user_reports(db: Session, *, user_id: int, limit: int = 20) -> list[PeriodicReport]:
+    def list_user_reports(
+        db: Session,
+        *,
+        user_id: int,
+        limit: int = 20,
+        statuses: Iterable[str] | None = None,
+    ) -> list[PeriodicReport]:
+        query = select(PeriodicReport).where(PeriodicReport.user_id == user_id)
+        if statuses is not None:
+            query = query.where(PeriodicReport.status.in_(tuple(statuses)))
         return list(
             db.scalars(
-                select(PeriodicReport)
-                .where(PeriodicReport.user_id == user_id)
+                query
                 .order_by(PeriodicReport.period_end.desc(), PeriodicReport.id.desc())
                 .limit(limit)
             )
         )
 
+    @staticmethod
+    def enqueue_for_user(
+        db: Session,
+        *,
+        user: User,
+        report_type: ReportType,
+        period_start: datetime | None = None,
+        period_end: datetime | None = None,
+        generated_by: str = "manual",
+        force: bool = True,
+    ) -> PeriodicReport:
+        period_start, period_end = _resolve_period(report_type, period_start, period_end)
+        report = _find_report(
+            db,
+            user_id=user.id,
+            report_type=report_type,
+            period_start=period_start,
+            period_end=period_end,
+        )
+        if report is None:
+            report = _new_report(
+                user=user,
+                report_type=report_type,
+                period_start=period_start,
+                period_end=period_end,
+            )
+            try:
+                # 同周期并发入队由唯一约束兜底，不会产生重复报告任务。
+                with db.begin_nested():
+                    db.add(report)
+                    db.flush()
+            except IntegrityError:
+                report = _find_report(
+                    db,
+                    user_id=user.id,
+                    report_type=report_type,
+                    period_start=period_start,
+                    period_end=period_end,
+                )
+                if report is None:
+                    raise
+        elif not force:
+            return report
+
+        if report.status == "running":
+            return report
+        report.title = _report_title(report_type, period_start, period_end)
+        report.status = "pending"
+        report.error_message = None
+        report.generated_by = generated_by
+        report.attempt_count = 0
+        report.max_attempts = max(1, get_settings().periodic_report_max_attempts)
+        report.next_run_at = _now()
+        report.locked_at = None
+        report.locked_by = None
+        report.last_started_at = None
+        report.finished_at = None
+        db.add(report)
+        db.flush()
+        return report
+
     @staticmethod
     def generate_for_user(
         db: Session,
@@ -57,25 +128,44 @@ class PeriodicReportService:
         period_end: datetime | None = None,
         generated_by: str = "manual",
     ) -> PeriodicReport:
+        period_start, period_end = _resolve_period(report_type, period_start, period_end)
+        report = _find_report(
+            db,
+            user_id=user.id,
+            report_type=report_type,
+            period_start=period_start,
+            period_end=period_end,
+        ) or _new_report(
+            user=user,
+            report_type=report_type,
+            period_start=period_start,
+            period_end=period_end,
+        )
+        report.generated_by = generated_by
+        PeriodicReportService.generate_existing(db, report=report, user=user)
+        report.finished_at = _now()
+        report.next_run_at = None
+        report.locked_at = None
+        report.locked_by = None
+        db.add(report)
+        db.commit()
+        db.refresh(report)
+        return report
+
+    @staticmethod
+    def generate_existing(db: Session, *, report: PeriodicReport, user: User) -> PeriodicReport:
+        report_type = report.report_type
         if report_type not in REPORT_TYPE_LABELS:
             raise ValueError("不支持的报告类型")
-        if period_start is None or period_end is None:
-            default_start, default_end = PeriodicReportService.default_period(report_type)
-            period_start = period_start or default_start
-            period_end = period_end or default_end
-        period_start = period_start.replace(tzinfo=None, microsecond=0)
-        period_end = period_end.replace(tzinfo=None, microsecond=0)
-        if period_start >= period_end:
-            raise ValueError("报告开始时间必须早于结束时间")
-
-        report = _get_or_create_report(db, user=user, report_type=report_type, period_start=period_start, period_end=period_end)
+        period_start = report.period_start
+        period_end = report.period_end
+        report.title = _report_title(report_type, period_start, period_end)
         summaries = _period_summaries(db, user_id=user.id, period_start=period_start, period_end=period_end)
         profile = db.scalar(select(UserGrowthProfile).where(UserGrowthProfile.user_id == user.id))
         topic_ids = sorted({int(item.topic_session_id) for item in summaries})
         summary_ids = [int(item.id) for item in summaries]
         report.source_topic_ids = json.dumps(topic_ids, ensure_ascii=False)
         report.source_summary_ids = json.dumps(summary_ids, ensure_ascii=False)
-        report.generated_by = generated_by
         report.generated_at = _now()
 
         if not summaries:
@@ -83,8 +173,6 @@ class PeriodicReportService:
             report.error_message = None
             report.content = _empty_report_content(report_type=report_type, period_start=period_start, period_end=period_end)
             db.add(report)
-            db.commit()
-            db.refresh(report)
             return report
 
         try:
@@ -99,8 +187,6 @@ class PeriodicReportService:
             report.error_message = str(exc)[:2000]
             report.content = _fallback_report(summaries)
         db.add(report)
-        db.commit()
-        db.refresh(report)
         return report
 
 
@@ -110,8 +196,8 @@ def periodic_report_dict(report: PeriodicReport) -> dict:
         "userId": report.user_id,
         "reportType": report.report_type,
         "reportTypeLabel": REPORT_TYPE_LABELS.get(report.report_type, report.report_type),
-        "periodStart": report.period_start,
-        "periodEnd": report.period_end,
+        "periodStart": _local_datetime(report.period_start),
+        "periodEnd": _local_datetime(report.period_end),
         "title": report.title,
         "content": report.content,
         "sourceSummaryIds": _parse_json_list(report.source_summary_ids),
@@ -120,40 +206,119 @@ def periodic_report_dict(report: PeriodicReport) -> dict:
         "status": report.status,
         "errorMessage": report.error_message,
         "generatedBy": report.generated_by,
+        "attemptCount": report.attempt_count,
+        "maxAttempts": report.max_attempts,
+        "nextRunAt": report.next_run_at,
+        "lastStartedAt": report.last_started_at,
+        "finishedAt": report.finished_at,
         "generatedAt": report.generated_at,
         "createdAt": report.created_at,
         "updatedAt": report.updated_at,
     }
 
 
-def _get_or_create_report(
+def calendar_period(
+    report_type: str,
+    now_utc: datetime,
+    timezone_name: str = "Asia/Shanghai",
+    *,
+    respect_schedule_delay: bool,
+) -> tuple[datetime, datetime]:
+    """计算最近一个完整自然周或自然月，返回 UTC 无时区时间。"""
+
+    try:
+        timezone = ZoneInfo(timezone_name)
+    except ZoneInfoNotFoundError:
+        timezone = ZoneInfo("Asia/Shanghai")
+    aware_utc = now_utc.replace(tzinfo=UTC) if now_utc.tzinfo is None else now_utc.astimezone(UTC)
+    local_now = aware_utc.astimezone(timezone)
+
+    if report_type == "weekly":
+        period_end_local = (local_now - timedelta(days=local_now.weekday())).replace(
+            hour=0,
+            minute=0,
+            second=0,
+            microsecond=0,
+        )
+        if respect_schedule_delay and local_now < period_end_local + timedelta(hours=2):
+            period_end_local -= timedelta(days=7)
+        period_start_local = period_end_local - timedelta(days=7)
+    elif report_type == "monthly":
+        period_end_local = local_now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+        if respect_schedule_delay and local_now < period_end_local + timedelta(hours=3):
+            period_end_local = _previous_month_start(period_end_local)
+        period_start_local = _previous_month_start(period_end_local)
+    else:
+        raise ValueError("不支持的自然周期报告类型")
+
+    return (
+        period_start_local.astimezone(UTC).replace(tzinfo=None),
+        period_end_local.astimezone(UTC).replace(tzinfo=None),
+    )
+
+
+def _find_report(
     db: Session,
+    *,
+    user_id: int,
+    report_type: ReportType,
+    period_start: datetime,
+    period_end: datetime,
+) -> PeriodicReport | None:
+    return db.scalar(
+        select(PeriodicReport).where(
+            PeriodicReport.user_id == user_id,
+            PeriodicReport.report_type == report_type,
+            PeriodicReport.period_start == period_start,
+            PeriodicReport.period_end == period_end,
+        )
+    )
+
+
+def _new_report(
     *,
     user: User,
     report_type: ReportType,
     period_start: datetime,
     period_end: datetime,
 ) -> PeriodicReport:
-    report = db.scalar(
-        select(PeriodicReport).where(
-            PeriodicReport.user_id == user.id,
-            PeriodicReport.report_type == report_type,
-            PeriodicReport.period_start == period_start,
-            PeriodicReport.period_end == period_end,
-        )
+    return PeriodicReport(
+        user_id=user.id,
+        report_type=report_type,
+        period_start=period_start,
+        period_end=period_end,
+        title=_report_title(report_type, period_start, period_end),
     )
-    title = f"{REPORT_TYPE_LABELS[report_type]}（{period_start:%Y-%m-%d} 至 {period_end:%Y-%m-%d}）"
-    if report is None:
-        report = PeriodicReport(
-            user_id=user.id,
-            report_type=report_type,
-            period_start=period_start,
-            period_end=period_end,
-            title=title,
-        )
-    else:
-        report.title = title
-    return report
+
+
+def _resolve_period(
+    report_type: ReportType,
+    period_start: datetime | None,
+    period_end: datetime | None,
+) -> tuple[datetime, datetime]:
+    if report_type not in REPORT_TYPE_LABELS:
+        raise ValueError("不支持的报告类型")
+    if period_start is None or period_end is None:
+        default_start, default_end = PeriodicReportService.default_period(report_type)
+        period_start = period_start or default_start
+        period_end = period_end or default_end
+    start = period_start.replace(tzinfo=None, microsecond=0)
+    end = period_end.replace(tzinfo=None, microsecond=0)
+    if start >= end:
+        raise ValueError("报告开始时间必须早于结束时间")
+    return start, end
+
+
+def _report_title(report_type: str, period_start: datetime, period_end: datetime) -> str:
+    local_start = _local_datetime(period_start)
+    local_end = _local_datetime(period_end)
+    return f"{REPORT_TYPE_LABELS[report_type]}（{local_start:%Y-%m-%d} 至 {local_end:%Y-%m-%d}）"
+
+
+def _previous_month_start(value: datetime) -> datetime:
+    if value.month == 1:
+        return value.replace(year=value.year - 1, month=12, day=1)
+    return value.replace(month=value.month - 1, day=1)
 
 
 def _period_summaries(db: Session, *, user_id: int, period_start: datetime, period_end: datetime) -> list[TopicSummary]:
@@ -164,7 +329,7 @@ def _period_summaries(db: Session, *, user_id: int, period_start: datetime, peri
                 TopicSummary.user_id == user_id,
                 TopicSummary.status == "success",
                 TopicSummary.generated_at >= period_start,
-                TopicSummary.generated_at <= period_end,
+                TopicSummary.generated_at < period_end,
             )
             .order_by(TopicSummary.generated_at.asc(), TopicSummary.id.asc())
             .limit(200)
@@ -181,6 +346,8 @@ def _report_prompt(
     summaries: list[TopicSummary],
     profile: UserGrowthProfile | None,
 ) -> str:
+    local_start = _local_datetime(period_start)
+    local_end = _local_datetime(period_end)
     summary_text = "\n\n".join(
         (
             f"主题摘要 {index}\n"
@@ -196,7 +363,7 @@ def _report_prompt(
     )
     return (
         f"请为学员“{user.name or user.nickname or user.phone}”生成一份{REPORT_TYPE_LABELS[report_type]}。\n"
-        f"周期：{period_start:%Y-%m-%d %H:%M} 至 {period_end:%Y-%m-%d %H:%M}。\n\n"
+        f"周期：{local_start:%Y-%m-%d %H:%M} 至 {local_end:%Y-%m-%d %H:%M}。\n\n"
         "产品定位：这是大本营千问千答的实修陪伴报告，不写成医疗诊断、心理咨询结论或营销文。\n"
         "表达方向：回到当下、回到自身、觉察情绪和身体感受，如实释放；建议适度，不要给过多术层面的复杂方案。\n"
         "请使用 Markdown 输出，结构包含：本周期主要议题、做过或被建议的功课、反复出现的情绪/身体模式、已有变化、下一步观察方向、可以带给老师确认的问题。\n"
@@ -218,9 +385,11 @@ def _fallback_report(summaries: list[TopicSummary]) -> str:
 
 
 def _empty_report_content(*, report_type: ReportType, period_start: datetime, period_end: datetime) -> str:
+    local_start = _local_datetime(period_start)
+    local_end = _local_datetime(period_end)
     return (
         f"## {REPORT_TYPE_LABELS[report_type]}\n\n"
-        f"周期：{period_start:%Y-%m-%d} 至 {period_end:%Y-%m-%d}\n\n"
+        f"周期：{local_start:%Y-%m-%d} 至 {local_end:%Y-%m-%d}\n\n"
         "本周期还没有可用于生成报告的主题沉淀。可以在完成一次主题对话后，先点击“沉淀本主题”，再生成报告。"
     )
 
@@ -239,3 +408,12 @@ def _parse_json_list(raw: str | None) -> list[int]:
 
 def _now() -> datetime:
     return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _local_datetime(value: datetime) -> datetime:
+    try:
+        timezone = ZoneInfo(get_settings().periodic_report_timezone)
+    except ZoneInfoNotFoundError:
+        timezone = ZoneInfo("Asia/Shanghai")
+    aware_utc = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+    return aware_utc.astimezone(timezone).replace(tzinfo=None)

ai_knowledge_base_v2/apps/backend/app/services/periodic_report_worker.py

@@ -0,0 +1,321 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import socket
+import uuid
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import exists, or_, select
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.database import SessionLocal
+from app.models.ai_config import SystemConfig
+from app.models.growth import PeriodicReport, TopicSummary
+from app.models.user import User
+from app.services.entitlement_service import EntitlementService
+from app.services.periodic_report_service import PeriodicReportService, calendar_period
+from app.services.redis_client import get_sync_redis_client
+
+
+logger = logging.getLogger(__name__)
+WORKER_LOCK_KEY = "periodic-report:worker:lock"
+WORKER_LOCK_TTL_SECONDS = 900
+SCHEDULE_STATE_KEYS = {
+    "weekly": "periodic_report_weekly_last_period_end",
+    "monthly": "periodic_report_monthly_last_period_end",
+}
+
+
+class PeriodicReportWorker:
+    """数据库持久化报告队列。
+
+    Redis 只用于限制多进程并发；任务状态、重试和调度进度均以 MySQL 为准，
+    因此 Redis 暂时不可用或服务重启时不会丢任务。
+    """
+
+    @classmethod
+    async def run_forever(cls) -> None:
+        settings = get_settings()
+        if not settings.periodic_report_worker_enabled:
+            logger.info("periodic report worker disabled")
+            return
+        worker_id = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
+        poll_seconds = max(1, settings.periodic_report_poll_seconds)
+        while True:
+            try:
+                processed = await asyncio.to_thread(cls.run_once, worker_id)
+            except Exception:
+                processed = False
+                logger.exception("periodic report worker iteration failed")
+            await asyncio.sleep(0 if processed else poll_seconds)
+
+    @classmethod
+    def run_once(cls, worker_id: str) -> bool:
+        redis, acquired = _acquire_worker_lock(worker_id)
+        if not acquired:
+            return False
+        try:
+            now = _now()
+            with SessionLocal() as db:
+                cls.recover_stale_jobs(db, now=now)
+                cls.enqueue_due_schedules(db, now_utc=now)
+                db.commit()
+            with SessionLocal() as db:
+                report_id = cls.claim_next(db, worker_id=worker_id, now=now)
+            if report_id is None:
+                return False
+            with SessionLocal() as db:
+                cls.execute_claimed(db, report_id=report_id, worker_id=worker_id)
+            return True
+        finally:
+            _release_worker_lock(redis, worker_id)
+
+    @staticmethod
+    def claim_next(db: Session, *, worker_id: str, now: datetime | None = None) -> int | None:
+        current = now or _now()
+        report = db.scalar(
+            select(PeriodicReport)
+            .where(
+                PeriodicReport.status == "pending",
+                PeriodicReport.attempt_count < PeriodicReport.max_attempts,
+                or_(PeriodicReport.next_run_at.is_(None), PeriodicReport.next_run_at <= current),
+            )
+            .order_by(PeriodicReport.next_run_at.asc(), PeriodicReport.id.asc())
+            .with_for_update(skip_locked=True)
+            .limit(1)
+        )
+        if report is None:
+            db.rollback()
+            return None
+        report.status = "running"
+        report.attempt_count += 1
+        report.locked_at = current
+        report.locked_by = worker_id
+        report.last_started_at = current
+        report.finished_at = None
+        db.add(report)
+        db.commit()
+        return report.id
+
+    @staticmethod
+    def execute_claimed(db: Session, *, report_id: int, worker_id: str) -> PeriodicReport | None:
+        report = db.get(PeriodicReport, report_id)
+        if report is None or report.status != "running" or report.locked_by != worker_id:
+            return report
+        user = db.get(User, report.user_id)
+        if user is None or user.is_deleted or user.status != 1:
+            _finish_permanent_failure(report, "用户不存在或已停用")
+            db.commit()
+            return report
+        if report.generated_by.startswith("schedule:"):
+            entitlement = EntitlementService.active_entitlement(db, user)
+            if not entitlement.enable_periodic_reports:
+                _finish_permanent_failure(report, "用户当前权益未开启周期报告")
+                db.commit()
+                return report
+
+        try:
+            PeriodicReportService.generate_existing(db, report=report, user=user)
+        except Exception as exc:
+            report.status = "failed"
+            report.error_message = str(exc)[:2000]
+
+        now = _now()
+        if report.status == "failed" and report.attempt_count < report.max_attempts:
+            retry_seconds = min(300, 30 * (2 ** max(0, report.attempt_count - 1)))
+            report.status = "pending"
+            report.next_run_at = now + timedelta(seconds=retry_seconds)
+            report.finished_at = None
+        else:
+            report.next_run_at = None
+            report.finished_at = now
+        report.locked_at = None
+        report.locked_by = None
+        db.add(report)
+        db.commit()
+        db.refresh(report)
+        return report
+
+    @staticmethod
+    def recover_stale_jobs(db: Session, *, now: datetime | None = None) -> int:
+        current = now or _now()
+        stale_before = current - timedelta(minutes=max(5, get_settings().periodic_report_stale_minutes))
+        reports = list(
+            db.scalars(
+                select(PeriodicReport)
+                .where(
+                    PeriodicReport.status == "running",
+                    PeriodicReport.locked_at.is_not(None),
+                    PeriodicReport.locked_at < stale_before,
+                )
+                .order_by(PeriodicReport.id.asc())
+                .limit(100)
+                .with_for_update(skip_locked=True)
+            )
+        )
+        for report in reports:
+            if report.attempt_count >= report.max_attempts:
+                report.status = "failed"
+                report.finished_at = current
+                report.next_run_at = None
+            else:
+                report.status = "pending"
+                report.next_run_at = current
+                report.finished_at = None
+            report.error_message = _append_error(report.error_message, "任务执行进程中断，系统已自动恢复")
+            report.locked_at = None
+            report.locked_by = None
+            db.add(report)
+        return len(reports)
+
+    @staticmethod
+    def enqueue_due_schedules(db: Session, *, now_utc: datetime | None = None) -> dict[str, int]:
+        settings = get_settings()
+        current = now_utc or _now()
+        result: dict[str, int] = {}
+        schedule_flags = {
+            "weekly": settings.periodic_report_weekly_enabled,
+            "monthly": settings.periodic_report_monthly_enabled,
+        }
+        for report_type, enabled in schedule_flags.items():
+            if not enabled:
+                continue
+            period = scheduled_period(report_type, current, settings.periodic_report_timezone)
+            if period is None:
+                continue
+            period_start, period_end = period
+            state_key = SCHEDULE_STATE_KEYS[report_type]
+            state = db.scalar(select(SystemConfig).where(SystemConfig.config_key == state_key))
+            period_marker = period_end.isoformat()
+            if state is not None and state.config_value == period_marker:
+                continue
+            enqueued = _enqueue_scheduled_users(
+                db,
+                report_type=report_type,
+                period_start=period_start,
+                period_end=period_end,
+                now=current,
+            )
+            if state is None:
+                state = SystemConfig(
+                    config_key=state_key,
+                    config_value=period_marker,
+                    description=f"{report_type} 周期报告最近一次完成入队的周期结束时间",
+                )
+            else:
+                state.config_value = period_marker
+            db.add(state)
+            result[report_type] = enqueued
+        return result
+
+
+def scheduled_period(
+    report_type: str,
+    now_utc: datetime,
+    timezone_name: str = "Asia/Shanghai",
+) -> tuple[datetime, datetime] | None:
+    """返回最近一个已经到达生成时间的完整自然周或自然月。"""
+    if report_type not in {"weekly", "monthly"}:
+        return None
+    return calendar_period(
+        report_type,
+        now_utc,
+        timezone_name,
+        respect_schedule_delay=True,
+    )
+
+
+def _enqueue_scheduled_users(
+    db: Session,
+    *,
+    report_type: str,
+    period_start: datetime,
+    period_end: datetime,
+    now: datetime,
+) -> int:
+    has_summary = exists(
+        select(TopicSummary.id).where(
+            TopicSummary.user_id == User.id,
+            TopicSummary.status == "success",
+            TopicSummary.generated_at >= period_start,
+            TopicSummary.generated_at < period_end,
+        )
+    )
+    users = list(
+        db.scalars(
+            select(User)
+            .where(
+                User.is_deleted == 0,
+                User.status == 1,
+                or_(User.effective_at.is_(None), User.effective_at <= now),
+                or_(User.expired_at.is_(None), User.expired_at >= now),
+                has_summary,
+            )
+            .order_by(User.id.asc())
+        )
+    )
+    enqueued = 0
+    for user in users:
+        entitlement = EntitlementService.active_entitlement(db, user)
+        if not entitlement.enable_periodic_reports:
+            continue
+        report = PeriodicReportService.enqueue_for_user(
+            db,
+            user=user,
+            report_type=report_type,  # type: ignore[arg-type]
+            period_start=period_start,
+            period_end=period_end,
+            generated_by=f"schedule:{report_type}",
+            force=False,
+        )
+        if report.status == "pending":
+            enqueued += 1
+    return enqueued
+
+
+def _finish_permanent_failure(report: PeriodicReport, message: str) -> None:
+    report.status = "failed"
+    report.error_message = message
+    report.next_run_at = None
+    report.locked_at = None
+    report.locked_by = None
+    report.finished_at = _now()
+
+
+def _append_error(current: str | None, message: str) -> str:
+    if not current:
+        return message
+    return f"{current}\n{message}"[-2000:]
+
+
+def _acquire_worker_lock(worker_id: str):
+    redis = get_sync_redis_client()
+    if redis is None:
+        return None, True
+    try:
+        return redis, bool(redis.set(WORKER_LOCK_KEY, worker_id, nx=True, ex=WORKER_LOCK_TTL_SECONDS))
+    except Exception:
+        logger.warning("redis unavailable for periodic report worker lock; falling back to database claim", exc_info=True)
+        return None, True
+
+
+def _release_worker_lock(redis, worker_id: str) -> None:
+    if redis is None:
+        return
+    try:
+        redis.eval(
+            "if redis.call('get', KEYS[1]) == ARGV[1] then "
+            "return redis.call('del', KEYS[1]) else return 0 end",
+            1,
+            WORKER_LOCK_KEY,
+            worker_id,
+        )
+    except Exception:
+        logger.warning("failed to release periodic report worker lock", exc_info=True)
+
+
+def _now() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)