PacketStation/server/packetstation/domain/pm.py

"""Project Management: see everything, change nothing, report on it. A port of the old pm.js.

Every report is built from one card list (list_cards over the find bar's scope, sorted by project) plus a few
scoped reads: the stage clocks, the work-order target and ticket hours, the transitions, Q.C. history, the open
shorts and change orders. Nothing here reads VISUAL; the ticket hours come from the visual_labor mirror the feed
keeps. Nothing here writes, except the export log the API adds.

Shapes:
    scope = {label, count, period: {value, label, start}, project: {id, number, name} | None, unit: {id, number} | None}
    table = {key, title, note, head, rows}
    row = {cells, pid, project_id, unit_id, lvl, health, flag}
"""

from __future__ import annotations

import re
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field
from datetime import UTC, date, datetime, timedelta
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from .. import policy
from . import packets, search

DAY_S = 86400.0
CARD_LIMIT = 20000

PERIODS: dict[str, tuple[str, int | None]] = {
    "30": ("Last 30 days", 30),
    "90": ("Last 90 days", 90),
    "365": ("Last 12 months", 365),
    "all": ("All time", None),
}
DEFAULT_PERIOD = "90"

PM_STEPS: tuple[tuple[str, str], ...] = (
    ("draft", "Ready to release (Clerical)"),
    ("wh_incoming", "Warehouse: waiting"),
    ("wh_pulling", "Warehouse: being pulled"),
    ("wh_hold", "Warehouse: on hold"),
    ("wh_pulled", "Warehouse: pulled, ready for production"),
    ("ready", "Ready to assign"),
    ("station", "At a station"),
    ("review", "Supervisor review"),
    ("qc", "At Q.C."),
    ("complete", "Back for data book"),
)
PM_STEP_N = dict(PM_STEPS)

HEALTH: dict[str, tuple[str, str]] = {
    "ok": ("On track", "st-in_progress"),
    "risk": ("At risk", "st-warehouse"),
    "late": ("Late", "st-review"),
    "done": ("Complete", "st-filed"),
    "empty": ("No packets yet", "st-draft"),
}
SEVERITY = {"late": 4, "risk": 3, "ok": 2, "done": 1, "empty": 0}

SENT_BACK = ("return_to_clerical", "wh_send_back", "send_back_to_warehouse", "return_to_station", "qc_reject")
HOURS_STATUSES = ("review", "qc", "complete", "filed")
MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
DASH = "—"

# Table keys, in the order of the "all" export.
TABLE_KEYS = (
    "projects",
    "aging_steps",
    "aging_open",
    "aging_history",
    "hours_dept",
    "hours_worker",
    "hours_project",
    "hours_over",
    "quality_rejects",
    "quality_shorts",
    "quality_cos",
    "quality_holds",
    "quality_sentback",
    "jobs",
)

NOTE_PROJECTS = (
    "Tap a project to show its units. Done = accepted by Q.C. or filed. At risk = due within 7 days and under 75%"
    " done, or a hold or open change order."
)
NOTE_AGING_STEPS = "How long packets have been sitting where they are right now."
NOTE_AGING_OPEN = "Over 3 days in one step is flagged."
NOTE_AGING_HISTORY = "From the step history recorded on every move."
NOTE_HOURS_OVER = (
    "VISUAL ticket hours against the work-order target hours decide over target. The station timer is shown beside"
    " them. A packet with no ticket hours is not over target."
)

PROJECTS_HEAD = [
    "Project / unit",
    "Name",
    "Customer",
    "Due",
    "Left",
    "Units",
    "Packets",
    "Draft",
    "Warehouse",
    "Shop",
    "Q.C.",
    "Done",
    "% done",
    "Health",
]
ATTENTION_HEAD = ["Unit", "Description", "Due", "Left", "Packets", "% done", "Health"]
HOURS_HEAD = [
    "Packets",
    "Timer h",
    "Ticket h",
    "Target h",
    "Ticket variance h",
    "Ticket h / target",
    "Over target (tickets)",
]
OVER_HEAD = ["Job", "Station / worker", "Status", "Timer h", "Ticket h", "Target h", "Over by (tickets)"]
JOBS_HEAD = [
    "Job",
    "Project",
    "Unit",
    "Operation",
    "Title",
    "Department",
    "Step",
    "With",
    "Station",
    "Worker",
    "Drawing",
    "Rev",
    "Unit due",
    "Days in step",
    "Timer h",
    "Ticket h",
    "Target h",
    "Created",
    "Updated",
    "Change orders open",
    "Shorts open",
]


# ---------------------------------------------------------------- small texts


def half_up(x: float) -> int:
    """Math.round: halves go up."""
    return int(x + 0.5) if x >= 0 else -int(-x + 0.5)


def fmt_days(seconds: float) -> str:
    """The old fmtDays: under a day in whole hours, then days with one decimal under 10 and whole above."""
    d = seconds / DAY_S
    if d < 1:
        return f"{max(0, half_up(seconds / 3600))} h"
    return f"{d:.1f} d" if d < 10 else f"{half_up(d)} d"


def fmt_h(hours: float) -> str:
    """Hours with one decimal (the old fmtH)."""
    v = round(hours, 1)
    return f"{0.0 if v == 0 else v:.1f}"


def signed_h(hours: float) -> str:
    v = round(hours, 1)
    return ("+" if v >= 0 else "") + fmt_h(v)


def day_str(d: Any) -> str:
    """Oct 20 '26 (the old dayStr); empty for no date."""
    if not isinstance(d, date):
        return ""
    return f"{MONTHS[d.month - 1]} {d.day} '{str(d.year)[2:]}"


def left_text(days_left: int | None) -> str:
    if days_left is None:
        return DASH
    return f"{-days_left} d over" if days_left < 0 else f"{days_left} d"


def due_text(due: Any, today: date, closed: bool = False) -> str:
    """The due label without "Due ": "Oct 20 '26 · 3 d left", "· 2 d overdue", "· today"."""
    if not isinstance(due, date):
        return DASH
    lab = day_str(due)
    if closed:
        return lab
    days = (due - today).days
    if days < 0:
        return f"{lab} · {-days} d overdue"
    if days <= 7:
        return f"{lab} · {'today' if days == 0 else f'{days} d left'}"
    return f"{lab} · {days} d left"


def plural(n: int, word: str) -> str:
    return f"{n} {word}{'' if n == 1 else 's'}"


def job_name(c: Mapping[str, Any]) -> str:
    unit = (c.get("unit") or {}).get("number") or ""
    return f"{unit} · {c['op']}" if c.get("op") else str(unit)


def zone(tz: str | None) -> Any:
    try:
        return ZoneInfo(tz or "America/Denver")
    except (ZoneInfoNotFoundError, ValueError):
        return UTC


def as_dt(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=UTC)
    if isinstance(value, str) and value.strip():
        try:
            d = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
        except ValueError:
            return None
        return d if d.tzinfo else d.replace(tzinfo=UTC)
    return None


def fmt_dt(value: Any, tz: str | None) -> str:
    """M/D/YYYY h:mm AM in the site time zone; a dash for no value."""
    d = as_dt(value)
    if d is None:
        return DASH
    d = d.astimezone(zone(tz))
    h = d.hour % 12 or 12
    return f"{d.month}/{d.day}/{d.year} {h}:{d.minute:02d} {'AM' if d.hour < 12 else 'PM'}"


def fmt_d(value: Any, tz: str | None) -> str:
    d = as_dt(value)
    if d is None:
        return value.isoformat() if isinstance(value, date) else ""
    d = d.astimezone(zone(tz))
    return f"{d.month}/{d.day}/{d.year}"


def row(
    cells: Sequence[Any],
    *,
    pid: int | None = None,
    project_id: int | None = None,
    unit_id: int | None = None,
    lvl: int = 0,
    health: str | None = None,
    flag: bool = False,
) -> dict[str, Any]:
    return {
        "cells": ["" if c is None else str(c) for c in cells],
        "pid": pid,
        "project_id": project_id,
        "unit_id": unit_id,
        "lvl": lvl,
        "health": health,
        "flag": flag,
    }


def table(key: str, title: str, head: Sequence[str], rows: list[dict[str, Any]], note: str | None = None) -> dict:
    return {"key": key, "title": title, "note": note, "head": list(head), "rows": rows}


# ---------------------------------------------------------------- facts about one packet


def step_of(c: Mapping[str, Any]) -> str:
    st = str(c["status"])
    if st == "warehouse":
        return "wh_" + str(c.get("wh_stage") or "incoming")
    if st in policy.ACTIVE:
        return "station"
    return st


def step_since(c: Mapping[str, Any]) -> datetime:
    wh = c.get("wh") or {}
    if c["status"] == "warehouse" and wh.get("at") is not None:
        return as_dt(wh["at"]) or datetime.now(UTC)
    return as_dt(c.get("status_at")) or as_dt(c.get("created_at")) or datetime.now(UTC)


def step_label(c: Mapping[str, Any]) -> str:
    if c["status"] == "filed":
        return "Filed"
    return PM_STEP_N.get(step_of(c), str(c["status"]))


def on_hold(c: Mapping[str, Any]) -> bool:
    return c["status"] == "warehouse" and c.get("wh_stage") == "hold"


def risky(c: Mapping[str, Any]) -> bool:
    return bool(c.get("open_cos")) or on_hold(c)


def health_of(cards: Sequence[Mapping[str, Any]], due: date | None, today: date) -> dict[str, Any]:
    """The old healthOf: n, filed, done, pct, days_left, health, due."""
    n = len(cards)
    filed = sum(1 for c in cards if c["status"] == "filed")
    done = sum(1 for c in cards if c["status"] in ("complete", "filed"))
    days_left = (due - today).days if isinstance(due, date) else None
    if not n:
        h = "empty"
    elif filed == n:
        h = "done"
    elif days_left is not None and days_left < 0:
        h = "late"
    elif (days_left is not None and days_left <= 7 and done / n < 0.75) or any(risky(c) for c in cards):
        h = "risk"
    else:
        h = "ok"
    return {
        "n": n,
        "filed": filed,
        "done": done,
        "pct": half_up(done / n * 100) if n else 0,
        "days_left": days_left,
        "health": h,
        "due": due,
    }


def stage_buckets(cards: Iterable[Mapping[str, Any]]) -> dict[str, int]:
    b = {"draft": 0, "wh": 0, "shop": 0, "qc": 0, "done": 0}
    for c in cards:
        s = c["status"]
        if s == "draft":
            b["draft"] += 1
        elif s == "warehouse":
            b["wh"] += 1
        elif s in policy.SHOP:
            b["shop"] += 1
        elif s == "qc":
            b["qc"] += 1
        else:
            b["done"] += 1
    return b


# ---------------------------------------------------------------- the report context


HOURS_SQL = """
SELECT p.id,
       coalesce((SELECT sum(s.clock_elapsed_ms
                            + CASE WHEN s.clock_running_since IS NOT NULL
                                   THEN greatest(0, extract(epoch FROM (%(now)s - s.clock_running_since)) * 1000)
                                   ELSE 0 END)
                 FROM packet_stages s WHERE s.packet_id = p.id), 0) AS timer_ms,
       coalesce((SELECT sum(w.target_hrs) FROM packet_work_orders w WHERE w.packet_id = p.id), 0) AS target_hrs,
       CASE WHEN p.source_kind = 'visual' THEN
           (SELECT sum(l.hours) FROM packet_work_orders w
              JOIN visual_labor l ON l.wo_base = w.wo_base AND l.lot = w.lot AND l.split = w.split
                                 AND l.sub = w.sub AND l.seq = w.seq
             WHERE w.packet_id = p.id)
       END AS ticket_hrs
FROM packets p WHERE p.id = ANY(%(ids)s)
"""

TRANSITIONS_SQL = """
SELECT t.packet_id, t.at, t.action, t.from_status::text AS from_status, t.to_status::text AS to_status, t.reason,
       u.display_name AS by_name
FROM packet_transitions t LEFT JOIN users u ON u.id = t.by
WHERE t.packet_id = ANY(%s)
ORDER BY t.packet_id, t.at, t.id
"""

REJECTS_SQL = """
SELECT h.packet_id, h.at, h.inspector, h.note, h.snapshot->>'result_at' AS result_at, h.snapshot->>'ncr' AS ncr
FROM qc_history h WHERE h.packet_id = ANY(%s) AND h.result = 'rejected'
"""
OPEN_REJECTS_SQL = """
SELECT q.packet_id, coalesce(q.result_at, q.sent_at) AS at, q.inspector, q.result_note AS note, NULL AS result_at,
       q.ncr
FROM qc_inspections q WHERE q.packet_id = ANY(%s) AND q.closed_at IS NULL AND q.result = 'rejected'
"""

SHORTS_SQL = """
SELECT s.id, s.packet_id, s.qty_open, s.reason, s.ref, s.eta, s.opened_at, b.item, b.description
FROM bom_shorts s JOIN bom_lines b ON b.id = s.bom_line_id
WHERE s.closed_at IS NULL AND s.packet_id = ANY(%s)
ORDER BY s.opened_at, s.id
"""

COS_SQL = """
SELECT c.packet_id, c.co_id, c.co_no, c.pending_depts, co.origin::text AS origin, co.review::text AS review, co.ecn,
       co.reason, co.vis, co.issued_at
FROM v_open_cos c JOIN change_orders co ON co.id = c.co_id
WHERE c.packet_id = ANY(%s)
ORDER BY co.issued_at NULLS LAST, c.packet_id, c.co_no
"""


@dataclass
class Report:
    """Everything one PM answer needs, read once."""

    conn: Any
    filters: dict[str, Any]
    period: str
    now: datetime
    tz: str
    today: date
    start: datetime | None
    cards: list[dict[str, Any]]
    projects: list[dict[str, Any]]
    units: list[dict[str, Any]]
    dept_labels: dict[str, str]
    expand: set[int] | None = None  # None: every project open (expand=all)
    _hours: dict[int, dict[str, float]] | None = field(default=None, repr=False)
    _transitions: dict[int, list[dict[str, Any]]] | None = field(default=None, repr=False)
    _projects_by_id: dict[int, dict[str, Any]] = field(default_factory=dict, repr=False)
    _units_by_id: dict[int, dict[str, Any]] = field(default_factory=dict, repr=False)

    def __post_init__(self) -> None:
        self._projects_by_id = {int(p["id"]): p for p in self.projects}
        self._units_by_id = {int(u["id"]): u for u in self.units}

    # ------------------------------------------------------------ lookups

    @property
    def ids(self) -> list[int]:
        return [int(c["id"]) for c in self.cards]

    @property
    def open_cards(self) -> list[dict[str, Any]]:
        return [c for c in self.cards if c["status"] != "filed"]

    @property
    def period_label(self) -> str:
        return PERIODS[self.period][0]

    def project_by_id(self, pid: int | None) -> dict[str, Any] | None:
        return None if pid is None else self._projects_by_id.get(int(pid))

    def unit_by_id(self, uid: int | None) -> dict[str, Any] | None:
        return None if uid is None else self._units_by_id.get(int(uid))

    def hours(self) -> dict[int, dict[str, float]]:
        """Per packet: timer_h (every stage's clock, the running span included), ticket_h (VISUAL ticket hours on
        its work orders, 0 when none) and target_h (the work orders' target hours)."""
        if self._hours is None:
            out: dict[int, dict[str, float]] = {}
            if self.cards:
                for r in packets.rows(self.conn, HOURS_SQL, {"now": self.now, "ids": self.ids}):
                    out[int(r["id"])] = {
                        "timer_h": float(r["timer_ms"] or 0) / 3_600_000,
                        "ticket_h": float(r["ticket_hrs"] or 0),
                        "target_h": float(r["target_hrs"] or 0),
                    }
            self._hours = out
        return self._hours

    def transitions(self) -> dict[int, list[dict[str, Any]]]:
        if self._transitions is None:
            out: dict[int, list[dict[str, Any]]] = {}
            if self.cards:
                for r in packets.rows(self.conn, TRANSITIONS_SQL, (self.ids,)):
                    out.setdefault(int(r["packet_id"]), []).append(r)
            self._transitions = out
        return self._transitions

    def in_period(self, at: Any) -> bool:
        d = as_dt(at)
        return d is not None and (self.start is None or d >= self.start)

    def unit_due(self, c: Mapping[str, Any]) -> date | None:
        d = c.get("due_effective")
        return d if isinstance(d, date) else None

    def unit_closed(self, c: Mapping[str, Any]) -> bool:
        u = self.unit_by_id(int(c["unit"]["id"]))
        if u is None:
            return False
        return u["status"] == "closed" or u["project_status"] == "closed"

    def dept_short(self, code: str) -> str:
        return self.dept_labels.get(code, code)

    # ------------------------------------------------------------ scope

    def scope_label(self) -> str:
        f = self.filters
        parts: list[str] = []
        u = self.unit_by_id(f.get("unit_id"))
        pr = self.project_by_id(f.get("project_id"))
        if u is not None:
            parts.append(f"Unit {u['number']}")
        elif pr is not None:
            parts.append(f"Project {pr['number']}")
        else:
            parts.append("All projects")
        if f.get("dept"):
            parts.append(self.dept_short(str(f["dept"])))
        if f.get("status"):
            st = str(f["status"])
            parts.append(search.FLT_STATUS_LABELS.get(st, st))
        q = str(f.get("q") or "").strip()
        if q:
            parts.append(f'search "{q}"')
        flags = [x for x in f.get("flags") or [] if x != "mine"]
        if flags:
            parts.append(", ".join(flags))
        return " · ".join(parts)

    def scope(self) -> dict[str, Any]:
        f = self.filters
        pr = self.project_by_id(f.get("project_id"))
        u = self.unit_by_id(f.get("unit_id"))
        if pr is None and u is not None:
            pr = self.project_by_id(u["project_id"])
        return {
            "label": self.scope_label(),
            "count": len(self.cards),
            "period": {
                "value": self.period,
                "label": self.period_label,
                "start": self.start.isoformat() if self.start else None,
            },
            "project": None if pr is None else {"id": pr["id"], "number": pr["number"], "name": pr["name"]},
            "unit": None if u is None else {"id": u["id"], "number": u["number"]},
        }


def period_of(value: Any) -> str:
    v = str(value or "").strip().lower()
    return v if v in PERIODS else DEFAULT_PERIOD


def load(
    conn: Any,
    pp: policy.Principal | None,
    params: Mapping[str, Any],
    *,
    now: datetime | None = None,
    tz: str | None = None,
    period: Any = None,
    expand: str | None = None,
) -> Report:
    """Read the scope once. `params` are the find bar values plus project_id and unit_id; Mine is ignored here (the
    chip is hidden on the PM screen)."""
    now = (now or datetime.now(UTC)).astimezone(UTC)
    tz = tz or "America/Denver"
    today = now.astimezone(zone(tz)).date()
    principal = pp or policy.Principal(user_id=None, pm=True)
    filters = search.parse_filters(params, principal, screen="pm", today=today)
    filters["flags"] = [f for f in filters["flags"] if f != "mine"]
    filters["mine"] = None
    per = period_of(period)
    days = PERIODS[per][1]
    start = None if days is None else now - timedelta(days=days)
    cards = packets.list_cards(conn, principal, {**filters, "sort": "project", "limit": CARD_LIMIT, "lite": True})
    projects = packets.rows(
        conn,
        "SELECT id, number, name, customer, due, status::text AS status FROM projects WHERE removed_at IS NULL"
        ' ORDER BY number COLLATE "C", id',
    )
    units = packets.rows(
        conn,
        "SELECT u.id, u.project_id, u.lot, pr.number || u.lot AS number, u.name, u.due, u.status::text AS status,"
        " pr.status::text AS project_status FROM units u JOIN projects pr ON pr.id = u.project_id"
        ' WHERE u.removed_at IS NULL ORDER BY u.lot COLLATE "C", u.id',
    )
    rep = Report(
        conn=conn,
        filters=filters,
        period=per,
        now=now,
        tz=tz,
        today=today,
        start=start,
        cards=cards,
        projects=projects,
        units=units,
        dept_labels=packets.dept_labels(conn),
    )
    rep.expand = parse_expand(expand)
    return rep


def parse_expand(expand: str | None) -> set[int] | None:
    """`all` opens every project (None); a comma list of project ids opens those."""
    text = str(expand or "").strip().lower()
    if text == "all":
        return None
    out: set[int] = set()
    for part in text.split(","):
        part = part.strip()
        if part.isdigit():
            out.add(int(part))
    return out


# ---------------------------------------------------------------- projects and units


def _units_by_project(rep: Report) -> dict[int, list[dict[str, Any]]]:
    out: dict[int, list[dict[str, Any]]] = {}
    for u in rep.units:
        out.setdefault(int(u["project_id"]), []).append(u)
    return out


def projects_table(rep: Report, *, expand_all: bool = False) -> dict[str, Any]:
    """The old rptProjects."""
    f = rep.filters
    any_filter = search.narrowing(f) or f.get("project_id") is not None or f.get("unit_id") is not None
    by_project: dict[int, list[dict[str, Any]]] = {}
    by_unit: dict[int, list[dict[str, Any]]] = {}
    for c in rep.cards:
        by_project.setdefault(int(c["project"]["id"]), []).append(c)
        by_unit.setdefault(int(c["unit"]["id"]), []).append(c)
    picked = f.get("project_id")
    units_of = _units_by_project(rep)
    rows: list[dict[str, Any]] = []
    for t in rep.projects:
        tid = int(t["id"])
        cards = by_project.get(tid, [])
        if t["status"] == "closed" and not cards:
            continue
        if any_filter and not cards:
            continue
        hl = health_of(cards, t["due"], rep.today)
        b = stage_buckets(cards)
        us = units_of.get(tid, [])
        for u in us:
            uc = by_unit.get(int(u["id"]), [])
            if not uc:
                continue
            uh = health_of(uc, u["due"] or t["due"], rep.today)
            if SEVERITY[uh["health"]] > SEVERITY[hl["health"]] and hl["health"] != "empty":
                hl["health"] = uh["health"]
        rows.append(
            row(
                [
                    t["number"],
                    t["name"] or "",
                    t["customer"] or "",
                    day_str(t["due"]) or DASH,
                    left_text(hl["days_left"]),
                    len(us),
                    hl["n"],
                    b["draft"],
                    b["wh"],
                    b["shop"],
                    b["qc"],
                    b["done"],
                    f"{hl['pct']}%",
                    HEALTH[hl["health"]][0],
                ],
                project_id=tid,
                lvl=0,
                health=hl["health"],
            )
        )
        opened = expand_all or rep.expand is None or tid in rep.expand or picked == tid
        if not opened:
            continue
        for u in us:
            uc = by_unit.get(int(u["id"]), [])
            if any_filter and not uc and picked != tid:
                continue
            due = u["due"] or t["due"]
            uh = health_of(uc, due, rep.today)
            ub = stage_buckets(uc)
            due_txt = (day_str(due) + ("" if u["due"] else " (project)")) if due else DASH
            rows.append(
                row(
                    [
                        u["number"],
                        u["name"] or "",
                        "",
                        due_txt,
                        left_text(uh["days_left"]),
                        "",
                        uh["n"],
                        ub["draft"],
                        ub["wh"],
                        ub["shop"],
                        ub["qc"],
                        ub["done"],
                        f"{uh['pct']}%",
                        HEALTH[uh["health"]][0],
                    ],
                    project_id=tid,
                    unit_id=int(u["id"]),
                    lvl=1,
                    health=uh["health"],
                )
            )
    return table("projects", "Projects and units", PROJECTS_HEAD, rows, NOTE_PROJECTS)


def unit_healths(rep: Report) -> list[dict[str, Any]]:
    """Health of every unit with packets in scope (the KPIs count them)."""
    by_unit: dict[int, list[dict[str, Any]]] = {}
    for c in rep.cards:
        by_unit.setdefault(int(c["unit"]["id"]), []).append(c)
    out: list[dict[str, Any]] = []
    for uid, cards in by_unit.items():
        u = rep.unit_by_id(uid)
        t = rep.project_by_id(u["project_id"]) if u else None
        due = (u["due"] if u else None) or (t["due"] if t else None)
        out.append(health_of(cards, due, rep.today))
    return out


def attention_table(rep: Report) -> dict[str, Any]:
    """The units late or at risk, from the projects table with every unit open."""
    rows = []
    for r in projects_table(rep, expand_all=True)["rows"]:
        if r["lvl"] == 1 and r["health"] in ("late", "risk"):
            c = r["cells"]
            rows.append(
                row(
                    [c[0], c[1], c[3], c[4], c[6], c[12], c[13]],
                    project_id=r["project_id"],
                    unit_id=r["unit_id"],
                    lvl=1,
                    health=r["health"],
                )
            )
    return table("attention", "Units late or at risk", ATTENTION_HEAD, rows)


# ---------------------------------------------------------------- aging


def pipeline(rep: Report) -> list[dict[str, Any]]:
    """One entry per PM step: the open packets in it, the average and longest time there, the oldest packet, and
    `slowest` on the step with the highest average among the steps with packets."""
    out: list[dict[str, Any]] = []
    open_cards = rep.open_cards
    for key, label in PM_STEPS:
        here = [c for c in open_cards if step_of(c) == key]
        ages = [(rep.now - step_since(c)).total_seconds() for c in here]
        oldest = min(here, key=lambda c: step_since(c)) if here else None
        out.append(
            {
                "key": key,
                "label": label,
                "count": len(here),
                "avg_s": round(sum(ages) / len(ages), 1) if ages else 0,
                "max_s": round(max(ages), 1) if ages else 0,
                "oldest": None if oldest is None else {"pid": oldest["id"], "job": job_name(oldest)},
                "slowest": False,
            }
        )
    busy = [e for e in out if e["count"]]
    if busy:
        max(busy, key=lambda e: e["avg_s"])["slowest"] = True
    return out


def aging_tables(rep: Report, pipe: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
    """The old rptAging: time in the current step, every open packet, and the completed stays in the period."""
    pipe = pipe if pipe is not None else pipeline(rep)
    steps = table(
        "aging_steps",
        "Time in the current step (open packets)",
        ["Step", "Packets", "Average", "Longest", "Oldest packet"],
        [
            row(
                [
                    s["label"],
                    s["count"],
                    fmt_days(s["avg_s"]) if s["count"] else DASH,
                    fmt_days(s["max_s"]) if s["count"] else DASH,
                    s["oldest"]["job"] if s["oldest"] else DASH,
                ],
                pid=s["oldest"]["pid"] if s["oldest"] else None,
            )
            for s in pipe
        ],
        NOTE_AGING_STEPS,
    )

    open_rows = []
    for c in sorted(rep.open_cards, key=lambda c: (step_since(c), c["id"])):
        age = (rep.now - step_since(c)).total_seconds()
        created = as_dt(c.get("created_at")) or step_since(c)
        flags = [
            "Change order" if c.get("open_cos") else "",
            f"{c['open_shorts']} short" if c.get("open_shorts") else "",
            "On hold: " + str((c.get("wh") or {}).get("reason") or "") if on_hold(c) else "",
        ]
        u = rep.unit_by_id(int(c["unit"]["id"]))
        open_rows.append(
            row(
                [
                    job_name(c),
                    (u["name"] if u else "") or "",
                    step_label(c),
                    c.get("holder") or "",
                    fmt_days(age),
                    fmt_days((rep.now - created).total_seconds()),
                    due_text(rep.unit_due(c), rep.today, rep.unit_closed(c)),
                    " · ".join(x for x in flags if x),
                ],
                pid=c["id"],
                flag=age > 3 * DAY_S,
            )
        )
    every = table(
        "aging_open",
        "Every open packet, longest waiting first",
        ["Job", "Unit", "Step", "With", "In this step", "Total age", "Unit due", "Flags"],
        open_rows,
        NOTE_AGING_OPEN,
    )

    done: dict[str, list[float]] = {}
    trans = rep.transitions()
    for c in rep.cards:
        prev = as_dt(c.get("created_at"))
        for t in trans.get(int(c["id"]), []):
            at = as_dt(t["at"])
            if at is None:
                continue
            if prev is not None and rep.in_period(at) and t["from_status"]:
                done.setdefault(str(t["from_status"]), []).append(max(0.0, (at - prev).total_seconds()))
            prev = at
    order = {s: i for i, s in enumerate(policy.STATUSES)}
    hist_rows = [
        row(
            [
                search.STATUS_LABEL.get(k, k),
                len(a),
                fmt_days(sum(a) / len(a)),
                fmt_days(max(a)),
            ]
        )
        for k, a in sorted(done.items(), key=lambda kv: order.get(kv[0], 99))
    ]
    history = table(
        "aging_history",
        f"Average time per step, moves completed ({rep.period_label})",
        ["Step", "Moves", "Average", "Longest"],
        hist_rows,
        NOTE_AGING_HISTORY,
    )
    return [steps, every, history]


# ---------------------------------------------------------------- hours


def _counts_in_hours(c: Mapping[str, Any], h: Mapping[str, float]) -> bool:
    return h["target_h"] > 0 and (h["timer_h"] > 0 or h["ticket_h"] > 0 or c["status"] in HOURS_STATUSES)


def is_over(h: Mapping[str, float]) -> bool:
    """Over target: the VISUAL ticket hours are over the work-order target (change C1). The timer never decides, and
    a packet with no ticket hours is not over."""
    return h["target_h"] > 0 and h["ticket_h"] > 0 and h["ticket_h"] > h["target_h"]


def station_worker(c: Mapping[str, Any]) -> str:
    parts = [(c.get("station") or {}).get("name"), (c.get("welder") or {}).get("display_name")]
    return " · ".join(str(x) for x in parts if x) or DASH


def hours_tables(rep: Report) -> list[dict[str, Any]]:
    """The old rptHours, with the station timer and the VISUAL ticket hours side by side against the target. The
    ticket hours decide "over target" (change C1); the timer is shown as today."""
    hrs = rep.hours()
    zero = {"timer_h": 0.0, "ticket_h": 0.0, "target_h": 0.0}
    hp = [(c, hrs.get(int(c["id"]), zero)) for c in rep.cards]
    hp = [(c, h) for c, h in hp if _counts_in_hours(c, h)]

    def group(key_of: Any) -> list[tuple[str, dict[str, float]]]:
        m: dict[str, dict[str, float]] = {}
        for c, h in hp:
            k = key_of(c) or DASH
            g = m.setdefault(k, {"timer": 0.0, "ticket": 0.0, "target": 0.0, "n": 0, "over": 0})
            g["timer"] += h["timer_h"]
            g["ticket"] += h["ticket_h"]
            g["target"] += h["target_h"]
            g["n"] += 1
            if is_over(h):
                g["over"] += 1
        return sorted(m.items(), key=lambda kv: -(kv[1]["ticket"] - kv[1]["target"]))

    def rows_of(groups: list[tuple[str, dict[str, float]]]) -> list[dict[str, Any]]:
        return [
            row(
                [
                    k,
                    int(v["n"]),
                    fmt_h(v["timer"]),
                    fmt_h(v["ticket"]),
                    fmt_h(v["target"]),
                    signed_h(v["ticket"] - v["target"]),
                    f"{half_up(v['ticket'] / v['target'] * 100)}%" if v["target"] else DASH,
                    int(v["over"]),
                ],
                flag=v["ticket"] > v["target"],
            )
            for k, v in groups
        ]

    def project_key(c: Mapping[str, Any]) -> str:
        p = c.get("project") or {}
        return f"{p.get('number') or ''} {p.get('name') or ''}".strip()

    t1 = table(
        "hours_dept",
        "By department",
        ["Department", *HOURS_HEAD],
        rows_of(group(lambda c: rep.dept_short(c["dept"]["code"]))),
    )
    t2 = table(
        "hours_worker",
        "By welder / worker",
        ["Worker", *HOURS_HEAD],
        rows_of(group(lambda c: (c.get("welder") or {}).get("display_name"))),
    )
    t3 = table("hours_project", "By project", ["Project", *HOURS_HEAD], rows_of(group(project_key)))
    over = sorted(
        ((c, h) for c, h in hp if is_over(h)),
        key=lambda ch: -(ch[1]["ticket_h"] - ch[1]["target_h"]),
    )
    t4 = table(
        "hours_over",
        "Packets over target",
        OVER_HEAD,
        [
            row(
                [
                    job_name(c),
                    station_worker(c),
                    search.STATUS_LABEL.get(str(c["status"]), str(c["status"])),
                    fmt_h(h["timer_h"]),
                    fmt_h(h["ticket_h"]),
                    fmt_h(h["target_h"]),
                    "+" + fmt_h(h["ticket_h"] - h["target_h"]) + " h",
                ],
                pid=c["id"],
                flag=True,
            )
            for c, h in over
        ],
        NOTE_HOURS_OVER,
    )
    return [t1, t2, t3, t4]


def over_target_count(rep: Report) -> int:
    hrs = rep.hours()
    n = 0
    for c in rep.cards:
        h = hrs.get(int(c["id"]))
        if h and is_over(h):
            n += 1
    return n


# ---------------------------------------------------------------- quality


def rejects(rep: Report) -> list[dict[str, Any]]:
    """Q.C. rejects in the period: closed inspections from qc_history and an open inspection still showing its
    reject, newest first."""
    if not rep.cards:
        return []
    found = packets.rows(rep.conn, REJECTS_SQL, (rep.ids,)) + packets.rows(rep.conn, OPEN_REJECTS_SQL, (rep.ids,))
    out = []
    for r in found:
        at = as_dt(r.get("result_at")) or as_dt(r["at"])
        if at is None or not rep.in_period(at):
            continue
        out.append({**r, "at": at})
    out.sort(key=lambda r: r["at"], reverse=True)
    return out


def open_cos(rep: Report) -> list[dict[str, Any]]:
    if not rep.cards:
        return []
    cos = packets.rows(rep.conn, COS_SQL, (rep.ids,))
    if cos:
        items: dict[int, list[dict[str, Any]]] = {}
        for it in packets.rows(
            rep.conn,
            "SELECT co_id, old_dwg, old_rev, new_dwg, new_rev, added FROM co_items WHERE co_id = ANY(%s) ORDER BY id",
            ([int(c["co_id"]) for c in cos],),
        ):
            items.setdefault(int(it["co_id"]), []).append(it)
        for c in cos:
            c["items"] = items.get(int(c["co_id"]), [])
    return cos


def co_summary(co: Mapping[str, Any]) -> str:
    """The old coSummary, through the data book's port."""
    from ..pdf import databook

    return str(databook.co_summary(co))


def quality_tables(rep: Report) -> list[dict[str, Any]]:
    by_id = {int(c["id"]): c for c in rep.cards}
    now = rep.now

    t1_rows = []
    for r in rejects(rep):
        c = by_id[int(r["packet_id"])]
        note = str(r.get("note") or "").strip()
        ncr = str(r.get("ncr") or "").strip()
        text = note or DASH
        if ncr and ncr not in note:
            text = f"{ncr} · {note}" if note else ncr
        t1_rows.append(
            row(
                [job_name(c), fmt_dt(r["at"], rep.tz), r.get("inspector") or DASH, text, station_worker(c)],
                pid=c["id"],
                flag=True,
            )
        )
    t1 = table(
        "quality_rejects",
        f"Q.C. rejects ({rep.period_label})",
        ["Job", "Date", "Inspector", "Reason / NCR", "Station / worker"],
        t1_rows,
    )

    t2_rows = []
    if rep.cards:
        from .pipeline import qty_text

        for s in packets.rows(rep.conn, SHORTS_SQL, (rep.ids,)):
            c = by_id[int(s["packet_id"])]
            eta = s["eta"]
            late = isinstance(eta, date) and eta < rep.today
            opened = as_dt(s["opened_at"]) or now
            t2_rows.append(
                row(
                    [
                        job_name(c),
                        f"{s['item']} {s['description'] or ''}".strip(),
                        qty_text(s["qty_open"]),
                        s["reason"] or "",
                        s["ref"] or DASH,
                        (day_str(eta) + (" (late)" if late else "")) if isinstance(eta, date) else DASH,
                        fmt_days((now - opened).total_seconds()),
                        step_label(c),
                    ],
                    pid=c["id"],
                    flag=late,
                )
            )
    t2 = table(
        "quality_shorts",
        "Open shorts",
        ["Job", "Item", "Short qty", "Reason", "PO / req", "ETA", "Age", "Packet is"],
        t2_rows,
    )

    t3_rows = []
    for co in open_cos(rep):
        c = by_id[int(co["packet_id"])]
        issued = as_dt(co["issued_at"])
        age = (now - issued).total_seconds() if issued else None
        label = packets.co_label(co["co_no"]) + (f" · {co['ecn']}" if co.get("ecn") else "")
        waiting = ", ".join(rep.dept_short(d) for d in co.get("pending_depts") or []) or DASH
        t3_rows.append(
            row(
                [job_name(c), label, co_summary(co), waiting, fmt_days(age) if age is not None else DASH],
                pid=c["id"],
                flag=age is not None and age > 2 * DAY_S,
            )
        )
    t3 = table(
        "quality_cos", "Open change orders", ["Job", "Change order", "Change", "Waiting on", "Open for"], t3_rows
    )

    t4_rows = []
    for c in rep.cards:
        if not on_hold(c):
            continue
        wh = c.get("wh") or {}
        at = as_dt(wh.get("at")) or now
        t4_rows.append(
            row(
                [
                    job_name(c),
                    wh.get("reason") or DASH,
                    wh.get("note") or "",
                    (wh.get("by") or {}).get("display_name") or DASH,
                    fmt_days((now - at).total_seconds()),
                ],
                pid=c["id"],
                flag=True,
            )
        )
    t4 = table("quality_holds", "Warehouse holds", ["Job", "Reason", "Note", "Put on hold by", "On hold for"], t4_rows)

    from .pipeline import LABELS

    sent: list[tuple[datetime, dict[str, Any], dict[str, Any]]] = []
    for pid, ts in rep.transitions().items():
        for t in ts:
            when = as_dt(t["at"])
            if t["action"] in SENT_BACK and when is not None and rep.in_period(when):
                sent.append((when, by_id[pid], t))
    sent.sort(key=lambda x: x[0], reverse=True)
    t5_rows = []
    for at, c, t in sent:
        try:
            label = LABELS.get(policy.Action(t["action"]), str(t["action"]))
        except ValueError:
            label = str(t["action"])
        t5_rows.append(
            row([job_name(c), fmt_dt(at, rep.tz), label, t.get("by_name") or DASH, t.get("reason") or ""], pid=c["id"])
        )
    t5 = table(
        "quality_sentback", f"Sent back ({rep.period_label})", ["Job", "When", "Action", "By", "Reason"], t5_rows
    )
    return [t1, t2, t3, t4, t5]


# ---------------------------------------------------------------- KPIs, summary, jobs


def filed_in_period(rep: Report) -> list[tuple[dict[str, Any], datetime]]:
    """Packets in scope with their latest transition to filed, when it is in the period."""
    by_id = {int(c["id"]): c for c in rep.cards}
    out = []
    for pid, ts in rep.transitions().items():
        filed = [as_dt(t["at"]) for t in ts if t["to_status"] == "filed"]
        last = max((f for f in filed if f is not None), default=None)
        if last is not None and rep.in_period(last):
            out.append((by_id[pid], last))
    return out


def kpis(rep: Report) -> list[dict[str, Any]]:
    """The old pmKpis, in order: each names the tab it opens; `alert` marks the ones shown hot when not zero."""
    open_cards = rep.open_cards
    uh = unit_healths(rep)
    filed = filed_in_period(rep)
    lead = [(f - (as_dt(c.get("created_at")) or f)).total_seconds() for c, f in filed]
    per = rep.period_label.lower()
    lead_text = f"{sum(lead) / len(lead) / DAY_S:.1f} d" if lead else DASH
    return [
        {"key": "open", "n": len(open_cards), "label": "Open packets", "tab": "aging", "alert": False},
        {
            "key": "late_units",
            "n": sum(1 for h in uh if h["health"] == "late"),
            "label": "Late units",
            "tab": "projects",
            "alert": True,
        },
        {
            "key": "risk_units",
            "n": sum(1 for h in uh if h["health"] == "risk"),
            "label": "Units at risk",
            "tab": "projects",
            "alert": True,
        },
        {
            "key": "holds",
            "n": sum(1 for c in open_cards if on_hold(c)),
            "label": "Warehouse holds",
            "tab": "quality",
            "alert": True,
        },
        {
            "key": "cos",
            "n": sum(len(c.get("open_cos") or []) for c in rep.cards),
            "label": "Open change orders",
            "tab": "quality",
            "alert": True,
        },
        {
            "key": "shorts",
            "n": sum(int(c.get("open_shorts") or 0) for c in rep.cards),
            "label": "Open shorts",
            "tab": "quality",
            "alert": True,
        },
        {"key": "rejects", "n": len(rejects(rep)), "label": f"Q.C. rejects · {per}", "tab": "quality", "alert": True},
        {
            "key": "over",
            "n": over_target_count(rep),
            "label": "Packets over target hours (tickets)",
            "tab": "hours",
            "alert": True,
        },
        {"key": "filed", "n": len(filed), "label": f"Filed · {per}", "tab": "projects", "alert": False},
        {
            "key": "lead",
            "n": lead_text,
            "label": "Average lead time, created → filed",
            "tab": "aging",
            "alert": False,
        },
    ]


def summary(rep: Report, k: list[dict[str, Any]] | None = None, pipe: list[dict[str, Any]] | None = None) -> list[str]:
    """The old pmSummaryLines: the plain-words panel."""
    k = k if k is not None else kpis(rep)
    pipe = pipe if pipe is not None else pipeline(rep)
    n = {x["key"]: x["n"] for x in k}
    busy = [s for s in pipe if s["count"]]
    slow = max(busy, key=lambda s: s["avg_s"]) if busy else None
    big = max(busy, key=lambda s: s["count"]) if busy else None
    late = [
        r["cells"][0] for r in projects_table(rep, expand_all=True)["rows"] if r["lvl"] == 1 and r["health"] == "late"
    ]
    out = []
    o = int(n["open"])
    out.append(f"{o} open packet{'' if o == 1 else 's'} in scope ({rep.scope_label()}).")
    if late:
        more = f" and {len(late) - 6} more" if len(late) > 6 else ""
        out.append(f"Late: {', '.join(late[:6])}{more}.")
    else:
        out.append("No units are past their due date.")
    if big:
        out.append(f'Most packets are at "{big["label"]}" ({big["count"]}).')
    if slow:
        out.append(
            f'Slowest step right now: "{slow["label"]}", averaging {fmt_days(slow["avg_s"])}'
            f" (longest {fmt_days(slow['max_s'])})."
        )
    issues = [
        plural(int(n["holds"]), "warehouse hold") if n["holds"] else "",
        plural(int(n["cos"]), "open change order") if n["cos"] else "",
        plural(int(n["shorts"]), "open short") if n["shorts"] else "",
        plural(int(n["rejects"]), "Q.C. reject") if n["rejects"] else "",
    ]
    issues = [x for x in issues if x]
    out.append(
        f"Needs attention: {', '.join(issues)}." if issues else "No holds, open change orders, shorts or Q.C. rejects."
    )
    over = int(n["over"])
    if over:
        out.append(f"{over} packet{' is' if over == 1 else 's are'} over target hours on the VISUAL tickets.")
    return out


def jobs_table(rep: Report) -> dict[str, Any]:
    """The old pmJobsTable: every job in scope, with the hours."""
    hrs = rep.hours()
    rows = []
    for c in rep.cards:
        h = hrs.get(int(c["id"]), {"timer_h": 0.0, "ticket_h": 0.0, "target_h": 0.0})
        filed = c["status"] == "filed"
        due = rep.unit_due(c)
        rows.append(
            row(
                [
                    job_name(c),
                    (c.get("project") or {}).get("number") or "",
                    (c.get("unit") or {}).get("number") or "",
                    c.get("op") or "",
                    c.get("title") or "",
                    rep.dept_short(c["dept"]["code"]),
                    step_label(c),
                    c.get("holder") or "",
                    (c.get("station") or {}).get("name") or "",
                    (c.get("welder") or {}).get("display_name") or "",
                    c.get("drawing") or "",
                    c.get("rev") or "",
                    due.isoformat() if due else "",
                    "" if filed else f"{(rep.now - step_since(c)).total_seconds() / DAY_S:.1f}",
                    fmt_h(h["timer_h"]),
                    fmt_h(h["ticket_h"]),
                    fmt_h(h["target_h"]),
                    fmt_d(c.get("created_at"), rep.tz),
                    fmt_dt(c.get("updated_at"), rep.tz),
                    len(c.get("open_cos") or []),
                    int(c.get("open_shorts") or 0),
                ],
                pid=c["id"],
                project_id=(c.get("project") or {}).get("id"),
                unit_id=(c.get("unit") or {}).get("id"),
            )
        )
    return table("jobs", "All jobs in scope", JOBS_HEAD, rows)


# ---------------------------------------------------------------- the answers


def overview(rep: Report) -> dict[str, Any]:
    pipe = pipeline(rep)
    k = kpis(rep)
    return {
        "scope": rep.scope(),
        "kpis": k,
        "summary": summary(rep, k, pipe),
        "pipeline": pipe,
        "attention": attention_table(rep),
    }


def tables_for(rep: Report, key: str) -> tuple[str, list[dict[str, Any]]] | None:
    """The CSV export's name and tables for a table key, `all`, or None for an unknown key."""
    if key == "all":
        return "All reports", all_tables(rep)
    if key == "jobs":
        return "Jobs", [jobs_table(rep)]
    if key not in TABLE_KEYS:
        return None
    for t in all_tables(rep, jobs=False):
        if t["key"] == key:
            return str(t["title"]), [t]
    return None


def all_tables(rep: Report, *, jobs: bool = True) -> list[dict[str, Any]]:
    """The old pmAllTables: every report with every unit open, then the jobs."""
    out = [projects_table(rep, expand_all=True), *aging_tables(rep), *hours_tables(rep), *quality_tables(rep)]
    if jobs:
        out.append(jobs_table(rep))
    return out


def file_stem(name: str) -> str:
    return re.sub(r"[^A-Za-z0-9]+", "_", name).strip("_") or "Report"


def csv_rows(rep: Report, name: str, tables: Sequence[Mapping[str, Any]], who_name: str) -> list[list[str]]:
    """The old pmSaveCsv rows: the report name, the scope, the period, who exported it and when, then each table."""
    out: list[list[str]] = [
        [f"PESCO Packet Station — {name}"],
        ["Scope", rep.scope_label()],
        ["History period", rep.period_label],
        ["Exported", fmt_dt(rep.now, rep.tz), "by", who_name],
    ]
    for t in tables:
        out.append([])
        out.append([str(t["title"])])
        out.append([str(h) for h in t["head"]])
        out += [list(r["cells"]) for r in t["rows"]]
    return out