| """ |
| Report export — YashTech Ltd. AI Labs. |
| |
| Turns whatever is currently in an output pane into a Markdown file and a PDF. |
| Works mid-stream: the handler reads the component's current value, so a user can |
| export a partial report while generation is still running. |
| |
| No GPU. Never decorate these with @spaces.GPU — rendering a PDF inside a ZeroGPU |
| slot bills the daily quota for CPU work. |
| |
| PDF is optional. If fpdf2 is not installed the Markdown file is still produced |
| and the caller is told, rather than the whole export failing. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import datetime as _dt |
| import pathlib |
| import re |
| import unicodedata |
|
|
| import settings |
| import version |
|
|
| try: |
| from fpdf import FPDF |
|
|
| PDF_AVAILABLE = True |
| except ImportError: |
| PDF_AVAILABLE = False |
|
|
| DISCLAIMER = ( |
| "Research and development use only. MedGemma outputs are not clinically " |
| "validated, are not a medical device, and must not be used for diagnosis or " |
| "patient care without qualified clinician review." |
| ) |
|
|
| |
| |
| _TRANSLITERATE = { |
| "\u2014": "-", "\u2013": "-", "\u2018": "'", "\u2019": "'", |
| "\u201c": '"', "\u201d": '"', "\u2026": "...", "\u2265": ">=", |
| "\u2264": "<=", "\u00d7": "x", "\u2192": "->", "\u26a0": "!", |
| "\u2713": "[x]", "\u2022": "-", |
| } |
|
|
|
|
| def _slug(value: str, fallback: str = "report") -> str: |
| """Filename-safe, no traversal, no surprises from a task id.""" |
| cleaned = re.sub(r"[^A-Za-z0-9_-]+", "-", (value or "")).strip("-").lower() |
| return cleaned[:48] or fallback |
|
|
|
|
| def _latin1(text: str) -> str: |
| for src, dst in _TRANSLITERATE.items(): |
| text = text.replace(src, dst) |
| text = unicodedata.normalize("NFKD", text) |
| return text.encode("latin-1", "replace").decode("latin-1") |
|
|
|
|
| def build_markdown( |
| report: str, |
| task_label: str = "Report", |
| *, |
| patient_lines: list[str] | None = None, |
| task_id: str = "", |
| question: str = "", |
| system_prompt: str = "", |
| space: str = "", |
| model: str = "", |
| partial: bool = False, |
| now: _dt.datetime | None = None, |
| ) -> str: |
| """Markdown with the provenance a report needs to be re-read later.""" |
| |
| |
| space = space or settings.SPACE_ID |
| model = model or settings.MODEL_ID |
| stamp = (now or _dt.datetime.now(_dt.timezone.utc)).strftime("%Y-%m-%d %H:%M:%S UTC") |
| prov = version.provenance() |
| head = [ |
| f"# {task_label}", |
| "", |
| f"- **Generated:** {stamp}", |
| f"- **Space:** `{space}`", |
| f"- **Model:** `{model}` @ `{prov['model_revision'][:12]}`", |
| f"- **Build:** `{prov['build']}`", |
| f"- **Prompt registry:** `{prov['registry_digest']}`", |
| ] |
| if task_id: |
| head.append(f"- **Task:** `{task_id}`") |
| head.extend(patient_lines or []) |
| if partial: |
| head.append("- **Status:** PARTIAL - exported while generation was still running") |
| head += ["", f"> {DISCLAIMER}", "", "---", "", report.strip(), ""] |
|
|
| if question or system_prompt: |
| head += ["---", "", "## Prompt (for audit)", ""] |
| if question: |
| head += ["**Question**", "", f"> {question.strip()}", ""] |
| if system_prompt: |
| head += ["**System prompt**", "", f"> {system_prompt.strip()}", ""] |
| return "\n".join(head) |
|
|
|
|
| def _mc(pdf, height: float, text: str, **kwargs) -> None: |
| """multi_cell that always returns to the left margin on the next line. |
| |
| fpdf2 defaults to new_x=RIGHT/new_y=TOP, so consecutive calls march across |
| the page until it raises "not enough horizontal space". |
| """ |
| pdf.multi_cell(0, height, text, new_x="LMARGIN", new_y="NEXT", **kwargs) |
|
|
|
|
| def _render_pdf(md_text: str, out_path: str, footer_note: str = DISCLAIMER) -> str: |
| class _Doc(FPDF): |
| def footer(self) -> None: |
| self.set_y(-16) |
| self.set_font("Helvetica", "I", 7) |
| _mc(self, 3.5, _latin1(footer_note), align="C") |
| self.set_font("Helvetica", "", 7) |
| self.cell(0, 4, f"Page {self.page_no()}/{{nb}}", align="C") |
|
|
| pdf = _Doc(format="A4") |
| pdf.set_auto_page_break(auto=True, margin=22) |
| pdf.add_page() |
| pdf.set_font("Helvetica", "", 10) |
|
|
| in_code = False |
| for raw in md_text.splitlines(): |
| line = _latin1(raw.rstrip()) |
| if not in_code: |
| line = line.replace("`", "") |
| if line.startswith("```"): |
| in_code = not in_code |
| pdf.set_font("Courier" if in_code else "Helvetica", "", 9 if in_code else 10) |
| continue |
| if in_code: |
| _mc(pdf, 4.2, line or " ") |
| continue |
| if line.startswith("# "): |
| pdf.set_font("Helvetica", "B", 15) |
| _mc(pdf, 8, line[2:]) |
| pdf.set_font("Helvetica", "", 10) |
| elif line.startswith("## "): |
| pdf.ln(2) |
| pdf.set_font("Helvetica", "B", 12) |
| _mc(pdf, 6.5, line[3:]) |
| pdf.set_font("Helvetica", "", 10) |
| elif line.startswith("---"): |
| y = pdf.get_y() + 1 |
| pdf.line(pdf.l_margin, y, pdf.w - pdf.r_margin, y) |
| pdf.ln(3) |
| elif line.startswith("> "): |
| pdf.set_font("Helvetica", "I", 9) |
| _mc(pdf, 4.6, line[2:]) |
| pdf.set_font("Helvetica", "", 10) |
| elif line.startswith(("- ", "* ")): |
| _mc(pdf, 4.8, f" \u00b7 {line[2:].replace('**', '')}") |
| elif not line: |
| pdf.ln(2.5) |
| else: |
| _mc(pdf, 4.8, line.replace("**", "")) |
|
|
| pdf.output(out_path) |
| return out_path |
|
|
|
|
| def write_exports( |
| report: str, |
| out_dir: str | pathlib.Path, |
| task_label: str = "Report", |
| *, |
| patient_lines: list[str] | None = None, |
| filename_tag: str = "", |
| task_id: str = "", |
| question: str = "", |
| system_prompt: str = "", |
| partial: bool = False, |
| want_pdf: bool = True, |
| space: str = "", |
| model: str = "", |
| now: _dt.datetime | None = None, |
| ) -> tuple[list[str], str]: |
| """Write <task>-<stamp>.md (and .pdf). Returns (paths, status message).""" |
| if not (report or "").strip(): |
| return [], "Nothing to export yet - run a task first." |
|
|
| stamp = (now or _dt.datetime.now(_dt.timezone.utc)).strftime("%Y%m%d-%H%M%S") |
| tag = _slug(filename_tag, fallback="") if filename_tag else "" |
| stem = "-".join(part for part in (_slug(task_id or task_label), tag, stamp) if part) |
| out_dir = pathlib.Path(out_dir) |
| out_dir.mkdir(parents=True, exist_ok=True) |
|
|
| md_text = build_markdown( |
| report, |
| task_label, |
| patient_lines=patient_lines, |
| task_id=task_id, |
| question=question, |
| system_prompt=system_prompt, |
| partial=partial, |
| space=space, |
| model=model, |
| now=now, |
| ) |
| md_path = out_dir / f"{stem}.md" |
| md_path.write_text(md_text, encoding="utf-8") |
| paths = [str(md_path)] |
|
|
| if want_pdf and PDF_AVAILABLE: |
| try: |
| paths.append(_render_pdf(md_text, str(out_dir / f"{stem}.pdf"))) |
| except Exception as exc: |
| return paths, f"Markdown ready. PDF failed ({type(exc).__name__}); use the .md." |
| elif want_pdf: |
| return paths, "Markdown ready. Install `fpdf2` for PDF export." |
|
|
| suffix = " (partial - generation was still running)" if partial else "" |
| return paths, f"Exported {len(paths)} file(s){suffix}." |
|
|
