feat: add feishu notification data models

tests/test_feishu_models.py

@@ -0,0 +1,104 @@
+from django.utils import timezone
+import pytest
+
+from review_agent.models import (
+    Conversation,
+    FeishuAccessTokenCache,
+    FeishuQuestionLog,
+    FeishuUserMapping,
+    FileSummaryBatch,
+    WorkflowNotificationRecord,
+)
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_feishu_user_mapping_preferred_identifier(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    mapping = FeishuUserMapping.objects.create(
+        system_user=user,
+        feishu_display_name="负责人",
+        feishu_open_id="ou_open",
+        feishu_user_id="user_id",
+        feishu_mobile="13800000000",
+    )
+
+    assert mapping.preferred_identifier() == ("open_id", "ou_open")
+
+    mapping.feishu_open_id = ""
+    assert mapping.preferred_identifier() == ("user_id", "user_id")
+
+    mapping.feishu_user_id = ""
+    assert mapping.preferred_identifier() == ("mobile", "13800000000")
+
+
+def test_feishu_access_token_cache_expiry():
+    now = timezone.now()
+    cache = FeishuAccessTokenCache.objects.create(
+        app_id_hash="hash",
+        tenant_access_token="token",
+        expires_at=now + timezone.timedelta(minutes=5),
+    )
+
+    assert cache.is_valid(now=now)
+
+    cache.expires_at = now - timezone.timedelta(seconds=1)
+    assert not cache.is_valid(now=now)
+
+
+def test_workflow_notification_success_dedupe_only_success(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="飞书")
+    batch = FileSummaryBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="FS-FEISHU",
+        status=FileSummaryBatch.Status.SUCCESS,
+    )
+    dedupe_key = WorkflowNotificationRecord.build_dedupe_key("file_summary", batch.pk, "success")
+    WorkflowNotificationRecord.objects.create(
+        workflow_type="file_summary",
+        workflow_batch_id=batch.pk,
+        workflow_batch_no=batch.batch_no,
+        workflow_status="success",
+        dedupe_key=dedupe_key,
+        trigger_user=user,
+        channel=WorkflowNotificationRecord.Channel.FEISHU_API,
+        send_status=WorkflowNotificationRecord.SendStatus.FAILED,
+        message_title="失败通知",
+    )
+
+    assert not WorkflowNotificationRecord.already_successfully_sent(dedupe_key)
+
+    WorkflowNotificationRecord.objects.create(
+        workflow_type="file_summary",
+        workflow_batch_id=batch.pk,
+        workflow_batch_no=batch.batch_no,
+        workflow_status="success",
+        dedupe_key=dedupe_key,
+        trigger_user=user,
+        channel=WorkflowNotificationRecord.Channel.FEISHU_API,
+        send_status=WorkflowNotificationRecord.SendStatus.SUCCESS,
+        message_title="成功通知",
+    )
+
+    assert WorkflowNotificationRecord.already_successfully_sent(dedupe_key)
+
+
+def test_feishu_question_log_records_summary_without_full_answer(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    log = FeishuQuestionLog.objects.create(
+        system_user=user,
+        source_type=FeishuQuestionLog.SourceType.SIMULATE,
+        question_text="查最新法规核查",
+        intent="batch_status",
+        query_object={"workflow_type": "regulatory_review", "latest": True},
+        answer_summary="RR-001 成功，阻断项 0，高风险 1。",
+        permission_result="allowed",
+        status=FeishuQuestionLog.Status.SUCCESS,
+        processed_at=timezone.now(),
+    )
+
+    assert "完整回答" not in log.answer_summary
+    assert log.query_object["latest"] is True