paliad/exports/gen-deadline-list.py

#!/usr/bin/env python3
"""Generate a markdown deadline-list export for UPC PA training (work/head delegation #2572).

Sorts by proceeding-type display_order then sequence_order. Sections by proceeding.

t-paliad-348 / yoUPC#178 update: matches the engine's `IncludeOptional=false`
default (`pkg/litigationplanner/engine.go`). Optional rules (priority='optional')
are SUPPRESSED by default so the manuscript shows the same "naked proceeding
backbone" the UI now renders. Pass `--include-optional` to opt back in for an
exhaustive catalog dump.

Usage:
    uv run exports/gen-deadline-list.py [--include-optional] [-o OUT]
"""
# /// script
# requires-python = ">=3.10"
# dependencies = ["psycopg2-binary"]
# ///
import argparse
import os
import sys
from datetime import date
from pathlib import Path

import psycopg2
import psycopg2.extras

DSN = os.environ.get(
    "PALIAD_DEADLINE_EXPORT_DSN",
    "postgres://postgres:rpsak3yf4lu1izgefx9p9xweg3qroojw@100.99.98.201:11833/postgres?sslmode=disable",
)

# `priority` filter is wired at the SQL level (not post-filter in Python) so
# the row counter in the markdown header reflects what's actually in the
# manuscript — matching what the lawyer sees on /tools/procedures.
SQL_TEMPLATE = """
SELECT
  pt.code AS pt_code,
  pt.display_order,
  COALESCE(pt.name_en, pt.name) AS pt_label_en,
  pt.name AS pt_label_de,
  COALESCE(pe.name_en, pe.name) AS event_en,
  pe.name AS event_de,
  sr.duration_value,
  sr.duration_unit,
  sr.timing,
  sr.alt_duration_value,
  sr.alt_duration_unit,
  sr.combine_op,
  sr.rule_code,
  COALESCE(te.name, te.name_de) AS trigger_label,
  te.code AS trigger_code,
  sr.primary_party,
  sr.is_court_set,
  sr.is_spawn,
  sr.priority,
  sr.deadline_notes_en,
  sr.deadline_notes,
  sr.condition_expr,
  sr.sequence_order
FROM paliad.sequencing_rules sr
JOIN paliad.procedural_events pe ON pe.id = sr.procedural_event_id
LEFT JOIN paliad.proceeding_types pt ON pt.id = sr.proceeding_type_id
LEFT JOIN paliad.trigger_events te ON te.id = sr.trigger_event_id
WHERE sr.lifecycle_state = 'published'
  AND sr.is_active = true
  AND pt.id IS NOT NULL
  {priority_filter}
ORDER BY pt.display_order NULLS LAST, pt.code, sr.sequence_order NULLS LAST, sr.rule_code, pe.name;
"""


def format_frist(duration_value, duration_unit, timing, alt_value, alt_unit, combine_op):
    """Format the deadline duration cleanly."""
    if duration_value is None or duration_unit is None:
        return ""
    unit_map = {
        "days": "d",
        "weeks": "w",
        "months": "M",
        "years": "y",
        "calendar_days": "CD",
        "working_days": "WD",
    }
    unit = unit_map.get(duration_unit, duration_unit)
    main = f"{duration_value} {unit}"
    if alt_value is not None and alt_unit is not None:
        alt_unit_short = unit_map.get(alt_unit, alt_unit)
        op = combine_op or "or"
        main = f"{main} {op} {alt_value} {alt_unit_short}"
    if timing == "before":
        main = f"{main} before"
    elif timing == "after":
        main = f"{main} after"
    return main


def format_party(primary_party, is_court_set):
    if is_court_set:
        return "court-set"
    if primary_party == "claimant":
        return "claimant"
    if primary_party == "defendant":
        return "defendant"
    if primary_party == "both":
        return "either"
    if primary_party == "court":
        return "court"
    return primary_party or "—"


def detect_r94(notes_en, notes_de):
    """Flag R.9.4 non-extendable from notes text (heuristic — no DB field)."""
    blobs = " ".join(filter(None, [notes_en or "", notes_de or ""])).lower()
    if "r.9.4" in blobs or "r 9.4" in blobs or "r9.4" in blobs:
        return "✗"
    if "non-extendable" in blobs or "nicht verlängerbar" in blobs or "nicht verlaengerbar" in blobs:
        return "✗"
    return ""


def conditional_marker(condition_expr):
    if condition_expr in (None, "", {}):
        return ""
    # condition_expr is JSONB → returns dict
    if isinstance(condition_expr, dict):
        if "flag" in condition_expr:
            return f"if `{condition_expr['flag']}`"
        if condition_expr.get("op") == "and" and "args" in condition_expr:
            flags = [a.get("flag", "?") for a in condition_expr["args"]]
            return "if " + " & ".join(f"`{f}`" for f in flags)
        if condition_expr.get("op") == "or" and "args" in condition_expr:
            flags = [a.get("flag", "?") for a in condition_expr["args"]]
            return "if " + " | ".join(f"`{f}`" for f in flags)
    return "cond"


def md_escape(s):
    if s is None:
        return ""
    return str(s).replace("|", "\\|").replace("\n", " ")


def render(rows, *, include_optional: bool, generated_for: str) -> str:
    by_pt = {}
    for r in rows:
        key = (r["display_order"] or 9999, r["pt_code"], r["pt_label_de"], r["pt_label_en"])
        by_pt.setdefault(key, []).append(r)

    out = []
    today = date.today().isoformat()
    out.append(f"# UPC + DE/EP Deadline Catalog — Stand {today}")
    out.append("")
    out.append(f"Source: `paliad.sequencing_rules` (lifecycle_state=published, is_active=true).")
    out.append(f"Generated for {generated_for}. {len(rows)} rules across {len(by_pt)} proceedings.")
    if include_optional:
        out.append("")
        out.append(
            "**Mode:** `--include-optional` — every published rule, including "
            "`priority='optional'` rules suppressed by the engine's default "
            "(`IncludeOptional=false`). This is the exhaustive catalog dump."
        )
    else:
        out.append("")
        out.append(
            "**Mode:** default — matches the engine's `IncludeOptional=false` "
            "behaviour (pkg/litigationplanner/engine.go). `priority='optional'` "
            "rules are suppressed; the manuscript shows only the mandatory "
            "backbone the lawyer sees by default on /tools/procedures. "
            "Re-run with `--include-optional` for the full catalog. "
            "(t-paliad-348 / yoUPC#178)"
        )
    out.append("")
    out.append("**Spalten:**")
    out.append("- **Phase/Event** = procedural event (German primary)")
    out.append("- **Frist** = duration + timing (`d` days, `w` weeks, `M` months, `CD` calendar days, `WD` working days; `before` = relative to anchor)")
    out.append("- **Rule** = legal source (RoP / § ZPO / § PatG / Art. EPÜ)")
    out.append("- **Anchor** = trigger event the deadline runs from")
    out.append("- **Seite** = filing party (claimant / defendant / either / court-set)")
    out.append("- **Priorität** = mandatory / recommended / optional / informational (only when `--include-optional`)")
    out.append("- **R.9.4** = ✗ marked non-extendable in notes (heuristic — confirm against rule text)")
    out.append("- **Bedingung** = scenario flag(s) that must be set for the rule to fire (blank = always)")
    out.append("")
    out.append("---")
    out.append("")

    for (order, pt_code, pt_de, pt_en) in sorted(by_pt.keys()):
        prules = by_pt[(order, pt_code, pt_de, pt_en)]
        out.append(f"## {pt_de} · `{pt_code}`")
        out.append("")
        if pt_en and pt_en != pt_de:
            out.append(f"*{pt_en}*")
            out.append("")
        if include_optional:
            out.append("| # | Phase / Event | Frist | Rule | Anchor | Seite | Priorität | R.9.4 | Bedingung |")
            out.append("|---:|---|---|---|---|---|---|:---:|---|")
        else:
            out.append("| # | Phase / Event | Frist | Rule | Anchor | Seite | R.9.4 | Bedingung |")
            out.append("|---:|---|---|---|---|---|:---:|---|")
        for i, r in enumerate(prules, 1):
            event = md_escape(r["event_de"] or r["event_en"] or "")
            frist = md_escape(
                format_frist(
                    r["duration_value"], r["duration_unit"], r["timing"],
                    r["alt_duration_value"], r["alt_duration_unit"], r["combine_op"],
                )
            )
            rule = md_escape(r["rule_code"] or "")
            anchor = md_escape(r["trigger_label"] or "")
            party = format_party(r["primary_party"], r["is_court_set"])
            r94 = detect_r94(r["deadline_notes_en"], r["deadline_notes"])
            cond = md_escape(conditional_marker(r["condition_expr"]))
            spawn_marker = " ⤴" if r["is_spawn"] else ""
            if include_optional:
                priority = md_escape(r["priority"] or "")
                out.append(
                    f"| {i} | {event}{spawn_marker} | {frist} | {rule} | {anchor} | {party} | {priority} | {r94} | {cond} |"
                )
            else:
                out.append(
                    f"| {i} | {event}{spawn_marker} | {frist} | {rule} | {anchor} | {party} | {r94} | {cond} |"
                )
        out.append("")

    out.append("---")
    out.append("")
    out.append("**Lesehilfe:**")
    out.append("- ⤴ Spawn-Marker: event opens a sub-proceeding (e.g. CCR forks revocation track)")
    out.append("- `with_ccr` = Widerklage auf Nichtigkeit gefilt | `with_amend` = Patentänderungsantrag | `with_cci` = Widerklage auf Verletzung (in rev.cfi)")
    out.append("- Catalog ist work-in-progress: 7 compound-name rules + Patentänderung-Duplikation noch in m's split-review backlog (m/paliad#149).")
    return "\n".join(out)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--include-optional",
        action="store_true",
        help="Include priority='optional' rules. Default false matches the engine's IncludeOptional=false default.",
    )
    parser.add_argument(
        "-o",
        "--out",
        default="exports/upc-deadlines-2026-05-28.md",
        help="Output path (relative to repo root).",
    )
    parser.add_argument(
        "--generated-for",
        default="PA-Schulung 2026-05-28",
        help="Free-text label rendered in the markdown header.",
    )
    args = parser.parse_args()

    priority_filter = "" if args.include_optional else "AND sr.priority != 'optional'"
    sql = SQL_TEMPLATE.format(priority_filter=priority_filter)

    conn = psycopg2.connect(DSN)
    try:
        cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
        cur.execute(sql)
        rows = cur.fetchall()
    finally:
        conn.close()

    md = render(rows, include_optional=args.include_optional, generated_for=args.generated_for)
    # Resolve out path relative to the repo root (= the script's grandparent).
    out_path = Path(args.out)
    if not out_path.is_absolute():
        out_path = Path(__file__).resolve().parent.parent / out_path
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(md)
    n_pt = len({(r["display_order"] or 9999, r["pt_code"]) for r in rows})
    print(
        f"WROTE {out_path} ({len(rows)} rules, {n_pt} proceedings, "
        f"include_optional={args.include_optional})"
    )


if __name__ == "__main__":
    sys.exit(main())