f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
333 lines
13 KiB
Python
333 lines
13 KiB
Python
"""FCC Form 499-A shared utilities.
|
||
|
||
De minimis calculator (Appendix A), safe-harbor percentage lookup, Line 612
|
||
filing-type detection, and Line 105 box-tick derivation.
|
||
|
||
Used by form_499a.py, form_499_initial.py, and the /validate endpoint's
|
||
Python counterpart (if we ever move validation server-side).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
import psycopg2
|
||
import os
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ── Line 105 box-tick derivation ────────────────────────────────────────
|
||
# Mirrors site/src/lib/line_105_catalog.ts::derivedLine105Boxes. When a
|
||
# CLEC/IXC/Wireless has infra_type='reseller' or 'mvno', the handler
|
||
# automatically ticks the corresponding derived Line 105 box on the form.
|
||
|
||
LINE_105_BOX_NUMBERS = {
|
||
"voip_interconnected": 1,
|
||
"voip_non_interconnected": 2,
|
||
"clec": 3,
|
||
"ilec": 4,
|
||
"local_reseller": 5, # derived from clec + reseller
|
||
"toll_reseller": 6, # derived from ixc + reseller
|
||
"ixc": 7,
|
||
"wireless": 8,
|
||
"mvno": 9, # derived from wireless + mvno
|
||
"prepaid_calling_card": 10,
|
||
"private_line": 11,
|
||
"satellite": 12,
|
||
"payphone": 13,
|
||
"osp": 14,
|
||
"shared_tenant": 15,
|
||
"audio_bridging": 16,
|
||
"toll_free": 17,
|
||
"paging": 18,
|
||
"smr": 19,
|
||
"fixed_wireless": 20,
|
||
"mobile_satellite": 21,
|
||
"other": 22,
|
||
}
|
||
|
||
|
||
def derived_line_105_boxes(category_id: str, infra_type: Optional[str]) -> list[int]:
|
||
"""Return extra Line 105 boxes to tick because of the infra_type flag."""
|
||
boxes: list[int] = []
|
||
if infra_type == "reseller":
|
||
if category_id == "clec":
|
||
boxes.append(5)
|
||
elif category_id == "ixc":
|
||
boxes.append(6)
|
||
if infra_type == "mvno" and category_id == "wireless":
|
||
boxes.append(9)
|
||
return boxes
|
||
|
||
|
||
def all_line_105_boxes_to_tick(line_105_categories: list[dict]) -> list[int]:
|
||
"""Return every Line 105 box number to tick for this filer."""
|
||
boxes: set[int] = set()
|
||
for cat in line_105_categories or []:
|
||
cat_id = cat.get("id")
|
||
if cat_id and cat_id in LINE_105_BOX_NUMBERS:
|
||
boxes.add(LINE_105_BOX_NUMBERS[cat_id])
|
||
boxes.update(derived_line_105_boxes(cat_id, cat.get("infra_type")))
|
||
return sorted(boxes)
|
||
|
||
|
||
# ── Safe harbor lookup ──────────────────────────────────────────────────
|
||
|
||
SAFE_HARBOR_DISALLOWED_CATEGORIES = {"voip_non_interconnected"}
|
||
|
||
|
||
def _db_connect():
|
||
return psycopg2.connect(os.environ.get("DATABASE_URL", ""))
|
||
|
||
|
||
def load_safe_harbor_pct(form_year: int, category_id: str) -> Optional[float]:
|
||
"""Return the safe-harbor interstate % for (year, category), or None.
|
||
|
||
None is returned for categories that have no safe harbor (e.g.,
|
||
non-interconnected VoIP) or if the year/category combination isn't in
|
||
the fcc_safe_harbor_percentages seed table.
|
||
"""
|
||
if category_id in SAFE_HARBOR_DISALLOWED_CATEGORIES:
|
||
return None
|
||
try:
|
||
conn = _db_connect()
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT interstate_pct FROM fcc_safe_harbor_percentages "
|
||
"WHERE form_year = %s AND line_105_category = %s",
|
||
(form_year, category_id),
|
||
)
|
||
row = cur.fetchone()
|
||
conn.close()
|
||
return float(row[0]) if row else None
|
||
except Exception as exc:
|
||
logger.warning("safe-harbor lookup failed: %s", exc)
|
||
return None
|
||
|
||
|
||
def safe_harbor_allowed(category_id: str) -> bool:
|
||
return category_id not in SAFE_HARBOR_DISALLOWED_CATEGORIES
|
||
|
||
|
||
# ── De minimis calculator (Appendix A, 11-line worksheet) ───────────────
|
||
|
||
|
||
@dataclass
|
||
class DeMinimisWorksheet:
|
||
"""Appendix A de minimis determination worksheet.
|
||
|
||
Every field corresponds to a line in the 2026 Form 499-A Appendix A.
|
||
Mirrors the PDF layout exactly so auditors can follow along.
|
||
"""
|
||
form_year: int
|
||
# Lines 1-4 — interstate/intl contribution bases for filer + affiliates
|
||
line_1_filer_interstate_cents: int = 0
|
||
line_2_filer_intl_cents: int = 0
|
||
line_3_affiliates_interstate_cents: int = 0
|
||
line_4_affiliates_intl_cents: int = 0
|
||
# Line 5 — consolidated interstate
|
||
line_5_consolidated_interstate_cents: int = 0
|
||
# Line 6 — consolidated interstate+intl (pre-LIRE exclusion)
|
||
line_6_consolidated_total_cents: int = 0
|
||
# Line 7 — interstate as % of consolidated total (LIRE test)
|
||
line_7_interstate_pct: float = 0.0
|
||
# Line 8 — LIRE exempt? (line_7 ≤ 12%)
|
||
line_8_lire_exempt: bool = False
|
||
# Line 9 — contribution base to test
|
||
line_9_contribution_base_cents: int = 0
|
||
# Line 10 — year-specific factor (0.256 for 2026)
|
||
line_10_factor: float = 0.0
|
||
# Line 11 — estimated annual contribution = line_9 × line_10
|
||
line_11_estimated_contrib_cents: int = 0
|
||
# Result
|
||
is_de_minimis: bool = False
|
||
threshold_usd: int = 10000
|
||
notes: list[str] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"form_year": self.form_year,
|
||
"line_1_filer_interstate_cents": self.line_1_filer_interstate_cents,
|
||
"line_2_filer_intl_cents": self.line_2_filer_intl_cents,
|
||
"line_3_affiliates_interstate_cents": self.line_3_affiliates_interstate_cents,
|
||
"line_4_affiliates_intl_cents": self.line_4_affiliates_intl_cents,
|
||
"line_5_consolidated_interstate_cents": self.line_5_consolidated_interstate_cents,
|
||
"line_6_consolidated_total_cents": self.line_6_consolidated_total_cents,
|
||
"line_7_interstate_pct": self.line_7_interstate_pct,
|
||
"line_8_lire_exempt": self.line_8_lire_exempt,
|
||
"line_9_contribution_base_cents": self.line_9_contribution_base_cents,
|
||
"line_10_factor": self.line_10_factor,
|
||
"line_11_estimated_contrib_cents": self.line_11_estimated_contrib_cents,
|
||
"is_de_minimis": self.is_de_minimis,
|
||
"threshold_usd": self.threshold_usd,
|
||
"notes": self.notes,
|
||
}
|
||
|
||
|
||
def load_deminimis_factor(form_year: int) -> float:
|
||
"""Return the Appendix A Line 10 factor for a form year.
|
||
|
||
Raises ValueError if the year isn't in the seed table — an unknown
|
||
form year is a code bug, not a graceful-fallback situation.
|
||
"""
|
||
try:
|
||
conn = _db_connect()
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT factor FROM fcc_deminimis_factors WHERE form_year = %s",
|
||
(form_year,),
|
||
)
|
||
row = cur.fetchone()
|
||
conn.close()
|
||
except Exception as exc:
|
||
logger.error("deminimis factor lookup failed: %s", exc)
|
||
raise
|
||
if not row:
|
||
raise ValueError(f"No de minimis factor configured for form year {form_year}")
|
||
return float(row[0])
|
||
|
||
|
||
def calculate_de_minimis(
|
||
*,
|
||
form_year: int,
|
||
filer_total_revenue_cents: int,
|
||
filer_interstate_pct: float,
|
||
filer_international_pct: float,
|
||
affiliates: Optional[list[dict]] = None,
|
||
) -> DeMinimisWorksheet:
|
||
"""Compute the Appendix A de minimis worksheet.
|
||
|
||
affiliates is a list of {total_revenue_cents, interstate_pct,
|
||
international_pct} records for each affiliated filer. Empty list =
|
||
no affiliates.
|
||
"""
|
||
affiliates = affiliates or []
|
||
w = DeMinimisWorksheet(form_year=form_year)
|
||
|
||
# Line 1: filer interstate revenue
|
||
w.line_1_filer_interstate_cents = int(
|
||
filer_total_revenue_cents * (filer_interstate_pct / 100.0)
|
||
)
|
||
# Line 2: filer international revenue
|
||
w.line_2_filer_intl_cents = int(
|
||
filer_total_revenue_cents * (filer_international_pct / 100.0)
|
||
)
|
||
|
||
# Lines 3-4: affiliates
|
||
for a in affiliates:
|
||
tot = int(a.get("total_revenue_cents", 0))
|
||
ipct = float(a.get("interstate_pct", 0))
|
||
intl = float(a.get("international_pct", 0))
|
||
w.line_3_affiliates_interstate_cents += int(tot * ipct / 100.0)
|
||
w.line_4_affiliates_intl_cents += int(tot * intl / 100.0)
|
||
|
||
# Line 5: consolidated interstate
|
||
w.line_5_consolidated_interstate_cents = (
|
||
w.line_1_filer_interstate_cents + w.line_3_affiliates_interstate_cents
|
||
)
|
||
# Line 6: consolidated interstate + intl
|
||
w.line_6_consolidated_total_cents = w.line_5_consolidated_interstate_cents + (
|
||
w.line_2_filer_intl_cents + w.line_4_affiliates_intl_cents
|
||
)
|
||
|
||
# Line 7: interstate as % of consolidated total (guard /0)
|
||
if w.line_6_consolidated_total_cents > 0:
|
||
w.line_7_interstate_pct = round(
|
||
100.0 * w.line_5_consolidated_interstate_cents /
|
||
w.line_6_consolidated_total_cents,
|
||
4,
|
||
)
|
||
else:
|
||
w.line_7_interstate_pct = 0.0
|
||
|
||
# Line 8: LIRE exempt if interstate ≤ 12% of combined
|
||
w.line_8_lire_exempt = w.line_7_interstate_pct <= 12.0
|
||
|
||
# Line 9: contribution base to test — interstate + (0 if LIRE else intl)
|
||
intl_total = w.line_2_filer_intl_cents + w.line_4_affiliates_intl_cents
|
||
w.line_9_contribution_base_cents = (
|
||
w.line_5_consolidated_interstate_cents +
|
||
(0 if w.line_8_lire_exempt else intl_total)
|
||
)
|
||
|
||
# Line 10: year factor
|
||
w.line_10_factor = load_deminimis_factor(form_year)
|
||
|
||
# Line 11: estimated annual contribution
|
||
w.line_11_estimated_contrib_cents = int(
|
||
w.line_9_contribution_base_cents * w.line_10_factor
|
||
)
|
||
|
||
# Result: de minimis if < $10,000
|
||
w.is_de_minimis = w.line_11_estimated_contrib_cents < (w.threshold_usd * 100)
|
||
|
||
if w.is_de_minimis:
|
||
w.notes.append(
|
||
f"De minimis: estimated contribution ${w.line_11_estimated_contrib_cents/100:,.2f}"
|
||
f" < ${w.threshold_usd:,.0f} threshold."
|
||
)
|
||
else:
|
||
w.notes.append(
|
||
f"NOT de minimis: estimated contribution ${w.line_11_estimated_contrib_cents/100:,.2f}"
|
||
f" ≥ ${w.threshold_usd:,.0f} threshold."
|
||
)
|
||
if w.line_8_lire_exempt:
|
||
w.notes.append(
|
||
f"LIRE exempt: interstate ({w.line_7_interstate_pct:.2f}%) ≤ 12% "
|
||
f"of combined interstate+intl — international revenue excluded."
|
||
)
|
||
return w
|
||
|
||
|
||
# ── Line 612 filing-type detection ──────────────────────────────────────
|
||
|
||
|
||
def detect_filing_type(
|
||
*,
|
||
entity: dict,
|
||
current_year_filing_exists: bool = False,
|
||
revised_reason: Optional[str] = None,
|
||
) -> str:
|
||
"""Return one of: original_april_1, registration_new_filer,
|
||
revised_registration, revised_revenue.
|
||
"""
|
||
if not entity.get("filer_id_499"):
|
||
return "registration_new_filer"
|
||
if current_year_filing_exists:
|
||
if revised_reason == "registration":
|
||
return "revised_registration"
|
||
if revised_reason == "revenue":
|
||
return "revised_revenue"
|
||
return "original_april_1"
|
||
|
||
|
||
# ── TRS contribution base (Lines 512-514) ───────────────────────────────
|
||
|
||
# Revenue lines that roll up into the TRS contribution base.
|
||
# Line 418.4 (non-interconnected VoIP) is included ONLY in TRS base —
|
||
# it's excluded from USF/NANPA/LNP/ITSP bases. Line 511 subtracts.
|
||
TRS_BASE_LINE_KEYS = [
|
||
"line_403", "line_404", "line_404_1", "line_404_3",
|
||
"line_405", "line_406", "line_407", "line_408",
|
||
"line_409", "line_410", "line_411", "line_412",
|
||
"line_413", "line_414_1", "line_414_2",
|
||
"line_415", "line_416", "line_417",
|
||
"line_418_4", # TRS-only
|
||
]
|
||
|
||
|
||
def compute_trs_contribution_base(revenue_lines: dict) -> tuple[int, int, int]:
|
||
"""Return (line_512, line_513, line_514) in cents.
|
||
|
||
line_512 = Σ(TRS_BASE_LINE_KEYS) - line_511
|
||
line_513 = line_513 (uncollectible for TRS, provided by filer)
|
||
line_514 = line_512 - line_513
|
||
"""
|
||
line_512 = sum(int(revenue_lines.get(k, 0) or 0) for k in TRS_BASE_LINE_KEYS)
|
||
line_512 -= int(revenue_lines.get("line_511", 0) or 0)
|
||
line_513 = int(revenue_lines.get("line_513", 0) or 0)
|
||
line_514 = line_512 - line_513
|
||
return line_512, line_513, line_514
|