feat(file-summary): 生成汇总报告和导出下载

This commit is contained in:
2026-06-06 01:22:49 +08:00
parent 18d045d487
commit 61bd31790b
8 changed files with 286 additions and 5 deletions

View File

@@ -0,0 +1,82 @@
from pathlib import Path
import pytest
from openpyxl import load_workbook
from review_agent.file_summary.services.export_excel import generate_excel_export
from review_agent.file_summary.services.report import generate_markdown_report
from review_agent.models import Conversation, FileSummaryBatch, FileSummaryItem, Message
pytestmark = pytest.mark.django_db
def make_batch(tmp_path, 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-R",
work_dir=str(tmp_path),
total_files=1,
success_files=1,
total_pages=2,
)
FileSummaryItem.objects.create(
batch=batch,
file_index=1,
file_name="a.xlsx",
file_type="xlsx",
relative_path="a.xlsx",
storage_path=str(tmp_path / "a.xlsx"),
page_count=2,
statistics_status=FileSummaryItem.StatisticsStatus.SUCCESS,
)
return batch
def test_generate_markdown_report_creates_export_and_summary(tmp_path, django_user_model):
batch = make_batch(tmp_path, django_user_model)
exported, summary = generate_markdown_report(batch)
assert exported.export_type == "markdown"
assert Path(exported.storage_path).exists()
assert "| 序号 | 目录层级 | 文件名 | 类型 | 页数 | 状态 | 异常说明 |" in summary
assert "a.xlsx" in Path(exported.storage_path).read_text(encoding="utf-8")
def test_generate_excel_export_contains_summary_and_items(tmp_path, django_user_model):
batch = make_batch(tmp_path, django_user_model)
exported = generate_excel_export(batch)
workbook = load_workbook(exported.storage_path)
assert workbook.sheetnames == ["汇总信息", "文件明细"]
assert workbook["文件明细"]["C2"].value == "a.xlsx"
def test_workflow_report_node_writes_assistant_message(tmp_path, settings, django_user_model):
from review_agent.file_summary.workflow import create_file_summary_batch, start_file_summary_workflow
from review_agent.models import FileAttachment
settings.MEDIA_ROOT = tmp_path
settings.FILE_SUMMARY_ASYNC = False
user = django_user_model.objects.create_user(username="owner", password="pass")
conversation = Conversation.objects.create(user=user, title="会话")
file_path = tmp_path / "a.xlsx"
file_path.write_bytes(b"not a real workbook")
FileAttachment.objects.create(
conversation=conversation,
user=user,
original_name="a.txt",
storage_path=str(file_path),
file_size=file_path.stat().st_size,
)
batch = create_file_summary_batch(conversation=conversation, user=user)
batch.work_dir = str(tmp_path / "batch")
batch.save(update_fields=["work_dir"])
start_file_summary_workflow(batch, async_run=False)
assert Message.objects.filter(conversation=conversation, role=Message.Role.ASSISTANT).exists()

View File

@@ -2,7 +2,7 @@ from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
import pytest
from review_agent.models import Conversation, FileAttachment
from review_agent.models import Conversation, ExportedSummaryFile, FileAttachment, FileSummaryBatch
pytestmark = pytest.mark.django_db
@@ -73,3 +73,26 @@ def test_delete_attachment_is_logical_and_scoped(client, settings, tmp_path, dja
assert response.status_code == 200
assert attachment.upload_status == FileAttachment.UploadStatus.DELETED
assert attachment.is_active is False
def test_export_download_requires_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")
conversation = Conversation.objects.create(user=owner, title="会话")
batch = FileSummaryBatch.objects.create(conversation=conversation, user=owner, batch_no="FS-DL")
report_path = tmp_path / "summary.md"
report_path.write_text("ok", encoding="utf-8")
exported = ExportedSummaryFile.objects.create(
batch=batch,
export_type=ExportedSummaryFile.ExportType.MARKDOWN,
file_name="summary.md",
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