new-site/scripts/workers/deminimis_factor_check.py

"""De minimis factor missing-check cron.

USAC publishes the annual de minimis factor for the next 499-A filing each
November. Our `fcc_deminimis_factors` table has one row per reporting
(calendar revenue) year. If the upcoming reporting year's factor isn't
seeded by March 1 of the filing year, every 499-A handler in that cohort
will blow up at `loadDeMinimisFactor()` time.

This script is a 3am cron that:
  1. Computes the current reporting year (previous calendar year, since
     499-As are filed April 1 for the prior year's revenue).
  2. Checks whether `fcc_deminimis_factors` has a row keyed by that year.
  3. If missing, inserts an admin ToDo in ERPNext + emails ops.

Also validates that the row for the NEXT reporting year is on file by
January 15 of each year — an early warning so we can ping USAC before
customers try to file.

Usage:
    python -m scripts.workers.deminimis_factor_check
    python -m scripts.workers.deminimis_factor_check --dry-run

CRON: 0 3 * * *  python -m scripts.workers.deminimis_factor_check
"""

from __future__ import annotations

import argparse
import logging
import os
import sys
from datetime import date

import psycopg2

DATABASE_URL = os.environ.get("DATABASE_URL", "")
ADMIN_EMAIL  = os.environ.get("ADMIN_EMAIL", "ops@performancewest.net")

log = logging.getLogger("deminimis_factor_check")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)


def _current_reporting_year(today: date | None = None) -> int:
    """499-A reports revenue earned in the previous calendar year."""
    today = today or date.today()
    return today.year - 1


def _has_factor(conn, form_year: int) -> bool:
    with conn.cursor() as cur:
        cur.execute(
            "SELECT 1 FROM fcc_deminimis_factors WHERE form_year = %s",
            (form_year,),
        )
        return cur.fetchone() is not None


def _already_todo(conn, form_year: int) -> bool:
    """Avoid spamming the same ToDo every day."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT 1 FROM order_audit_log
             WHERE action = 'deminimis_factor_missing'
               AND metadata->>'form_year' = %s
               AND created_at > NOW() - INTERVAL '7 days'
             LIMIT 1
            """,
            (str(form_year),),
        )
        return cur.fetchone() is not None


def _record_missing(conn, form_year: int, severity: str) -> None:
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO order_audit_log
              (order_type, order_id, action, actor_type, actor_name, note, metadata)
            VALUES ('service', 0, 'deminimis_factor_missing', 'system',
                    'deminimis_factor_check',
                    %s,
                    jsonb_build_object('form_year', %s::text, 'severity', %s))
            """,
            (
                f"No fcc_deminimis_factors row for reporting_year={form_year}. "
                f"Handler will error when customer tries to file. Seed from "
                f"USAC Appendix A.",
                form_year, severity,
            ),
        )
    conn.commit()


def _create_admin_todo(form_year: int, severity: str) -> None:
    try:
        from scripts.workers.erpnext_client import ERPNextClient
        ERPNextClient().create_resource(
            "ToDo",
            {
                "description": (
                    f"[{severity.upper()}] Missing fcc_deminimis_factors row "
                    f"for reporting year {form_year}. "
                    f"Seed from USAC Appendix A (published each November) "
                    f"with: INSERT INTO fcc_deminimis_factors "
                    f"(form_year, factor, source_citation) VALUES ({form_year}, "
                    f"<0.XXXX>, 'For CY{form_year} revenue — 499-A Appendix A');"
                ),
                "priority": "High" if severity == "blocker" else "Medium",
                "role": "Accounting Advisor",
            },
        )
    except Exception as exc:
        log.error("Could not create admin ToDo: %s", exc)


def run_once(dry_run: bool = False, today: date | None = None) -> dict:
    """Check current + next reporting-year factors. Returns a summary dict."""
    today = today or date.today()
    summary = {
        "current_year": None,
        "current_year_missing": False,
        "next_year_missing": False,
        "actions": [],
    }

    if not DATABASE_URL:
        log.error("DATABASE_URL not set — aborting")
        return summary

    conn = psycopg2.connect(DATABASE_URL)
    try:
        # Current reporting year — must be present, filings are active.
        current = _current_reporting_year(today)
        summary["current_year"] = current
        if not _has_factor(conn, current):
            summary["current_year_missing"] = True
            if not _already_todo(conn, current):
                severity = "blocker"
                log.error(
                    "BLOCKER: No de minimis factor for reporting year %s — "
                    "499-A handler will throw on every filing",
                    current,
                )
                if not dry_run:
                    _record_missing(conn, current, severity)
                    _create_admin_todo(current, severity)
                summary["actions"].append(f"alerted_current_{current}")

        # Next reporting year — early warning after Jan 15.
        next_year = current + 1
        if today >= date(today.year, 1, 15) and not _has_factor(conn, next_year):
            summary["next_year_missing"] = True
            if not _already_todo(conn, next_year):
                severity = "warning"
                log.warning(
                    "WARNING: No de minimis factor for reporting year %s — "
                    "will become a blocker once customers file for %s",
                    next_year, next_year,
                )
                if not dry_run:
                    _record_missing(conn, next_year, severity)
                    _create_admin_todo(next_year, severity)
                summary["actions"].append(f"alerted_next_{next_year}")

        if not summary["current_year_missing"] and not summary["next_year_missing"]:
            log.info(
                "De minimis factor check OK: both %s and %s are seeded",
                current, next_year,
            )
    finally:
        conn.close()
    return summary


def main() -> None:
    parser = argparse.ArgumentParser(
        description="De minimis factor missing-check (daily cron)",
    )
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    summary = run_once(dry_run=args.dry_run)
    log.info("Summary: %s", summary)


if __name__ == "__main__":
    main()