climaiq_report_pdf.py

"""
ClimaIQ — Loan officer assessment PDF (A4, branded header, Unicode via Noto fonts).
"""

from __future__ import annotations

import io
import os
import re
import tempfile
import urllib.request
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple

from fpdf import FPDF

PRIMARY = (46, 125, 50)
PRIMARY_DARK = (27, 94, 32)
INK = (31, 42, 32)
MUTED = (90, 109, 102)

NOTO_SANS_URL = (
    "https://raw.githubusercontent.com/googlefonts/noto-fonts/main/hinted/ttf/"
    "NotoSans/NotoSans-Regular.ttf"
)
NOTO_DEVA_URL = (
    "https://raw.githubusercontent.com/googlefonts/noto-fonts/main/hinted/ttf/"
    "NotoSansDevanagari/NotoSansDevanagari-Regular.ttf"
)


def _font_cache_dir() -> str:
    d = os.path.join(tempfile.gettempdir(), "climaiq_fonts")
    os.makedirs(d, exist_ok=True)
    return d


def _ensure_font(url: str, filename: str) -> Optional[str]:
    path = os.path.join(_font_cache_dir(), filename)
    if os.path.isfile(path) and os.path.getsize(path) > 10000:
        return path
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "ClimaIQ-Kisan/1.0"})
        with urllib.request.urlopen(req, timeout=45) as resp:
            data = resp.read()
        with open(path, "wb") as f:
            f.write(data)
        return path if os.path.isfile(path) and os.path.getsize(path) > 10000 else None
    except Exception:
        return None


def _font_path_for_language(language: str) -> Optional[str]:
    if language == "Hindi":
        return _ensure_font(NOTO_DEVA_URL, "NotoSansDevanagari-Regular.ttf")
    return _ensure_font(NOTO_SANS_URL, "NotoSans-Regular.ttf")


def split_report_sections(text: str) -> List[Tuple[Optional[str], str]]:
    """
    Split officer report into (numbered heading or None, body) tuples.
    Headings are lines like '1. Assessment Summary' at the start of a block.
    """
    text = (text or "").strip()
    if not text:
        return []
    matches = list(re.finditer(r"(?m)^\d+\.\s+[^\n]+", text))
    if not matches:
        return [(None, text)]
    out: List[Tuple[Optional[str], str]] = []
    if matches[0].start() > 0:
        pre = text[: matches[0].start()].strip()
        if pre:
            out.append((None, pre))
    for i, m in enumerate(matches):
        start = m.start()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        chunk = text[start:end].strip()
        first_nl = chunk.find("\n")
        if first_nl == -1:
            out.append((chunk.strip(), ""))
        else:
            title = chunk[:first_nl].strip()
            body = chunk[first_nl + 1 :].strip()
            out.append((title, body))
    return out


def _ascii_fallback(s: str) -> str:
    return "".join(c if ord(c) < 128 else "?" for c in str(s))


def build_officer_assessment_pdf(
    report_text: str,
    farmer_data: Dict,
    result: Dict,
    language: str,
) -> bytes:
    font_path = _font_path_for_language(language)
    use_core = font_path is None

    pdf = FPDF(orientation="P", unit="mm", format="A4")
    pdf.set_auto_page_break(auto=True, margin=18)
    pdf.set_margins(18, 18, 18)
    pdf.add_page()

    if font_path and not use_core:
        try:
            pdf.add_font("ClimaIQ", "", font_path)
            pdf.set_font("ClimaIQ", "", 11)
        except Exception:
            use_core = True
    if use_core:
        pdf.set_font("Helvetica", "", 10)

    def set_size(sz: float) -> None:
        if use_core:
            pdf.set_font("Helvetica", "", sz)
        else:
            pdf.set_font("ClimaIQ", "", sz)

    def cell_txt(w: float, h: float, txt: str, **kwargs) -> None:
        t = _ascii_fallback(txt) if use_core else str(txt)
        pdf.cell(w, h, t[:500], **kwargs)

    def multi_txt(w: float, h: float, txt: str) -> None:
        t = _ascii_fallback(txt) if use_core else str(txt)
        pdf.multi_cell(w, h, t)

    # Banner
    pdf.set_fill_color(*PRIMARY)
    pdf.rect(0, 0, 210, 16, "F")
    pdf.set_text_color(255, 255, 255)
    set_size(13)
    pdf.set_xy(18, 5)
    title = "Loan assessment report" if language != "Hindi" else "ऋण मूल्यांकन रिपोर्ट"
    cell_txt(0, 8, title, ln=True)
    pdf.set_text_color(*INK)

    # Meta
    pdf.set_xy(18, 22)
    set_size(9)
    pdf.set_text_color(*MUTED)
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    lang_lbl = "Language" if language != "Hindi" else "भाषा"
    pdf.cell(95, 5, f"{lang_lbl}: {language}" if use_core else f"{lang_lbl}: {language}")
    pdf.cell(0, 5, ts, ln=True)
    pdf.set_text_color(*INK)
    pdf.ln(3)

    # Scorecard title
    set_size(10)
    pdf.set_text_color(*PRIMARY_DARK)
    sc_title = "ClimaIQ scorecard" if language != "Hindi" else "ClimaIQ स्कोरकार्ड"
    cell_txt(0, 6, sc_title, ln=True)
    pdf.set_text_color(*INK)
    pdf.ln(1)

    col_w = (210 - 36) / 2
    pdf.set_fill_color(245, 248, 243)
    set_size(9)

    if language == "Hindi":
        rows = [
            ("क्रेडिट स्कोर", f"{result['credit_score']} / 850"),
            ("जोखिम श्रेणी", str(result["risk_band"])),
            ("डिफ़ॉल्ट संभावना", f"{result['default_probability']}%"),
            ("सुझाई कार्रवाई", str(result["recommended_action"])[:110]),
            ("फसल", str(farmer_data.get("crop_type", ""))),
            ("राज्य", str(farmer_data.get("state", ""))),
            ("ऋण (लाख ₹)", str(farmer_data.get("loan_amount_lakhs", ""))),
            ("बारिश कमी %", str(abs(float(farmer_data.get("rainfall_deficit_pct", 0))))),
            ("SPI", str(farmer_data.get("spi", ""))),
        ]
    else:
        rows = [
            ("Credit score", f"{result['credit_score']} / 850"),
            ("Risk band", str(result["risk_band"])),
            ("Default probability", f"{result['default_probability']}%"),
            ("Recommended action", str(result["recommended_action"])[:110]),
            ("Crop", str(farmer_data.get("crop_type", ""))),
            ("State", str(farmer_data.get("state", ""))),
            ("Loan (Rs lakhs)", str(farmer_data.get("loan_amount_lakhs", ""))),
            ("Rain deficit %", str(abs(float(farmer_data.get("rainfall_deficit_pct", 0))))),
            ("SPI", str(farmer_data.get("spi", ""))),
        ]

    for a, b in rows:
        pdf.set_x(18)
        cell_txt(col_w - 2, 7, a, border="B", fill=True)
        cell_txt(col_w + 2, 7, b, border="B", ln=True, fill=True)
    pdf.ln(4)

    # Narrative
    set_size(10)
    pdf.set_text_color(*PRIMARY)
    narr_h = "AI narrative" if language != "Hindi" else "AI विवरण"
    cell_txt(0, 6, narr_h, ln=True)
    pdf.set_text_color(*INK)
    pdf.ln(2)

    usable_w = 210 - 36
    for head, body in split_report_sections(report_text):
        if head:
            set_size(10.5)
            pdf.set_x(18)
            multi_txt(usable_w, 5.5, head)
            pdf.ln(1)
        if body:
            set_size(10)
            pdf.set_x(18)
            multi_txt(usable_w, 5.2, body)
            pdf.ln(3)

    # Footer
    pdf.set_y(-24)
    pdf.set_draw_color(200, 210, 200)
    yf = pdf.get_y()
    pdf.line(18, yf, 192, yf)
    pdf.ln(2)
    set_size(8)
    pdf.set_text_color(*MUTED)
    foot = (
        "ClimaIQ Kisan — climate-adjusted agricultural credit intelligence. "
        "For decision support only; not a substitute for policy and compliance review."
    )
    if language == "Hindi":
        foot = (
            "ClimaIQ Kisan — कृषि ऋण हेतु जलवायु-समायोजित विश्लेषण। "
            "केवल निर्णय सहायता; नीति व अनुपालन का स्थान नहीं लेता।"
        )
    pdf.set_x(18)
    multi_txt(usable_w, 4, foot)

    out = pdf.output(dest="S")
    if isinstance(out, str):
        return out.encode("latin-1", errors="replace")
    return bytes(out)


def pdf_to_bytesio(report_text: str, farmer_data: Dict, result: Dict, language: str) -> io.BytesIO:
    return io.BytesIO(build_officer_assessment_pdf(report_text, farmer_data, result, language))