diff --git a/review_agent/file_summary/views.py b/review_agent/file_summary/views.py index 680d4a3..860c13d 100644 --- a/review_agent/file_summary/views.py +++ b/review_agent/file_summary/views.py @@ -7,7 +7,7 @@ from pathlib import Path from django.http import FileResponse, Http404, JsonResponse from django.views.decorators.http import require_http_methods -from review_agent.models import Conversation, ExportedSummaryFile, FileAttachment, Message +from review_agent.models import ApplicationFormFillBatch, Conversation, ExportedSummaryFile, FileAttachment, Message from review_agent.models import FileSummaryBatch, WorkflowEvent from .events import serialize_event from .paths import resolve_storage_path @@ -271,10 +271,7 @@ def batch_events(request, batch_id: int): @require_http_methods(["GET"]) @login_required def export_download(request, export_id: int): - exported = ExportedSummaryFile.objects.filter( - pk=export_id, - batch__user=request.user, - ).first() + exported = _export_for_user(request.user, export_id) if not exported: raise Http404("导出文件不存在。") path = Path(exported.storage_path) @@ -288,6 +285,8 @@ def export_download(request, export_id: int): ExportedSummaryFile.ExportType.MARKDOWN: "text/markdown; charset=utf-8", ExportedSummaryFile.ExportType.EXCEL: "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ExportedSummaryFile.ExportType.JSON: "application/json; charset=utf-8", + ExportedSummaryFile.ExportType.WORD: "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ExportedSummaryFile.ExportType.PDF: "application/pdf", } content_type = content_types.get(exported.export_type, "application/octet-stream") logger.info( @@ -305,3 +304,21 @@ def export_download(request, export_id: int): filename=exported.file_name, content_type=content_type, ) + + +def _export_for_user(user, export_id: int) -> ExportedSummaryFile | None: + exported = ExportedSummaryFile.objects.filter(pk=export_id).first() + if not exported: + return None + if exported.workflow_type == "application_form_fill": + if not exported.workflow_batch_id: + return None + allowed = ApplicationFormFillBatch.objects.filter( + pk=exported.workflow_batch_id, + conversation__user=user, + is_deleted=False, + ).exists() + return exported if allowed else None + if exported.batch.user_id != user.pk: + return None + return exported diff --git a/review_agent/migrations/0006_alter_exportedsummaryfile_export_type_and_more.py b/review_agent/migrations/0006_alter_exportedsummaryfile_export_type_and_more.py new file mode 100644 index 0000000..b7821f1 --- /dev/null +++ b/review_agent/migrations/0006_alter_exportedsummaryfile_export_type_and_more.py @@ -0,0 +1,353 @@ +# Generated by Django 5.2.14 on 2026-06-07 10:19 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("review_agent", "0005_alter_regulatoryissue_status"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.AlterField( + model_name="exportedsummaryfile", + name="export_type", + field=models.CharField( + choices=[ + ("markdown", "Markdown"), + ("excel", "Excel"), + ("json", "JSON"), + ("word", "Word"), + ("pdf", "PDF"), + ], + max_length=20, + ), + ), + migrations.CreateModel( + name="ApplicationFormFillBatch", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("batch_no", models.CharField(max_length=64, unique=True)), + ( + "status", + models.CharField( + choices=[ + ("pending", "待执行"), + ("running", "执行中"), + ("waiting_user", "等待用户"), + ("success", "成功"), + ("partial_success", "部分成功"), + ("failed", "失败"), + ("cancelled", "已取消"), + ], + default="pending", + max_length=30, + ), + ), + ("requested_templates", models.JSONField(blank=True, default=list)), + ("selected_templates", models.JSONField(blank=True, default=list)), + ("output_types", models.JSONField(blank=True, default=list)), + ( + "registration_type", + models.CharField(blank=True, default="", max_length=80), + ), + ( + "registration_type_source", + models.CharField( + choices=[ + ("user_message", "用户话语"), + ("regulatory_batch", "法规核查批次"), + ("file_extract", "文件抽取"), + ("unknown", "未知"), + ], + default="unknown", + max_length=40, + ), + ), + ( + "product_name", + models.CharField(blank=True, default="", max_length=200), + ), + ("conflict_summary", models.JSONField(blank=True, default=list)), + ("risk_notes", models.JSONField(blank=True, default=list)), + ( + "template_config_version", + models.CharField(blank=True, default="", max_length=80), + ), + ( + "template_config_hash", + models.CharField(blank=True, default="", max_length=128), + ), + ("work_dir", models.CharField(blank=True, default="", max_length=500)), + ("error_message", models.TextField(blank=True, default="")), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("started_at", models.DateTimeField(blank=True, null=True)), + ("finished_at", models.DateTimeField(blank=True, null=True)), + ("archived_at", models.DateTimeField(blank=True, null=True)), + ("is_deleted", models.BooleanField(default=False)), + ( + "conversation", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="application_form_fill_batches", + to="review_agent.conversation", + ), + ), + ( + "source_regulatory_batch", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="application_form_fill_batches", + to="review_agent.regulatoryreviewbatch", + ), + ), + ( + "source_summary_batch", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="application_form_fill_batches", + to="review_agent.filesummarybatch", + ), + ), + ( + "trigger_message", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="triggered_application_form_fill_batches", + to="review_agent.message", + ), + ), + ( + "user", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="review_application_form_fill_batches", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "db_table": "ra_application_form_fill_batch", + "ordering": ["-created_at", "-id"], + }, + ), + migrations.CreateModel( + name="ApplicationFormFillArtifact", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "artifact_type", + models.CharField( + choices=[ + ("template_copy", "模板副本"), + ("field_extract_result", "字段抽取结果"), + ("merged_fields", "字段合并结果"), + ("traceability", "追溯清单"), + ("filled_template", "已填模板"), + ("notification_record", "通知记录"), + ], + max_length=60, + ), + ), + ( + "file_format", + models.CharField( + choices=[ + ("json", "JSON"), + ("excel", "Excel"), + ("docx", "DOCX"), + ("pdf", "PDF"), + ("markdown", "Markdown"), + ], + max_length=20, + ), + ), + ("name", models.CharField(max_length=160)), + ("file_name", models.CharField(max_length=255)), + ("storage_path", models.CharField(max_length=500)), + ("file_size", models.BigIntegerField(default=0)), + ( + "content_hash", + models.CharField(blank=True, default="", max_length=128), + ), + ("metadata", models.JSONField(blank=True, default=dict)), + ( + "created_by_node", + models.CharField(blank=True, default="", max_length=60), + ), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("is_deleted", models.BooleanField(default=False)), + ( + "batch", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="artifacts", + to="review_agent.applicationformfillbatch", + ), + ), + ], + options={ + "db_table": "ra_application_form_fill_artifact", + "ordering": ["-created_at", "-id"], + }, + ), + migrations.CreateModel( + name="ApplicationFormFillNotificationRecord", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "channel", + models.CharField( + choices=[ + ("feishu_cli", "飞书 CLI"), + ("feishu_api", "飞书 API"), + ("mock", "模拟"), + ], + default="mock", + max_length=30, + ), + ), + ("template_codes", models.JSONField(blank=True, default=list)), + ("export_ids", models.JSONField(blank=True, default=list)), + ("message_summary", models.TextField(blank=True, default="")), + ( + "send_status", + models.CharField( + choices=[ + ("pending", "待发送"), + ("success", "成功"), + ("failed", "失败"), + ], + default="pending", + max_length=20, + ), + ), + ("retry_count", models.PositiveIntegerField(default=0)), + ( + "external_message_id", + models.CharField(blank=True, default="", max_length=120), + ), + ("error_message", models.TextField(blank=True, default="")), + ("sent_at", models.DateTimeField(blank=True, null=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ("is_deleted", models.BooleanField(default=False)), + ( + "batch", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="notifications", + to="review_agent.applicationformfillbatch", + ), + ), + ( + "recipient", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="application_form_fill_notifications", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "db_table": "ra_application_form_fill_notification_record", + "ordering": ["-created_at", "-id"], + }, + ), + migrations.AddIndex( + model_name="applicationformfillbatch", + index=models.Index( + fields=["conversation", "status"], name="idx_ra_aff_batch_conv_status" + ), + ), + migrations.AddIndex( + model_name="applicationformfillbatch", + index=models.Index( + fields=["source_summary_batch"], name="idx_ra_aff_batch_summary" + ), + ), + migrations.AddIndex( + model_name="applicationformfillbatch", + index=models.Index( + fields=["source_regulatory_batch"], name="idx_ra_aff_batch_regulatory" + ), + ), + migrations.AddIndex( + model_name="applicationformfillbatch", + index=models.Index( + fields=["user", "created_at"], name="idx_ra_aff_batch_user_created" + ), + ), + migrations.AddIndex( + model_name="applicationformfillbatch", + index=models.Index(fields=["created_at"], name="idx_ra_aff_batch_created"), + ), + migrations.AddIndex( + model_name="applicationformfillartifact", + index=models.Index( + fields=["batch", "artifact_type"], name="idx_ra_aff_artifact_batch_type" + ), + ), + migrations.AddIndex( + model_name="applicationformfillartifact", + index=models.Index( + fields=["file_format"], name="idx_ra_aff_artifact_format" + ), + ), + migrations.AddIndex( + model_name="applicationformfillartifact", + index=models.Index( + fields=["created_at"], name="idx_ra_aff_artifact_created" + ), + ), + migrations.AddIndex( + model_name="applicationformfillnotificationrecord", + index=models.Index( + fields=["batch", "created_at"], name="idx_ra_aff_notify_batch" + ), + ), + migrations.AddIndex( + model_name="applicationformfillnotificationrecord", + index=models.Index( + fields=["recipient", "send_status"], name="idx_ra_aff_notify_recipient" + ), + ), + migrations.AddIndex( + model_name="applicationformfillnotificationrecord", + index=models.Index( + fields=["send_status", "retry_count"], name="idx_ra_aff_notify_status" + ), + ), + ] diff --git a/review_agent/models.py b/review_agent/models.py index 3cb703e..541a209 100644 --- a/review_agent/models.py +++ b/review_agent/models.py @@ -334,6 +334,8 @@ class ExportedSummaryFile(models.Model): MARKDOWN = "markdown", "Markdown" EXCEL = "excel", "Excel" JSON = "json", "JSON" + WORD = "word", "Word" + PDF = "pdf", "PDF" class Status(models.TextChoices): SUCCESS = "success", "成功" @@ -397,6 +399,92 @@ class RegulatoryRuleVersion(models.Model): return self.code +class ApplicationFormFillBatch(models.Model): + """Tracks one application-form auto-fill workflow run.""" + + class Status(models.TextChoices): + PENDING = "pending", "待执行" + RUNNING = "running", "执行中" + WAITING_USER = "waiting_user", "等待用户" + SUCCESS = "success", "成功" + PARTIAL_SUCCESS = "partial_success", "部分成功" + FAILED = "failed", "失败" + CANCELLED = "cancelled", "已取消" + + class RegistrationTypeSource(models.TextChoices): + USER_MESSAGE = "user_message", "用户话语" + REGULATORY_BATCH = "regulatory_batch", "法规核查批次" + FILE_EXTRACT = "file_extract", "文件抽取" + UNKNOWN = "unknown", "未知" + + conversation = models.ForeignKey( + Conversation, + on_delete=models.CASCADE, + related_name="application_form_fill_batches", + ) + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + related_name="review_application_form_fill_batches", + ) + trigger_message = models.ForeignKey( + Message, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="triggered_application_form_fill_batches", + ) + source_summary_batch = models.ForeignKey( + FileSummaryBatch, + on_delete=models.PROTECT, + related_name="application_form_fill_batches", + ) + source_regulatory_batch = models.ForeignKey( + "RegulatoryReviewBatch", + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="application_form_fill_batches", + ) + batch_no = models.CharField(max_length=64, unique=True) + status = models.CharField(max_length=30, choices=Status.choices, default=Status.PENDING) + requested_templates = models.JSONField(default=list, blank=True) + selected_templates = models.JSONField(default=list, blank=True) + output_types = models.JSONField(default=list, blank=True) + registration_type = models.CharField(max_length=80, blank=True, default="") + registration_type_source = models.CharField( + max_length=40, + choices=RegistrationTypeSource.choices, + default=RegistrationTypeSource.UNKNOWN, + ) + product_name = models.CharField(max_length=200, blank=True, default="") + conflict_summary = models.JSONField(default=list, blank=True) + risk_notes = models.JSONField(default=list, blank=True) + template_config_version = models.CharField(max_length=80, blank=True, default="") + template_config_hash = models.CharField(max_length=128, blank=True, default="") + work_dir = models.CharField(max_length=500, blank=True, default="") + error_message = models.TextField(blank=True, default="") + created_at = models.DateTimeField(auto_now_add=True) + started_at = models.DateTimeField(null=True, blank=True) + finished_at = models.DateTimeField(null=True, blank=True) + archived_at = models.DateTimeField(null=True, blank=True) + is_deleted = models.BooleanField(default=False) + + class Meta: + db_table = "ra_application_form_fill_batch" + ordering = ["-created_at", "-id"] + indexes = [ + models.Index(fields=["conversation", "status"], name="idx_ra_aff_batch_conv_status"), + models.Index(fields=["source_summary_batch"], name="idx_ra_aff_batch_summary"), + models.Index(fields=["source_regulatory_batch"], name="idx_ra_aff_batch_regulatory"), + models.Index(fields=["user", "created_at"], name="idx_ra_aff_batch_user_created"), + models.Index(fields=["created_at"], name="idx_ra_aff_batch_created"), + ] + + def __str__(self) -> str: + return self.batch_no + + class RegulatoryReviewBatch(models.Model): """Tracks one NMPA regulatory review workflow run.""" @@ -571,3 +659,98 @@ class RegulatoryNotificationRecord(models.Model): indexes = [ models.Index(fields=["batch", "status"], name="idx_ra_rr_notify_status"), ] + + +class ApplicationFormFillArtifact(models.Model): + """Stores auto-fill intermediate files and generated artifacts.""" + + class ArtifactType(models.TextChoices): + TEMPLATE_COPY = "template_copy", "模板副本" + FIELD_EXTRACT_RESULT = "field_extract_result", "字段抽取结果" + MERGED_FIELDS = "merged_fields", "字段合并结果" + TRACEABILITY = "traceability", "追溯清单" + FILLED_TEMPLATE = "filled_template", "已填模板" + NOTIFICATION_RECORD = "notification_record", "通知记录" + + class FileFormat(models.TextChoices): + JSON = "json", "JSON" + EXCEL = "excel", "Excel" + DOCX = "docx", "DOCX" + PDF = "pdf", "PDF" + MARKDOWN = "markdown", "Markdown" + + batch = models.ForeignKey( + ApplicationFormFillBatch, + on_delete=models.CASCADE, + related_name="artifacts", + ) + artifact_type = models.CharField(max_length=60, choices=ArtifactType.choices) + file_format = models.CharField(max_length=20, choices=FileFormat.choices) + name = models.CharField(max_length=160) + file_name = models.CharField(max_length=255) + storage_path = models.CharField(max_length=500) + file_size = models.BigIntegerField(default=0) + content_hash = models.CharField(max_length=128, blank=True, default="") + metadata = models.JSONField(default=dict, blank=True) + created_by_node = models.CharField(max_length=60, blank=True, default="") + created_at = models.DateTimeField(auto_now_add=True) + is_deleted = models.BooleanField(default=False) + + class Meta: + db_table = "ra_application_form_fill_artifact" + ordering = ["-created_at", "-id"] + indexes = [ + models.Index(fields=["batch", "artifact_type"], name="idx_ra_aff_artifact_batch_type"), + models.Index(fields=["file_format"], name="idx_ra_aff_artifact_format"), + models.Index(fields=["created_at"], name="idx_ra_aff_artifact_created"), + ] + + +class ApplicationFormFillNotificationRecord(models.Model): + """Stores mock/Feishu notification records for application-form auto-fill.""" + + class Channel(models.TextChoices): + FEISHU_CLI = "feishu_cli", "飞书 CLI" + FEISHU_API = "feishu_api", "飞书 API" + MOCK = "mock", "模拟" + + class SendStatus(models.TextChoices): + PENDING = "pending", "待发送" + SUCCESS = "success", "成功" + FAILED = "failed", "失败" + + batch = models.ForeignKey( + ApplicationFormFillBatch, + on_delete=models.CASCADE, + related_name="notifications", + ) + recipient = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + related_name="application_form_fill_notifications", + ) + channel = models.CharField(max_length=30, choices=Channel.choices, default=Channel.MOCK) + template_codes = models.JSONField(default=list, blank=True) + export_ids = models.JSONField(default=list, blank=True) + message_summary = models.TextField(blank=True, default="") + send_status = models.CharField( + max_length=20, + choices=SendStatus.choices, + default=SendStatus.PENDING, + ) + retry_count = models.PositiveIntegerField(default=0) + external_message_id = models.CharField(max_length=120, blank=True, default="") + error_message = models.TextField(blank=True, default="") + sent_at = models.DateTimeField(null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + is_deleted = models.BooleanField(default=False) + + class Meta: + db_table = "ra_application_form_fill_notification_record" + ordering = ["-created_at", "-id"] + indexes = [ + models.Index(fields=["batch", "created_at"], name="idx_ra_aff_notify_batch"), + models.Index(fields=["recipient", "send_status"], name="idx_ra_aff_notify_recipient"), + models.Index(fields=["send_status", "retry_count"], name="idx_ra_aff_notify_status"), + ] diff --git a/tests/test_application_form_fill_models.py b/tests/test_application_form_fill_models.py new file mode 100644 index 0000000..92be9df --- /dev/null +++ b/tests/test_application_form_fill_models.py @@ -0,0 +1,109 @@ +import pytest + +from review_agent.models import ( + ApplicationFormFillArtifact, + ApplicationFormFillBatch, + ApplicationFormFillNotificationRecord, + Conversation, + ExportedSummaryFile, + FileSummaryBatch, + Message, + RegulatoryReviewBatch, +) + + +pytestmark = pytest.mark.django_db + + +def test_application_form_fill_models_store_batch_artifact_notification_and_exports(django_user_model): + user = django_user_model.objects.create_user(username="owner", password="pass") + conversation = Conversation.objects.create(user=user, title="自动填表") + trigger = Message.objects.create( + conversation=conversation, + role=Message.Role.USER, + content="帮我填注册证", + ) + summary_batch = FileSummaryBatch.objects.create( + conversation=conversation, + user=user, + batch_no="FS-AFF-READY", + status=FileSummaryBatch.Status.SUCCESS, + ) + regulatory_batch = RegulatoryReviewBatch.objects.create( + conversation=conversation, + user=user, + source_summary_batch=summary_batch, + batch_no="RR-AFF-SOURCE", + condition_json={"confirmed": True, "registration_type": "首次注册"}, + ) + + batch = ApplicationFormFillBatch.objects.create( + conversation=conversation, + user=user, + trigger_message=trigger, + source_summary_batch=summary_batch, + source_regulatory_batch=regulatory_batch, + batch_no="AFF-20260607153000-abcdef", + requested_templates=["registration_certificate"], + selected_templates=["registration_certificate"], + output_types=["word", "excel", "json"], + registration_type="首次注册", + registration_type_source=ApplicationFormFillBatch.RegistrationTypeSource.USER_MESSAGE, + product_name="甲胎蛋白检测试剂盒", + conflict_summary=[{"field_key": "storage_condition"}], + risk_notes=[{"type": "pdf_pending"}], + template_config_version="application_form_templates_v1", + template_config_hash="hash", + work_dir="media/application_form_fill/1/1/AFF-20260607153000-abcdef", + ) + artifact = ApplicationFormFillArtifact.objects.create( + batch=batch, + artifact_type=ApplicationFormFillArtifact.ArtifactType.FILLED_TEMPLATE, + file_format=ApplicationFormFillArtifact.FileFormat.DOCX, + name="注册证格式", + file_name="filled.docx", + storage_path="media/application_form_fill/filled.docx", + file_size=123, + content_hash="sha256", + metadata={"template_code": "registration_certificate"}, + created_by_node="word_fill", + ) + notification = ApplicationFormFillNotificationRecord.objects.create( + batch=batch, + recipient=user, + channel=ApplicationFormFillNotificationRecord.Channel.MOCK, + template_codes=["registration_certificate"], + export_ids=[1], + message_summary="自动填表完成", + send_status=ApplicationFormFillNotificationRecord.SendStatus.FAILED, + retry_count=1, + error_message="mock failed", + ) + word_export = ExportedSummaryFile.objects.create( + batch=summary_batch, + workflow_type="application_form_fill", + workflow_batch_id=batch.pk, + export_category="filled_template", + export_type=ExportedSummaryFile.ExportType.WORD, + file_name="filled.docx", + storage_path="media/application_form_fill/filled.docx", + ) + pdf_export = ExportedSummaryFile.objects.create( + batch=summary_batch, + workflow_type="application_form_fill", + workflow_batch_id=batch.pk, + export_category="filled_template", + export_type=ExportedSummaryFile.ExportType.PDF, + file_name="filled.pdf", + storage_path="media/application_form_fill/filled.pdf", + ) + + assert batch.status == ApplicationFormFillBatch.Status.PENDING + assert batch.source_summary_batch == summary_batch + assert batch.source_regulatory_batch == regulatory_batch + assert artifact.content_hash == "sha256" + assert artifact.metadata["template_code"] == "registration_certificate" + assert notification.send_status == ApplicationFormFillNotificationRecord.SendStatus.FAILED + assert notification.retry_count == 1 + assert word_export.export_type == ExportedSummaryFile.ExportType.WORD + assert pdf_export.export_type == ExportedSummaryFile.ExportType.PDF diff --git a/tests/test_file_summary_views.py b/tests/test_file_summary_views.py index 6aeaa7f..ec0411f 100644 --- a/tests/test_file_summary_views.py +++ b/tests/test_file_summary_views.py @@ -4,6 +4,7 @@ import json import pytest from review_agent.models import ( + ApplicationFormFillBatch, Conversation, ExportedSummaryFile, FileAttachment, @@ -109,6 +110,54 @@ def test_export_download_requires_batch_owner(client, tmp_path, django_user_mode assert allowed["Content-Type"].startswith("text/markdown") +def test_export_download_checks_application_form_fill_batch_owner(client, tmp_path, django_user_model): + owner = django_user_model.objects.create_user(username="owner", password="pass") + other = django_user_model.objects.create_user(username="other", password="pass") + owner_conversation = Conversation.objects.create(user=owner, title="自动填表") + other_conversation = Conversation.objects.create(user=other, title="其他对话") + owner_summary = FileSummaryBatch.objects.create( + conversation=owner_conversation, + user=owner, + batch_no="FS-AFF-OWNER", + status=FileSummaryBatch.Status.SUCCESS, + ) + other_summary = FileSummaryBatch.objects.create( + conversation=other_conversation, + user=other, + batch_no="FS-AFF-OTHER", + status=FileSummaryBatch.Status.SUCCESS, + ) + form_batch = ApplicationFormFillBatch.objects.create( + conversation=owner_conversation, + user=owner, + source_summary_batch=owner_summary, + batch_no="AFF-DL", + ) + report_path = tmp_path / "filled.docx" + report_path.write_bytes(b"word-content") + exported = ExportedSummaryFile.objects.create( + batch=other_summary, + workflow_type="application_form_fill", + workflow_batch_id=form_batch.pk, + export_category="filled_template", + export_type=ExportedSummaryFile.ExportType.WORD, + file_name="filled.docx", + storage_path=str(report_path), + ) + + client.force_login(other) + denied = client.get(reverse("file_summary_export_download", args=[exported.pk])) + assert denied.status_code == 404 + + client.force_login(owner) + allowed = client.get(reverse("file_summary_export_download", args=[exported.pk])) + assert allowed.status_code == 200 + assert allowed["Content-Type"].startswith( + "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + ) + assert b"".join(allowed.streaming_content) == b"word-content" + + def test_conversation_messages_returns_incremental_messages(client, django_user_model): owner = django_user_model.objects.create_user(username="owner", password="pass") other = django_user_model.objects.create_user(username="other", password="pass")