egg/Task_Reporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add AI report generation with DIFY integration

Browse Source 
- Add Users table for display name resolution from AD authentication
- Integrate DIFY AI service for report content generation
- Create docx assembly service with image embedding from MinIO
- Add REST API endpoints for report generation and download
- Add WebSocket notifications for generation progress
- Add frontend UI with progress modal and download functionality
- Add integration tests for report generation flow

Report sections (Traditional Chinese):
- 事件摘要 (Summary)
- 時間軸 (Timeline)
- 參與人員 (Participants)
- 處理過程 (Resolution Process)
- 目前狀態 (Current Status)
- 最終處置結果 (Final Resolution)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
egg
2025-12-04 18:32:40 +08:00
parent 77091eefb5
commit 3927441103
32 changed files with 4374 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
tests/test_report_generation.py Normal file
View File

@@ -0,0 +1,344 @@
"""Integration tests for AI report generation
Tests the report generation flow:
1. Data collection from room
2. DIFY prompt construction
3. Document assembly
"""
import pytest
from datetime import datetime
from unittest.mock import Mock, patch, AsyncMock
import json
# Test data fixtures
@pytest.fixture
def sample_room_data():
"""Sample room metadata for testing"""
return {
"room_id": "test-room-123",
"title": "設備故障測試",
"incident_type": "equipment_failure",
"severity": "high",
"status": "active",
"location": "A棟生產線",
"description": "測試用事件描述",
"resolution_notes": None,
"created_at": datetime(2024, 1, 15, 9, 0),
"resolved_at": None,
"created_by": "test@example.com",
}
@pytest.fixture
def sample_messages():
"""Sample messages for testing"""
return [
{
"message_id": "msg-1",
"sender_id": "user1@example.com",
"sender_name": "張三",
"content": "發現設備異常",
"message_type": "text",
"created_at": datetime(2024, 1, 15, 9, 5),
"file_name": None,
},
{
"message_id": "msg-2",
"sender_id": "user2@example.com",
"sender_name": "李四",
"content": "收到，立即前往查看",
"message_type": "text",
"created_at": datetime(2024, 1, 15, 9, 10),
"file_name": None,
},
]
@pytest.fixture
def sample_members():
"""Sample members for testing"""
return [
{
"user_id": "user1@example.com",
"display_name": "張三",
"role": "owner",
},
{
"user_id": "user2@example.com",
"display_name": "李四",
"role": "editor",
},
]
@pytest.fixture
def sample_files():
"""Sample files for testing"""
return [
{
"file_id": "file-1",
"filename": "故障照片.jpg",
"file_type": "image",
"mime_type": "image/jpeg",
"uploaded_at": datetime(2024, 1, 15, 9, 15),
"uploader_id": "user1@example.com",
"uploader_name": "張三",
"minio_object_path": "rooms/test-room-123/file-1.jpg",
},
]
@pytest.fixture
def sample_ai_response():
"""Sample AI response JSON for testing"""
return {
"summary": {
"content": "A棟生產線設備於2024年1月15日發生異常，維修人員已前往處理。"
},
"timeline": {
"events": [
{"time": "09:05", "description": "張三發現設備異常並通報"},
{"time": "09:10", "description": "李四收到通報前往查看"},
]
},
"participants": {
"members": [
{"name": "張三", "role": "事件發起人"},
{"name": "李四", "role": "維修負責人"},
]
},
"resolution_process": {
"content": "維修人員接獲通報後立即前往現場查看設備狀況。"
},
"current_status": {
"status": "active",
"description": "維修人員正在現場處理中"
},
"final_resolution": {
"has_resolution": False,
"content": ""
}
}
class TestPromptConstruction:
"""Tests for prompt construction"""
def test_build_report_prompt_includes_room_info(self, sample_room_data, sample_messages, sample_members, sample_files):
"""Test that prompt includes room information"""
from app.modules.report_generation.prompts import build_report_prompt
prompt = build_report_prompt(
room_data=sample_room_data,
messages=sample_messages,
members=sample_members,
files=sample_files,
)
assert "設備故障測試" in prompt
assert "設備故障" in prompt # Translated incident type
assert "高" in prompt # Translated severity
assert "A棟生產線" in prompt
def test_build_report_prompt_includes_messages(self, sample_room_data, sample_messages, sample_members, sample_files):
"""Test that prompt includes message content"""
from app.modules.report_generation.prompts import build_report_prompt
prompt = build_report_prompt(
room_data=sample_room_data,
messages=sample_messages,
members=sample_members,
files=sample_files,
)
assert "發現設備異常" in prompt
assert "張三" in prompt
assert "李四" in prompt
def test_build_report_prompt_includes_files(self, sample_room_data, sample_messages, sample_members, sample_files):
"""Test that prompt includes file information"""
from app.modules.report_generation.prompts import build_report_prompt
prompt = build_report_prompt(
room_data=sample_room_data,
messages=sample_messages,
members=sample_members,
files=sample_files,
)
assert "故障照片.jpg" in prompt
assert "圖片" in prompt # File type label
class TestDifyJsonParsing:
"""Tests for DIFY response JSON parsing"""
def test_extract_json_from_pure_json(self, sample_ai_response):
"""Test parsing pure JSON response"""
from app.modules.report_generation.services.dify_client import DifyService
service = DifyService()
text = json.dumps(sample_ai_response)
result = service._extract_json(text)
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
def test_extract_json_from_markdown_code_block(self, sample_ai_response):
"""Test parsing JSON from markdown code block"""
from app.modules.report_generation.services.dify_client import DifyService
service = DifyService()
text = f"```json\n{json.dumps(sample_ai_response)}\n```"
result = service._extract_json(text)
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
def test_extract_json_from_mixed_text(self, sample_ai_response):
"""Test parsing JSON embedded in other text"""
from app.modules.report_generation.services.dify_client import DifyService
service = DifyService()
text = f"Here is the report:\n{json.dumps(sample_ai_response)}\nEnd of report."
result = service._extract_json(text)
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
def test_validate_schema_with_valid_data(self, sample_ai_response):
"""Test schema validation with valid data"""
from app.modules.report_generation.services.dify_client import DifyService
service = DifyService()
# Should not raise any exception
service._validate_schema(sample_ai_response)
def test_validate_schema_with_missing_section(self, sample_ai_response):
"""Test schema validation with missing section"""
from app.modules.report_generation.services.dify_client import DifyService, DifyValidationError
service = DifyService()
del sample_ai_response["summary"]
with pytest.raises(DifyValidationError) as exc_info:
service._validate_schema(sample_ai_response)
assert "summary" in str(exc_info.value)
class TestDocxGeneration:
"""Tests for docx document generation"""
def test_create_report_returns_bytesio(self, sample_room_data, sample_ai_response, sample_files):
"""Test that document assembly returns BytesIO"""
from app.modules.report_generation.services.docx_service import DocxAssemblyService
import io
service = DocxAssemblyService()
result = service.create_report(
room_data=sample_room_data,
ai_content=sample_ai_response,
files=sample_files,
include_images=False, # Skip image download for test
include_file_list=True,
)
assert isinstance(result, io.BytesIO)
assert result.getvalue() # Should have content
def test_create_report_is_valid_docx(self, sample_room_data, sample_ai_response, sample_files):
"""Test that generated document is valid DOCX"""
from app.modules.report_generation.services.docx_service import DocxAssemblyService
from docx import Document
service = DocxAssemblyService()
result = service.create_report(
room_data=sample_room_data,
ai_content=sample_ai_response,
files=sample_files,
include_images=False,
include_file_list=True,
)
# Should be able to parse as DOCX
doc = Document(result)
assert len(doc.paragraphs) > 0
class TestReportDataService:
"""Tests for report data collection service"""
def test_to_prompt_dict_format(self, sample_room_data, sample_messages, sample_members, sample_files):
"""Test data conversion to prompt dictionary format"""
from app.modules.report_generation.services.report_data_service import (
ReportDataService,
RoomReportData,
MessageData,
MemberData,
FileData,
)
# Create RoomReportData manually for testing
room_data = RoomReportData(
room_id=sample_room_data["room_id"],
title=sample_room_data["title"],
incident_type=sample_room_data["incident_type"],
severity=sample_room_data["severity"],
status=sample_room_data["status"],
location=sample_room_data["location"],
description=sample_room_data["description"],
resolution_notes=sample_room_data["resolution_notes"],
created_at=sample_room_data["created_at"],
resolved_at=sample_room_data["resolved_at"],
created_by=sample_room_data["created_by"],
messages=[
MessageData(
message_id=m["message_id"],
sender_id=m["sender_id"],
sender_name=m["sender_name"],
content=m["content"],
message_type=m["message_type"],
created_at=m["created_at"],
file_name=m.get("file_name"),
)
for m in sample_messages
],
members=[
MemberData(
user_id=m["user_id"],
display_name=m["display_name"],
role=m["role"],
)
for m in sample_members
],
files=[
FileData(
file_id=f["file_id"],
filename=f["filename"],
file_type=f["file_type"],
mime_type=f["mime_type"],
uploaded_at=f["uploaded_at"],
uploader_id=f["uploader_id"],
uploader_name=f["uploader_name"],
minio_object_path=f["minio_object_path"],
)
for f in sample_files
],
)
# Test conversion using class method directly (not from db session)
# Create mock db session
mock_db = Mock()
service = ReportDataService(mock_db)
result = service.to_prompt_dict(room_data)
assert "room_data" in result
assert "messages" in result
assert "members" in result
assert "files" in result
assert result["room_data"]["title"] == sample_room_data["title"]
assert len(result["messages"]) == 2
assert len(result["members"]) == 2
assert len(result["files"]) == 1
if __name__ == "__main__":
pytest.main([__file__, "-v"])

164
tests/test_user_service.py Normal file
View File

@@ -0,0 +1,164 @@
"""Unit tests for user service
Tests for the users table and upsert operations used in report generation.
"""
import pytest
from datetime import datetime, timedelta
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.database import Base
from app.modules.auth.models import User
from app.modules.auth.services.user_service import upsert_user, get_user_by_id, get_display_name
# Create in-memory SQLite database for testing
@pytest.fixture
def db_session():
"""Create a fresh database session for each test"""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionLocal()
try:
yield session
finally:
session.close()
class TestUpsertUser:
"""Tests for upsert_user function"""
def test_create_new_user(self, db_session):
"""Test creating a new user record"""
user = upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User 測試用戶",
office_location="Taipei",
job_title="Engineer",
)
assert user.user_id == "test@example.com"
assert user.display_name == "Test User 測試用戶"
assert user.office_location == "Taipei"
assert user.job_title == "Engineer"
assert user.last_login_at is not None
assert user.created_at is not None
def test_update_existing_user(self, db_session):
"""Test updating an existing user record"""
# Create initial user
user1 = upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Original Name",
office_location="Taipei",
job_title="Junior Engineer",
)
original_created_at = user1.created_at
original_last_login = user1.last_login_at
# Wait a tiny bit to ensure timestamp difference
import time
time.sleep(0.01)
# Update same user
user2 = upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Updated Name 更新名稱",
office_location="Kaohsiung",
job_title="Senior Engineer",
)
# Verify update
assert user2.user_id == "test@example.com"
assert user2.display_name == "Updated Name 更新名稱"
assert user2.office_location == "Kaohsiung"
assert user2.job_title == "Senior Engineer"
# created_at should be preserved
assert user2.created_at == original_created_at
# last_login_at should be updated
assert user2.last_login_at >= original_last_login
def test_upsert_with_null_optional_fields(self, db_session):
"""Test upsert with null office_location and job_title"""
user = upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User",
office_location=None,
job_title=None,
)
assert user.office_location is None
assert user.job_title is None
def test_update_clears_optional_fields(self, db_session):
"""Test that updating with None clears optional fields"""
# Create with values
upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User",
office_location="Taipei",
job_title="Engineer",
)
# Update with None
user = upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User",
office_location=None,
job_title=None,
)
assert user.office_location is None
assert user.job_title is None
class TestGetUserById:
"""Tests for get_user_by_id function"""
def test_get_existing_user(self, db_session):
"""Test getting an existing user"""
upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User",
)
user = get_user_by_id(db_session, "test@example.com")
assert user is not None
assert user.display_name == "Test User"
def test_get_nonexistent_user(self, db_session):
"""Test getting a user that doesn't exist"""
user = get_user_by_id(db_session, "nonexistent@example.com")
assert user is None
class TestGetDisplayName:
"""Tests for get_display_name function"""
def test_get_display_name_existing_user(self, db_session):
"""Test getting display name for existing user"""
upsert_user(
db=db_session,
user_id="test@example.com",
display_name="Test User 測試用戶",
)
name = get_display_name(db_session, "test@example.com")
assert name == "Test User 測試用戶"
def test_get_display_name_nonexistent_user(self, db_session):
"""Test fallback to email for nonexistent user"""
name = get_display_name(db_session, "unknown@example.com")
# Should return email as fallback
assert name == "unknown@example.com"