From 0008903e8da0f339146dcbb2226ce21d30a8648f Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Fri, 31 Jul 2026 17:11:21 +0800 Subject: [PATCH] feat: schedule periodic reports asynchronously --- ai_knowledge_base_v2/.env.example | 7 + .../apps/admin-web/src/App.vue | 41 ++- .../apps/admin-web/src/styles.css | 9 + .../apps/admin-web/src/types/api.ts | 7 +- .../apps/backend/.env.example | 9 + .../0021_periodic_report_async_jobs.py | 62 ++++ .../apps/backend/app/api/admin_users.py | 7 +- .../apps/backend/app/api/user.py | 7 +- .../apps/backend/app/core/config.py | 7 + ai_knowledge_base_v2/apps/backend/app/main.py | 14 +- .../apps/backend/app/models/growth.py | 9 + .../apps/backend/app/schemas/admin.py | 2 +- .../app/services/entitlement_service.py | 13 +- .../app/services/periodic_report_service.py | 288 +++++++++++++--- .../app/services/periodic_report_worker.py | 321 ++++++++++++++++++ .../backend/tests/test_entitlements_topics.py | 22 ++ .../backend/tests/test_periodic_reports.py | 177 +++++++++- .../deployment/DEPLOYMENT_GUIDE.md | 24 ++ ai_knowledge_base_v2/docker-compose.dev.yml | 4 + ai_knowledge_base_v2/docker-compose.prod.yml | 7 + .../docs/qianwen_product_todo.md | 77 +---- 21 files changed, 969 insertions(+), 145 deletions(-) create mode 100644 ai_knowledge_base_v2/apps/backend/alembic/versions/0021_periodic_report_async_jobs.py create mode 100644 ai_knowledge_base_v2/apps/backend/app/services/periodic_report_worker.py diff --git a/ai_knowledge_base_v2/.env.example b/ai_knowledge_base_v2/.env.example index 9029515..a450ca0 100644 --- a/ai_knowledge_base_v2/.env.example +++ b/ai_knowledge_base_v2/.env.example @@ -21,6 +21,13 @@ CONFIG_ENCRYPTION_KEY=replace-with-generated-fernet-key BOOTSTRAP_ADMIN_USERNAME=admin BOOTSTRAP_ADMIN_PASSWORD=replace-with-strong-password BOOTSTRAP_ADMIN_NAME=系统管理员 +PERIODIC_REPORT_WORKER_ENABLED=true +PERIODIC_REPORT_POLL_SECONDS=5 +PERIODIC_REPORT_STALE_MINUTES=30 +PERIODIC_REPORT_MAX_ATTEMPTS=3 +PERIODIC_REPORT_WEEKLY_ENABLED=true +PERIODIC_REPORT_MONTHLY_ENABLED=true +PERIODIC_REPORT_TIMEZONE=Asia/Shanghai # ============================================ # 内网穿透（frpc sidecar） diff --git a/ai_knowledge_base_v2/apps/admin-web/src/App.vue b/ai_knowledge_base_v2/apps/admin-web/src/App.vue index 8be6539..795fcc1 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/App.vue +++ b/ai_knowledge_base_v2/apps/admin-web/src/App.vue @@ -98,7 +98,7 @@ const studentImportResult = ref(null); const entitlementPlanForm = reactive({ name: "", - planType: "basic" as "basic" | "deep" | "addon" | "teacher", + planType: "basic" as "basic" | "deep" | "addon", description: "", validityDays: null as number | null, monthlyTopicLimit: 30 as number | null, @@ -343,7 +343,6 @@ function planTypeLabel(type: string) { basic: "基础版", deep: "深度陪伴版", addon: "高频加购包", - teacher: "老师工作版", }[type] || type; } @@ -423,7 +422,7 @@ async function generateUserReport(type: "weekly" | "monthly" | "stage") { try { await api.generateUserReport(selectedUserDetail.value.user.id, { reportType: type }); selectedUserDetail.value = await api.userDetail(selectedUserDetail.value.user.id); - ElMessage.success("报告已生成"); + ElMessage.success("报告任务已进入后台队列，不会阻塞聊天服务"); } catch (error) { ElMessage.error(error instanceof Error ? error.message : "报告生成失败"); } finally { @@ -431,6 +430,27 @@ async function generateUserReport(type: "weekly" | "monthly" | "stage") { } } +async function refreshSelectedUserDetail() { + if (!selectedUserDetail.value) return; + userDetailLoading.value = true; + try { + selectedUserDetail.value = await api.userDetail(selectedUserDetail.value.user.id); + } finally { + userDetailLoading.value = false; + } +} + +function reportStatusLabel(status: string) { + const labels: Record = { + pending: "等待生成", + running: "生成中", + success: "已完成", + empty: "暂无沉淀", + failed: "生成失败", + }; + return labels[status] || status; +} + async function loadRecordTab(tab = recordTab.value) { loading.value = true; try { @@ -1231,7 +1251,6 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") { - @@ -1727,6 +1746,7 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") {

基于已沉淀的主题摘要生成，不读取整段原始聊天；同一周期重复生成会覆盖旧报告。

+ 刷新状态 生成周报 生成月报 生成阶段总结 @@ -1736,19 +1756,24 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") {
+ 状态：{{ reportStatusLabel(report.status) }} 类型：{{ report.reportTypeLabel }} 周期：{{ report.periodStart }} 至 {{ report.periodEnd }} 来源主题：{{ report.sourceTopicIds.length }} 来源摘要：{{ report.sourceSummaryIds.length }} 模型：{{ report.modelName || '-' }} - 生成：{{ report.generatedAt }} + 触发方式：{{ report.generatedBy.startsWith('schedule:') ? '系统定时' : '管理员手动' }} + 执行次数：{{ report.attemptCount }}/{{ report.maxAttempts }} + 下次执行：{{ report.nextRunAt }} + 完成时间：{{ report.finishedAt || '-' }}
-

生成失败：{{ report.errorMessage }}

- 
{{ report.content }}
+

最近错误：{{ report.errorMessage }}

+ 
{{ report.content }}
+

后台正在生成报告，可以先关闭抽屉，稍后回来查看。

 diff --git a/ai_knowledge_base_v2/apps/admin-web/src/styles.css b/ai_knowledge_base_v2/apps/admin-web/src/styles.css index e42fd68..99c1c2f 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/styles.css +++ b/ai_knowledge_base_v2/apps/admin-web/src/styles.css @@ -2102,6 +2102,15 @@ textarea { color: #b42318; } +.report-pending { + margin: 0; + padding: 12px 14px; + border-radius: 10px; + background: #f2f8f6; + color: #557068; + line-height: 1.6; +} + .chat-summary { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); diff --git a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts index 79f01de..2e84e29 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts +++ b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts @@ -98,7 +98,7 @@ export interface AdminUserMetrics { export interface EntitlementPlan { id: number; name: string; - planType: "basic" | "deep" | "addon" | "teacher"; + planType: "basic" | "deep" | "addon"; description?: string | null; validityDays?: number | null; monthlyTopicLimit?: number | null; @@ -519,6 +519,11 @@ export interface PeriodicReportRecord { status: "success" | "failed" | "empty" | string; errorMessage?: string | null; generatedBy: string; + attemptCount: number; + maxAttempts: number; + nextRunAt?: string | null; + lastStartedAt?: string | null; + finishedAt?: string | null; generatedAt: string; createdAt: string; updatedAt: string; diff --git a/ai_knowledge_base_v2/apps/backend/.env.example b/ai_knowledge_base_v2/apps/backend/.env.example index aa9ceb8..2a5aabc 100644 --- a/ai_knowledge_base_v2/apps/backend/.env.example +++ b/ai_knowledge_base_v2/apps/backend/.env.example @@ -41,6 +41,15 @@ CHAT_MAX_QUEUE_SIZE=20 CHAT_QUEUE_TIMEOUT_SECONDS=60 CHAT_ACTIVE_LEASE_SECONDS=900 +# 周报在每周一 02:00、月报在每月 1 日 03:00 生成，时间按 PERIODIC_REPORT_TIMEZONE 解释。 +PERIODIC_REPORT_WORKER_ENABLED=true +PERIODIC_REPORT_POLL_SECONDS=5 +PERIODIC_REPORT_STALE_MINUTES=30 +PERIODIC_REPORT_MAX_ATTEMPTS=3 +PERIODIC_REPORT_WEEKLY_ENABLED=true +PERIODIC_REPORT_MONTHLY_ENABLED=true +PERIODIC_REPORT_TIMEZONE=Asia/Shanghai + # 本地开发可以使用开发密码；生产环境必须改成高强度密码，且 APP_ENV=production 时不能使用 admin123456 BOOTSTRAP_ADMIN_USERNAME=admin BOOTSTRAP_ADMIN_PASSWORD=admin123456 diff --git a/ai_knowledge_base_v2/apps/backend/alembic/versions/0021_periodic_report_async_jobs.py b/ai_knowledge_base_v2/apps/backend/alembic/versions/0021_periodic_report_async_jobs.py new file mode 100644 index 0000000..4e7f781 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/alembic/versions/0021_periodic_report_async_jobs.py @@ -0,0 +1,62 @@ +"""add durable periodic report job fields + +Revision ID: 0021_periodic_report_async_jobs +Revises: 0020_question_insight_persistence +Create Date: 2026-07-31 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "0021_periodic_report_async_jobs" +down_revision = "0020_question_insight_persistence" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "sys_periodic_report", + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + ) + op.add_column( + "sys_periodic_report", + sa.Column("max_attempts", sa.Integer(), nullable=False, server_default="3"), + ) + op.add_column("sys_periodic_report", sa.Column("next_run_at", sa.DateTime(), nullable=True)) + op.add_column("sys_periodic_report", sa.Column("locked_at", sa.DateTime(), nullable=True)) + op.add_column("sys_periodic_report", sa.Column("locked_by", sa.String(length=120), nullable=True)) + op.add_column("sys_periodic_report", sa.Column("last_started_at", sa.DateTime(), nullable=True)) + op.add_column("sys_periodic_report", sa.Column("finished_at", sa.DateTime(), nullable=True)) + op.execute( + "UPDATE sys_periodic_report " + "SET finished_at = generated_at " + "WHERE status IN ('success', 'empty', 'failed')" + ) + op.execute("UPDATE sys_entitlement_plan SET status = 0 WHERE plan_type = 'teacher'") + op.create_index( + "ix_periodic_report_job_due", + "sys_periodic_report", + ["status", "next_run_at", "id"], + ) + op.create_index( + "ix_periodic_report_job_stale", + "sys_periodic_report", + ["status", "locked_at"], + ) + + +def downgrade() -> None: + op.execute("UPDATE sys_entitlement_plan SET status = 1 WHERE plan_type = 'teacher'") + op.drop_index("ix_periodic_report_job_stale", table_name="sys_periodic_report") + op.drop_index("ix_periodic_report_job_due", table_name="sys_periodic_report") + op.drop_column("sys_periodic_report", "finished_at") + op.drop_column("sys_periodic_report", "last_started_at") + op.drop_column("sys_periodic_report", "locked_by") + op.drop_column("sys_periodic_report", "locked_at") + op.drop_column("sys_periodic_report", "next_run_at") + op.drop_column("sys_periodic_report", "max_attempts") + op.drop_column("sys_periodic_report", "attempt_count") diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py index 06a6f63..e8891d4 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py @@ -295,7 +295,7 @@ def generate_user_report( if payload.reportType not in ("weekly", "monthly", "stage"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="不支持的报告类型") try: - report = PeriodicReportService.generate_for_user( + report = PeriodicReportService.enqueue_for_user( db, user=user, report_type=payload.reportType, # type: ignore[arg-type] @@ -305,9 +305,10 @@ def generate_user_report( ) except ValueError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - OperationLogService.write(db, admin_id=current_admin.id, module="user_report", action="generate", target_id=report.id) + OperationLogService.write(db, admin_id=current_admin.id, module="user_report", action="enqueue", target_id=report.id) db.commit() - return api_success(periodic_report_dict(report)) + db.refresh(report) + return api_success(periodic_report_dict(report), message="报告任务已进入后台队列") @router.put("/user/{user_id}") diff --git a/ai_knowledge_base_v2/apps/backend/app/api/user.py b/ai_knowledge_base_v2/apps/backend/app/api/user.py index 3f99cb1..55af97b 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/user.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/user.py @@ -62,5 +62,10 @@ def periodic_reports( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ) -> dict: - reports = PeriodicReportService.list_user_reports(db, user_id=current_user.id, limit=max(1, min(limit, 50))) + reports = PeriodicReportService.list_user_reports( + db, + user_id=current_user.id, + limit=max(1, min(limit, 50)), + statuses=("success",), + ) return api_success([periodic_report_dict(item) for item in reports]) diff --git a/ai_knowledge_base_v2/apps/backend/app/core/config.py b/ai_knowledge_base_v2/apps/backend/app/core/config.py index a70dffe..e7c0f6e 100644 --- a/ai_knowledge_base_v2/apps/backend/app/core/config.py +++ b/ai_knowledge_base_v2/apps/backend/app/core/config.py @@ -63,6 +63,13 @@ class Settings(BaseSettings): chat_max_queue_size: int = 20 chat_queue_timeout_seconds: int = 60 chat_active_lease_seconds: int = 900 + periodic_report_worker_enabled: bool = True + periodic_report_poll_seconds: int = 5 + periodic_report_stale_minutes: int = 30 + periodic_report_max_attempts: int = 3 + periodic_report_weekly_enabled: bool = True + periodic_report_monthly_enabled: bool = True + periodic_report_timezone: str = "Asia/Shanghai" bootstrap_admin_username: str = "" bootstrap_admin_password: str = "" bootstrap_admin_name: str = "系统管理员" diff --git a/ai_knowledge_base_v2/apps/backend/app/main.py b/ai_knowledge_base_v2/apps/backend/app/main.py index 77f8197..874079c 100644 --- a/ai_knowledge_base_v2/apps/backend/app/main.py +++ b/ai_knowledge_base_v2/apps/backend/app/main.py @@ -12,6 +12,7 @@ from app.core.exception_handlers import register_exception_handlers from app.core.observability import RequestObservabilityMiddleware, configure_logging from app.services.secret_service import SecretService from app.services.maintenance_service import MaintenanceService +from app.services.periodic_report_worker import PeriodicReportWorker import asyncio @@ -22,14 +23,17 @@ async def lifespan(app: FastAPI): if settings.auto_create_tables: create_tables() maintenance_task = asyncio.create_task(MaintenanceService.run_forever()) + periodic_report_task = asyncio.create_task(PeriodicReportWorker.run_forever()) try: yield finally: - maintenance_task.cancel() - try: - await maintenance_task - except asyncio.CancelledError: - pass + for task in (maintenance_task, periodic_report_task): + task.cancel() + for task in (maintenance_task, periodic_report_task): + try: + await task + except asyncio.CancelledError: + pass settings = get_settings() diff --git a/ai_knowledge_base_v2/apps/backend/app/models/growth.py b/ai_knowledge_base_v2/apps/backend/app/models/growth.py index ea6a701..11a3b33 100644 --- a/ai_knowledge_base_v2/apps/backend/app/models/growth.py +++ b/ai_knowledge_base_v2/apps/backend/app/models/growth.py @@ -108,6 +108,8 @@ class PeriodicReport(Base): UniqueConstraint("user_id", "report_type", "period_start", "period_end", name="uq_sys_periodic_report_period"), Index("ix_sys_periodic_report_user_type_created", "user_id", "report_type", "created_at"), Index("ix_sys_periodic_report_status_created", "status", "created_at"), + Index("ix_periodic_report_job_due", "status", "next_run_at", "id"), + Index("ix_periodic_report_job_stale", "status", "locked_at"), ) id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) @@ -123,6 +125,13 @@ class PeriodicReport(Base): status: Mapped[str] = mapped_column(String(20), default="success", index=True, nullable=False) error_message: Mapped[str | None] = mapped_column(Text, nullable=True) generated_by: Mapped[str] = mapped_column(String(30), default="manual", nullable=False) + attempt_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + max_attempts: Mapped[int] = mapped_column(Integer, default=3, nullable=False) + next_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + locked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + locked_by: Mapped[str | None] = mapped_column(String(120), nullable=True) + last_started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) generated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False) diff --git a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py b/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py index dca5ce8..70001b9 100644 --- a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py +++ b/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py @@ -72,7 +72,7 @@ class AdminUserImportRequest(BaseModel): class EntitlementPlanSaveRequest(BaseModel): name: str = Field(min_length=1, max_length=80) - planType: Literal["basic", "deep", "addon", "teacher"] = "basic" + planType: Literal["basic", "deep", "addon"] = "basic" description: str | None = Field(default=None, max_length=255) validityDays: int | None = Field(default=None, ge=1, le=3650) monthlyTopicLimit: int | None = Field(default=None, ge=0, le=100000) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/entitlement_service.py b/ai_knowledge_base_v2/apps/backend/app/services/entitlement_service.py index 12bc2e8..b088fe6 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/entitlement_service.py +++ b/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: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/periodic_report_service.py b/ai_knowledge_base_v2/apps/backend/app/services/periodic_report_service.py index b8f2f46..794491b 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/periodic_report_service.py +++ b/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) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/periodic_report_worker.py b/ai_knowledge_base_v2/apps/backend/app/services/periodic_report_worker.py new file mode 100644 index 0000000..abec70d --- /dev/null +++ b/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) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_entitlements_topics.py b/ai_knowledge_base_v2/apps/backend/tests/test_entitlements_topics.py index 2ec419f..aebda6c 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_entitlements_topics.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_entitlements_topics.py @@ -58,6 +58,28 @@ def test_default_entitlement_uses_basic_plan_when_user_has_no_assignment(): assert view.monthly_topic_remaining == 27 +def test_legacy_teacher_plan_is_not_exposed_or_used_as_default(): + with _db() as db: + user, _session = _seed_user_session(db) + db.add( + EntitlementPlan( + id=10, + name="旧老师工作版", + plan_type="teacher", + monthly_topic_limit=None, + status=1, + sort_order=1, + ) + ) + db.commit() + + plans = EntitlementService.list_plans(db, include_disabled=True) + view = EntitlementService.active_entitlement(db, user) + + assert plans == [] + assert view.plan_type == "legacy" + + def test_assign_user_plan_replaces_previous_active_plan(): with _db() as db: user, _session = _seed_user_session(db) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py b/ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py index 5645764..c77986c 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py @@ -2,16 +2,19 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta -from sqlalchemy import create_engine +from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.models import Base from app.models.ai_config import SystemConfig from app.models.chat import ChatSession, TopicSession +from app.models.entitlement import EntitlementPlan, UserEntitlement from app.models.growth import PeriodicReport, TopicSummary, UserGrowthProfile from app.models.user import User +from app.services.model_service import ModelClientService from app.services.periodic_report_service import PeriodicReportService, periodic_report_dict +from app.services.periodic_report_worker import PeriodicReportWorker, scheduled_period def _db() -> Session: @@ -66,3 +69,175 @@ def test_generate_empty_periodic_report_when_no_summaries(): assert report.status == "empty" assert "还没有可用于生成报告的主题沉淀" in report.content assert periodic_report_dict(report)["sourceSummaryIds"] == [] + + +def test_async_report_job_is_durable_and_idempotent(): + with _db() as db: + now = _now() + db.add(SystemConfig(config_key="mock_model_enabled", config_value="true")) + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="本周主题", message_count=2, last_message_at=now, is_deleted=0) + topic = TopicSession(id=1, user_id=1, chat_session_id=1, title="本周主题", core_question="我该怎么观察", status="completed") + summary = TopicSummary( + id=1, + user_id=1, + topic_session_id=1, + summary="本周看见了身体紧张。", + generated_at=now - timedelta(days=1), + ) + db.add_all([user, session, topic, summary]) + db.commit() + + period_start = now - timedelta(days=7) + report = PeriodicReportService.enqueue_for_user( + db, + user=user, + report_type="weekly", + period_start=period_start, + period_end=now, + generated_by="admin:1", + ) + db.commit() + duplicate = PeriodicReportService.enqueue_for_user( + db, + user=user, + report_type="weekly", + period_start=period_start, + period_end=now, + generated_by="admin:1", + ) + db.commit() + + assert duplicate.id == report.id + assert db.query(PeriodicReport).count() == 1 + assert duplicate.status == "pending" + + report_id = PeriodicReportWorker.claim_next(db, worker_id="test-worker", now=_now() + timedelta(seconds=1)) + assert report_id == report.id + running = db.get(PeriodicReport, report.id) + assert running is not None + assert running.status == "running" + assert running.attempt_count == 1 + + completed = PeriodicReportWorker.execute_claimed(db, report_id=report.id, worker_id="test-worker") + assert completed is not None + assert completed.status == "success" + assert completed.finished_at is not None + assert completed.locked_by is None + + +def test_failed_async_report_is_retried(monkeypatch): + with _db() as db: + now = _now() + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="本周主题", message_count=2, last_message_at=now, is_deleted=0) + topic = TopicSession(id=1, user_id=1, chat_session_id=1, title="本周主题", core_question="我该怎么观察", status="completed") + summary = TopicSummary( + id=1, + user_id=1, + topic_session_id=1, + summary="本周看见了身体紧张。", + generated_at=now - timedelta(days=1), + ) + db.add_all([user, session, topic, summary]) + db.commit() + report = PeriodicReportService.enqueue_for_user( + db, + user=user, + report_type="weekly", + period_start=now - timedelta(days=7), + period_end=now, + ) + db.commit() + monkeypatch.setattr( + ModelClientService, + "generate_text_or_raise", + staticmethod(lambda _db, _prompt: (_ for _ in ()).throw(RuntimeError("模型暂时不可用"))), + ) + + report_id = PeriodicReportWorker.claim_next(db, worker_id="retry-worker", now=_now() + timedelta(seconds=1)) + result = PeriodicReportWorker.execute_claimed(db, report_id=report_id, worker_id="retry-worker") + + assert result is not None + assert result.status == "pending" + assert result.attempt_count == 1 + assert result.next_run_at is not None + assert "模型暂时不可用" in (result.error_message or "") + + +def test_schedule_enqueues_only_users_with_enabled_report_entitlement(): + with _db() as db: + now = datetime(2026, 8, 2, 19, 0, 0) # 上海时间 2026-08-03 周一 03:00 + plan = EntitlementPlan( + id=1, + name="深度陪伴版", + plan_type="deep", + enable_periodic_reports=1, + status=1, + ) + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="本周主题", message_count=2, last_message_at=now, is_deleted=0) + topic = TopicSession(id=1, user_id=1, chat_session_id=1, title="本周主题", core_question="本周问题", status="completed") + summary = TopicSummary( + id=1, + user_id=1, + topic_session_id=1, + summary="本周主题沉淀", + generated_at=datetime(2026, 8, 1, 8, 0, 0), + ) + entitlement = UserEntitlement(id=1, user_id=1, plan_id=1, status="active") + db.add_all([plan, user, session, topic, summary, entitlement]) + db.commit() + + first = PeriodicReportWorker.enqueue_due_schedules(db, now_utc=now) + db.commit() + second = PeriodicReportWorker.enqueue_due_schedules(db, now_utc=now) + db.commit() + + report = db.scalar(select(PeriodicReport).where(PeriodicReport.report_type == "weekly")) + assert first["weekly"] == 1 + assert second == {} + assert report is not None + assert report.status == "pending" + assert report.generated_by == "schedule:weekly" + + +def test_scheduled_period_uses_shanghai_calendar_boundaries(): + weekly = scheduled_period("weekly", datetime(2026, 8, 2, 19, 0, 0)) + monthly = scheduled_period("monthly", datetime(2026, 7, 31, 19, 0, 0)) + manual_weekly_a = PeriodicReportService.default_period("weekly", datetime(2026, 7, 29, 2, 0, 0)) + manual_weekly_b = PeriodicReportService.default_period("weekly", datetime(2026, 7, 31, 2, 0, 0)) + + assert weekly == (datetime(2026, 7, 26, 16, 0, 0), datetime(2026, 8, 2, 16, 0, 0)) + assert monthly == (datetime(2026, 6, 30, 16, 0, 0), datetime(2026, 7, 31, 16, 0, 0)) + assert manual_weekly_a == manual_weekly_b + + +def test_stale_running_job_is_recovered_after_restart(): + with _db() as db: + now = _now() + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + db.add(user) + db.commit() + report = PeriodicReportService.enqueue_for_user( + db, + user=user, + report_type="weekly", + period_start=now - timedelta(days=7), + period_end=now, + ) + report.status = "running" + report.attempt_count = 1 + report.locked_by = "stopped-worker" + report.locked_at = now - timedelta(minutes=31) + db.commit() + + recovered = PeriodicReportWorker.recover_stale_jobs(db, now=now) + db.commit() + db.refresh(report) + + assert recovered == 1 + assert report.status == "pending" + assert report.locked_by is None + assert report.next_run_at == now + assert "自动恢复" in (report.error_message or "") diff --git a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md index b781f90..910f8a5 100644 --- a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md +++ b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md @@ -140,6 +140,29 @@ RESTORE_CONFIRM=YES \ 恢复后必须检查迁移版本、`/api/ready`、管理员登录、用户登录和一次真实问答。备份文件与 `CONFIG_ENCRYPTION_KEY` 必须分别保存；只有数据库备份而没有对应加密密钥时，敏感配置无法解密。 +## 周期报告后台任务 + +周期报告使用数据库保存任务状态，Redis 只用于限制多进程并发。服务重启后，等待中的任务会继续执行，超过 30 分钟仍处于执行中的任务会自动恢复并重试。 + +默认调度规则： + +- 每周一 02:00（Asia/Shanghai）生成上一自然周周报； +- 每月 1 日 03:00（Asia/Shanghai）生成上一自然月月报； +- 只处理权益已开启周期报告、账号有效且该周期存在成功主题摘要的用户； +- 单个任务最多自动执行 3 次，失败后可在后台用户详情中手动重新生成。 + +可通过以下环境变量关闭或调整： + +```text +PERIODIC_REPORT_WORKER_ENABLED=true +PERIODIC_REPORT_WEEKLY_ENABLED=true +PERIODIC_REPORT_MONTHLY_ENABLED=true +PERIODIC_REPORT_TIMEZONE=Asia/Shanghai +PERIODIC_REPORT_POLL_SECONDS=5 +PERIODIC_REPORT_STALE_MINUTES=30 +PERIODIC_REPORT_MAX_ATTEMPTS=3 +``` + ## 回滚原则 1. 先停止新版本服务。 @@ -164,6 +187,7 @@ RESTORE_CONFIRM=YES \ - 5 分钟内 HTTP 5xx 比例超过 2%。 - `/api/ready` 连续失败。 - AI 请求超时或外部服务错误持续增长。 +- 周期报告任务持续失败、长期停留在“生成中”或队列持续积压。 - 问答队列持续接近上限或频繁拒绝请求。 - MySQL、Redis 容器重启或磁盘使用率超过 80%。 - 定时备份任务失败、校验文件缺失或超过 24 小时没有新备份。 diff --git a/ai_knowledge_base_v2/docker-compose.dev.yml b/ai_knowledge_base_v2/docker-compose.dev.yml index 1934b36..ed5c7dd 100644 --- a/ai_knowledge_base_v2/docker-compose.dev.yml +++ b/ai_knowledge_base_v2/docker-compose.dev.yml @@ -49,6 +49,10 @@ services: DB_MAX_OVERFLOW: "20" DB_POOL_TIMEOUT: "30" DB_POOL_RECYCLE: "1800" + PERIODIC_REPORT_WORKER_ENABLED: "true" + PERIODIC_REPORT_WEEKLY_ENABLED: "true" + PERIODIC_REPORT_MONTHLY_ENABLED: "true" + PERIODIC_REPORT_TIMEZONE: Asia/Shanghai JWT_SECRET_KEY: local-dev-secret-change-before-production MOCK_SMS_ENABLED: "true" MOCK_SMS_CODE: "123456" diff --git a/ai_knowledge_base_v2/docker-compose.prod.yml b/ai_knowledge_base_v2/docker-compose.prod.yml index b8cb5bb..86be6f5 100644 --- a/ai_knowledge_base_v2/docker-compose.prod.yml +++ b/ai_knowledge_base_v2/docker-compose.prod.yml @@ -45,6 +45,13 @@ services: BACKEND_WORKERS: ${BACKEND_WORKERS:-4} DB_POOL_SIZE: ${DB_POOL_SIZE:-20} DB_MAX_OVERFLOW: ${DB_MAX_OVERFLOW:-20} + PERIODIC_REPORT_WORKER_ENABLED: ${PERIODIC_REPORT_WORKER_ENABLED:-true} + PERIODIC_REPORT_POLL_SECONDS: ${PERIODIC_REPORT_POLL_SECONDS:-5} + PERIODIC_REPORT_STALE_MINUTES: ${PERIODIC_REPORT_STALE_MINUTES:-30} + PERIODIC_REPORT_MAX_ATTEMPTS: ${PERIODIC_REPORT_MAX_ATTEMPTS:-3} + PERIODIC_REPORT_WEEKLY_ENABLED: ${PERIODIC_REPORT_WEEKLY_ENABLED:-true} + PERIODIC_REPORT_MONTHLY_ENABLED: ${PERIODIC_REPORT_MONTHLY_ENABLED:-true} + PERIODIC_REPORT_TIMEZONE: ${PERIODIC_REPORT_TIMEZONE:-Asia/Shanghai} JWT_SECRET_KEY: ${JWT_SECRET_KEY:?必须配置 JWT_SECRET_KEY} CONFIG_ENCRYPTION_KEY: ${CONFIG_ENCRYPTION_KEY:?必须配置 CONFIG_ENCRYPTION_KEY} MOCK_SMS_ENABLED: "false" diff --git a/ai_knowledge_base_v2/docs/qianwen_product_todo.md b/ai_knowledge_base_v2/docs/qianwen_product_todo.md index 2edaa2e..1ba289e 100644 --- a/ai_knowledge_base_v2/docs/qianwen_product_todo.md +++ b/ai_knowledge_base_v2/docs/qianwen_product_todo.md @@ -82,8 +82,7 @@ Agent 调用链接入成长档案与权益 - 大本营基础版； - 五个月深度陪伴版； -- 高频加购包； -- 老师工作版。 +- 高频加购包。 #### 目标 @@ -100,7 +99,7 @@ Agent 调用链接入成长档案与权益 权益版本字段建议包含： - 名称； -- 类型：基础版 / 深度陪伴版 / 高频加购包 / 老师工作版； +- 类型：基础版 / 深度陪伴版 / 高频加购包； - 有效天数； - 每月主题会话额度； - 是否开启成长档案； @@ -134,7 +133,6 @@ Agent 调用链接入成长档案与权益 - 管理员可以给用户分配权益； - 用户登录后能看到自己的权益状态； - 聊天接口按权益判断是否可用； -- 老师工作版不消耗普通学员额度； - 所有权益变更有审计记录。 --- @@ -788,6 +786,8 @@ AI 日志增加： #### 开发进度 - 2026-07-31：一期已新增周期报告数据表 `sys_periodic_report`，支持周报、月报、阶段总结三种类型；后台用户详情可手动生成/重新生成报告并查看状态、来源主题/摘要、失败原因；用户端“我的实修档案”可查看已生成报告。当前版本先做同步手动生成，后续再接定时/异步任务，聊天链路不会被报告生成阻塞。 +- 2026-07-31：报告生成已改为数据库持久化异步任务。管理员点击生成后接口立即返回，后台 Worker 独立生成；支持等待、执行中、成功、空报告、失败状态，最多自动重试 3 次，服务重启后会恢复超时任务。同一用户、类型和周期由唯一约束保证幂等。 +- 2026-07-31：已接入自然周和自然月定时生成。默认按 `Asia/Shanghai` 在每周一 02:00 生成上一自然周周报、每月 1 日 03:00 生成上一自然月月报；只为权益开启周期报告且本周期存在成功主题摘要的有效用户入队。用户端只显示成功报告，后台可查看任务状态、执行次数、失败原因并手动重新生成。 --- @@ -831,7 +831,7 @@ AI 日志增加： - 2026-07-31：二期第一步已在现有统计结果中增加问题分类、关联 AI 请求数、无知识命中次数、请求失败次数、是否需要知识跟进和运营处理建议；后台问题洞察卡片已展示分类标签、无命中/失败标记和建议动作。暂未新增持久化清洗表和人工合并/拆分能力，避免一次性扩大数据模型。 - 2026-07-31：已新增 `sys_question_insight_cleaned_question` 持久化清洗表和清洗版本字段。后台“刷新并统计”只增量处理尚未清洗的用户消息；低价值消息也会写入过滤占位记录，避免后续反复读取聊天原文；统计、筛选和分页均直接读取清洗结果。并发刷新由数据库唯一约束和单消息事务隔离去重，不会重复沉淀同一消息。 -- 待继续：高频问题一键转知识库补充建议、人工合并/拆分问题组，以及清洗规则升级后的版本重建入口。 +- 暂缓：高频问题一键转知识库补充建议、人工合并/拆分问题组，以及清洗规则升级后的版本重建入口。当前先切换到其他产品板块。 #### 验收标准 @@ -842,44 +842,6 @@ AI 日志增加： --- -### 14. 老师工作版 - -- 优先级：P2 -- 改动大小：M -- 影响范围：用户类型、权限、权益、知识库访问 -- 依赖：权益版本体系、知识库类型规则 - -#### 产品边界 - -老师工作版属于千问千答的内部使用场景，但不是师资管理系统。 - -不做： - -- 老师收入； -- 老师排班； -- 老师认证； -- 见习流程。 - -只做： - -- 老师用 AI 查知识； -- 老师用 AI 生成答疑参考； -- 老师用 AI 检索固定信息和课程内容。 - -#### TODO - -- 增加老师工作版权益； -- 老师账号不消耗普通学员额度； -- 老师可访问内部允许的知识库； -- 老师不能默认查看普通用户隐私； -- 可生成答疑参考，但需要提示“请老师自行判断后使用”。 - -#### 验收标准 - -- 老师账号可使用千问千答； -- 老师使用不影响学员额度； -- 权限边界清楚。 - ## 5. 建议实施批次 ### 第一批：底层产品模型 @@ -925,28 +887,11 @@ AI 日志增加： 1. 周报； 2. 月报； -3. 五个月总结； -4. 老师工作版。 +3. 五个月总结。 -## 6. 当前最推荐下一步 +## 6. 当前阶段结论 -建议下一步不要先做报告、分享稿或老师工作版。 - -最推荐先做： - -```text -权益版本体系 → 主题会话机制 -``` - -原因： - -- 这是收费模式的底座； -- 会影响用户表、聊天表、额度判断、用户端展示和后台统计； -- 后续成长档案、求助卡、分享稿、报告全部依赖它； -- 如果后做，会导致前面功能大面积返工。 - -如果要进一步降低第一批风险，可以拆成： - -1. 先只建权益版本和用户权益，不立刻改完整扣费； -2. 再建主题会话，只做记录不做复杂自动判断； -3. 最后把主题会话接入权益扣减和成长档案。 +- 千问千答不建设老师端或“老师工作版”，老师仍由学员自行联系； +- 周报、月报的异步与定时生成完成后，本计划中的连续陪伴主链路先告一段落； +- 问题洞察的知识库补充建议、人工合并和拆分暂缓； +- 下一阶段切换到其他产品板块时，再按对应板块重新确认范围、依赖和验收标准。