Spaces:
Runtime error
Runtime error
File size: 3,708 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Assemble a generated report into output formats (JSON preview + DOCX)."""
from __future__ import annotations
from backend.config import settings
from backend.core.compat_adapter import payload_to_generated_section
from backend.models.report import GeneratedSection, ReportResult
from backend.models.schema import TemplateSchema
from backend.utils import docx_builder
_REVIEW_STATUSES = frozenset(
{"NO_RAG_MATCH", "GROUNDING_REVIEW", "UNASSIGNED", "NOTES_ONLY"}
)
def to_preview(result: ReportResult, schema: TemplateSchema) -> dict:
"""Schema-ordered JSON preview for the UI."""
sections_out = []
for s in sorted(
result.sections,
key=lambda s: schema.section_ids().index(s.section_id)
if s.section_id in schema.section_ids() else 1_000_000,
):
sections_out.append({
"section_id": s.section_id,
"title": s.title,
"text": s.text,
"paragraph": s.text,
"rating_value": s.rating_value,
"status": s.status,
"notes": s.notes,
"rag_sources": s.rag_sources,
"reference_sources": [rs.model_dump() for rs in s.reference_sources],
"grounding_passed": s.grounding_passed,
"unmatched_observations": s.unmatched_observations,
})
mapped_ok = sum(1 for s in result.sections if s.status == "OK")
needing_review = sum(1 for s in result.sections if s.status in _REVIEW_STATUSES)
total = len(result.sections) or 1
review_ratio = needing_review / total
manual_review_required = review_ratio > settings.grounding_alert_threshold
return {
"tenant_id": result.tenant_id,
"schema_version": result.schema_version,
"schema_report_type": schema.report_type,
"rating_system": schema.rating_system.model_dump(),
"sections_mapped": mapped_ok,
"sections_needing_review": needing_review,
"review_ratio": round(review_ratio, 4),
"manual_review_required": manual_review_required,
"sections": sections_out,
"unassigned_text": result.unassigned_text,
"property_type": result.property_type,
"tenure": result.tenure,
}
def from_sections_payload(
tenant_id: str,
schema: TemplateSchema,
sections_payload: dict[str, dict],
*,
property_type: str = "",
tenure: str = "",
unassigned_text: str = "",
) -> ReportResult:
"""Rebuild a ReportResult from persisted section payloads (legacy session or preview)."""
sections: list[GeneratedSection] = []
for sec in schema.ordered_sections():
payload = sections_payload.get(sec.id)
if not payload or not (payload.get("text") or "").strip():
continue
sections.append(
payload_to_generated_section(sec.id, sec.title, payload, schema)
)
return ReportResult(
tenant_id=tenant_id,
schema_version=schema.version,
property_type=property_type,
tenure=tenure,
sections=sections,
unassigned_text=unassigned_text,
)
def to_docx(
result: ReportResult,
schema: TemplateSchema,
*,
title: str = "Survey Report",
include_footer: bool = False,
template_docx_path: str | None = None,
section_photo_paths: dict[str, list[str]] | None = None,
) -> bytes:
path = template_docx_path
if path is None and settings.branded_template_path is not None:
path = str(settings.branded_template_path)
return docx_builder.build_docx(
result,
schema,
title=title,
include_footer=include_footer,
template_docx_path=path,
section_photo_paths=section_photo_paths,
)
|