feat: add periodic practice reports

ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import create_engine
+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.growth import PeriodicReport, TopicSummary, UserGrowthProfile
+from app.models.user import User
+from app.services.periodic_report_service import PeriodicReportService, periodic_report_dict
+
+
+def _db() -> Session:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def _now() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+def test_generate_periodic_report_from_topic_summaries():
+    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="本周反复看到表达时身体发紧。",
+            emotions="害怕、紧张",
+            body_feelings="喉咙紧、胸口堵",
+            recommended_homework="表达障碍练习",
+            next_observation="先观察身体反应",
+            generated_at=now - timedelta(days=1),
+        )
+        db.add_all([user, session, topic, summary, UserGrowthProfile(user_id=1, profile_text="用户常在表达前身体发紧。")])
+        db.commit()
+
+        report = PeriodicReportService.generate_for_user(db, user=user, report_type="weekly", period_start=now - timedelta(days=7), period_end=now)
+
+        assert report.status == "success"
+        assert report.content
+        data = periodic_report_dict(report)
+        assert data["sourceSummaryIds"] == [1]
+        assert data["sourceTopicIds"] == [1]
+        assert db.query(PeriodicReport).filter_by(user_id=1).count() == 1
+
+
+def test_generate_empty_periodic_report_when_no_summaries():
+    with _db() as db:
+        user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0)
+        db.add(user)
+        db.commit()
+
+        report = PeriodicReportService.generate_for_user(db, user=user, report_type="monthly")
+
+        assert report.status == "empty"
+        assert "还没有可用于生成报告的主题沉淀" in report.content
+        assert periodic_report_dict(report)["sourceSummaryIds"] == []