server/pr_lines.py

"""Build CLEAN PR-style line rows from form values — deterministic pipe format."""

from __future__ import annotations

import calendar
from datetime import date
from typing import Any

INTERVAL_ORDER = [
    "Daily",
    "Weekly",
    "Monthly",
    "Bi-Monthly",
    "Quarterly",
    "Bi-Annual",
    "Annual",
]

INTERVAL_SHORT: dict[str, str] = {
    "Daily": "D",
    "Weekly": "W",
    "Monthly": "M",
    "Bi-Monthly": "BM",
    "Quarterly": "Q",
    "Bi-Annual": "BA",
    "Annual": "A",
}


def _month_end_dates(year: int, months: list[int]) -> list[date]:
    out: list[date] = []
    for m in months:
        last = calendar.monthrange(year, m)[1]
        out.append(date(year, m, last))
    return out


def delivery_dates(count: int, interval: str, year: int) -> list[str]:
    """Return formatted dates like '31 Mar 26' for each delivery line."""
    count = max(1, min(int(count), 52))
    y = int(year)
    dates: list[date] = []

    if interval == "Quarterly":
        # Anchor at March of `year`, then Mar → Jun → Sep → Mar (next year), …
        # Skips December quarter-ends so multi-year schedules reach Q1 of the next year
        # (e.g. 4 deliveries in 2026: Mar, Jun, Sep 2026, then Mar 2027).
        yy = y
        trip = (3, 6, 9)
        i = 0
        while len(dates) < count:
            if i > 0 and i % 3 == 0:
                yy += 1
            m = trip[i % 3]
            last = calendar.monthrange(yy, m)[1]
            dates.append(date(yy, m, last))
            i += 1
    elif interval == "Monthly":
        if count == 1:
            dates.append(date(y, 6, 30))
        elif count <= 12:
            for i in range(count):
                month = min(12, max(1, round(1 + i * 11 / max(count - 1, 1))))
                last = calendar.monthrange(y, month)[1]
                dates.append(date(y, month, last))
        else:
            mm = 1
            yy = y
            while len(dates) < count:
                last = calendar.monthrange(yy, mm)[1]
                dates.append(date(yy, mm, last))
                mm += 1
                if mm > 12:
                    mm = 1
                    yy += 1
    elif interval == "Annual":
        for i in range(count):
            yy = y + i
            dates.append(date(yy, 12, 31))
    elif interval == "Weekly":
        # Approximate: spread week-ending Fridays across the year
        from datetime import timedelta

        start = date(y, 1, 1)
        step = max(1, 364 // max(count, 1))
        for i in range(count):
            d = start + timedelta(days=min(364, i * step))
            dates.append(d)
    else:
        # Bi-Monthly, Daily, Bi-Annual — month-end ladder across the year
        for i in range(count):
            month = min(12, max(1, round((i + 1) * 12 / max(count, 1))))
            last = calendar.monthrange(y, month)[1]
            dates.append(date(y, month, last))

    out: list[str] = []
    for d in dates[:count]:
        out.append(d.strftime("%d %b %y"))
    return out


def _abbrev_dynamic(field_id: str, label: str, value: str | int | float) -> str:
    s = str(value).strip()
    if not s:
        return "—"
    lid = field_id.lower()
    if "sample" in lid or lid.endswith("_size") or "size" in lid:
        nums = "".join(c for c in s if c.isdigit())
        if nums:
            return f"N{nums}"
    if len(s) <= 6 and s.isupper():
        return s
    tok = s.split()[0] if s.split() else s
    return tok[:12].upper()


def _long_part(field_id: str, label: str, value: str | int | float) -> str:
    s = str(value).strip()
    lid = field_id.lower()
    if "sample" in lid or lid.endswith("_size") or "size" in lid:
        nums = "".join(c for c in s if c.isdigit())
        if nums:
            return f"Sample Size {nums}"
    return s


def build_pr_rows(
    *,
    mat_grp: int,
    dynamic_fields_ordered: list[dict[str, Any]],
    dynamic_values: dict[str, Any],
    deliveries: int,
    interval: str,
    other_spec: str,
    year: int,
) -> list[dict[str, Any]]:
    """One row per delivery; SHORT/LONG pipe strings deterministic."""
    iv_short = INTERVAL_SHORT.get(interval, interval[:2].upper())
    iv_long = interval

    dyn_parts_short: list[str] = []
    dyn_parts_long: list[str] = []
    for f in dynamic_fields_ordered:
        fid = f["id"]
        label = f.get("label", fid)
        val = dynamic_values.get(fid, "")
        if val == "" or val is None:
            continue
        dyn_parts_short.append(_abbrev_dynamic(fid, label, val))
        dyn_parts_long.append(_long_part(fid, label, val))

    other_clean = (other_spec or "").strip() or "—"
    dates = delivery_dates(deliveries, interval, year)

    rows: list[dict[str, Any]] = []
    for i in range(deliveries):
        di = i + 1
        d_label = f"D{di}"
        short_core = " | ".join(
            [d_label, iv_short, *dyn_parts_short, other_clean]
        )
        long_core = " | ".join(
            [d_label, iv_long, *dyn_parts_long, other_clean]
        )
        line_no = 10 * di
        date_s = dates[i] if i < len(dates) else dates[-1] if dates else ""
        rows.append(
            {
                "pr_line_item": line_no,
                "mat_grp": mat_grp,
                "pr_short_text": short_core,
                "pr_long_text": long_core,
                "pr_quantity": 1,
                "delivery_date": date_s,
            }
        )
    return rows