Spaces:
Runtime error
Runtime error
File size: 8,311 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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 | """Map v2 generation results to legacy SectionPayload JSON for the UI."""
from __future__ import annotations
from backend.models.report import GeneratedSection, ReferenceSource, ReportResult
from backend.models.schema import TemplateSchema
_COMPOSITION_NOTES = {
"minimum": "Mapped from past report baseline with in-place fact updates (baseline preservation).",
"medium": "Technical in-place edit on past report baseline with proofread updates.",
"maximum": "Full narrative in-place edit retaining long-form baseline scaffolding.",
}
# Section statuses whose text is a system placeholder, not survey prose.
_UNPOLISHED_STATUSES = frozenset({"NO_RAG_MATCH", "UNASSIGNED"})
def _polish_section_text(text: str, status: str) -> str:
"""Final cleanup pass on a section's draft text (config-gated, non-destructive)."""
from backend.config import settings
if not getattr(settings, "postprocess_enabled", True):
return text or ""
if status in _UNPOLISHED_STATUSES or not text or not text.strip():
return text or ""
from backend.utils.report_postprocessor import polish_report
try:
return polish_report(text)
except ValueError:
return text
def _style_payload(style_profile: object | None) -> dict | None:
if style_profile is None:
return None
if hasattr(style_profile, "to_payload"):
return style_profile.to_payload() # type: ignore[union-attr]
if isinstance(style_profile, dict):
return style_profile
return None
def interference_for_mode(mode: str, interference_level: str | None) -> str:
il = (interference_level or "medium").strip().lower()
if il in _COMPOSITION_NOTES:
return il
if mode == "enhance":
return "maximum"
if mode == "proofread":
return "medium"
return "minimum"
def bullets_to_raw_notes(
template_id: str,
bullets: list[str],
bullets_by_section: dict[str, list[str]] | None = None,
) -> str:
lines: list[str] = []
if bullets_by_section:
for code, items in bullets_by_section.items():
for item in items:
if item.strip():
lines.append(f"{code}: {item.strip()}")
elif bullets:
for item in bullets:
if item.strip():
lines.append(f"{template_id}: {item.strip()}")
return "\n\n".join(lines)
def _paragraph_index_from_chunk_id(chunk_id: str) -> int:
if ":p" not in chunk_id:
return 0
try:
return int(chunk_id.rsplit(":p", 1)[-1])
except ValueError:
return 0
def reference_sources_from_payload(
payload: dict,
schema: TemplateSchema | None = None,
) -> list[ReferenceSource]:
"""Rebuild structured provenance from legacy payload or v2 preview fields."""
stored = payload.get("reference_sources")
if stored:
return [ReferenceSource.model_validate(item) for item in stored]
out: list[ReferenceSource] = []
seen: set[tuple[str, str, int]] = set()
for item in payload.get("provenance") or []:
if not isinstance(item, dict):
continue
filename = str(item.get("filename") or item.get("doc_id") or "").strip()
if filename.startswith("reference:"):
filename = filename.split(":", 1)[-1]
if not filename:
continue
section_id = str(item.get("section_hint") or "").strip().upper()
para = _paragraph_index_from_chunk_id(str(item.get("chunk_id") or ""))
key = (filename, section_id, para)
if key in seen:
continue
seen.add(key)
section_title = ""
if schema and section_id:
sec = schema.get_section(section_id)
if sec:
section_title = sec.title
out.append(
ReferenceSource(
report_filename=filename,
section_id=section_id,
section_title=section_title,
paragraph_index=para,
)
)
return out
def payload_to_generated_section(
section_id: str,
title: str,
payload: dict,
schema: TemplateSchema | None = None,
) -> GeneratedSection:
"""Map a persisted section payload back to a GeneratedSection for DOCX export."""
ref_sources = reference_sources_from_payload(payload, schema)
# rag_sources are live-generation display strings; do NOT synthesize them
# from provenance here, so reports rebuilt from provenance-only payloads omit
# the internal source attribution footnote on export.
rag_sources = list(payload.get("rag_sources") or [])
status = str(payload.get("status") or "OK")
grounding = payload.get("grounding_passed")
if grounding is None:
grounding = status == "OK" and bool((payload.get("text") or "").strip())
citation = payload.get("citation_audit") if isinstance(payload.get("citation_audit"), dict) else {}
unmatched = list(payload.get("unmatched_observations") or citation.get("dropped_claims") or [])
ai = payload.get("ai_transparency") if isinstance(payload.get("ai_transparency"), dict) else {}
return GeneratedSection(
section_id=section_id,
title=title,
text=payload.get("text") or "",
rating_value=payload.get("rating_value"),
status=status,
notes=str(payload.get("notes") or ai.get("plain_language") or ""),
rag_sources=rag_sources,
reference_sources=ref_sources,
grounding_passed=bool(grounding),
unmatched_observations=unmatched,
)
def section_to_payload(
section: GeneratedSection,
*,
interference_level: str,
mode: str,
style_profile: object | None = None,
) -> dict:
composition_note = _COMPOSITION_NOTES.get(
interference_level,
_COMPOSITION_NOTES["medium"],
)
provenance = [
{
"doc_id": src.report_filename,
"chunk_id": f"{src.section_id}:p{src.paragraph_index or 0}",
"score": 1.0,
"filename": src.report_filename,
"snippet_preview": (
f"Section {src.section_id}"
+ (f", paragraph {src.paragraph_index}" if src.paragraph_index else "")
)[:240],
"section_hint": src.section_id,
}
for src in section.reference_sources
]
confidence = 0.92 if section.status == "OK" and section.grounding_passed else 0.55
if section.status == "NO_RAG_MATCH":
confidence = 0.2
polished_text = _polish_section_text(section.text or "", section.status)
return {
"text": polished_text,
"confidence": confidence,
"provenance": provenance,
"reference_sources": [rs.model_dump() for rs in section.reference_sources],
"rag_sources": list(section.rag_sources),
"status": section.status,
"grounding_passed": section.grounding_passed,
"unmatched_observations": list(section.unmatched_observations),
"rating_value": section.rating_value,
"cached": False,
"mode": mode if mode in ("generate", "proofread", "enhance") else "generate",
"style_profile": _style_payload(style_profile),
"interference_level": interference_level,
"composition_depth": interference_level,
"ai_transparency": {
"plain_language": section.notes or composition_note,
"composition_note": composition_note,
},
"photos": [],
"pipeline": "v2_reference_mapping",
"citation_audit": {
"confidence": confidence,
"findings": len(provenance),
"contradictions": [],
"dropped_claims": section.unmatched_observations or [],
},
"word_count": len(polished_text.split()),
}
def result_to_sections_payload(
result: ReportResult,
*,
interference_level: str,
mode: str,
style_profile: object | None = None,
) -> dict[str, dict]:
out: dict[str, dict] = {}
for section in result.sections:
if section.section_id == "UNASSIGNED":
continue
out[section.section_id] = section_to_payload(
section,
interference_level=interference_level,
mode=mode,
style_profile=style_profile,
)
return out
|