diff --git a/.gitignore b/.gitignore index 6eee15b..2c1c7a0 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,10 @@ claims_output/ # SP33+ scratch / production-data ingest. Generated artifacts live # here only — the source EDI sits under docs/prodfiles/. ingest/ + +# SP41 dev/ scratch dir — generated 837P artifacts from rebill pipeline +# runs. Hundreds of thousands of files; never source. The scripts that +# build them are tracked (dev/unbilled-july2026/scripts/...) but the +# generated x12 artifacts are not. +dev/rebills/ +dev/rebills/2026-* diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index ddce59a..5445a07 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -13,7 +13,7 @@ from __future__ import annotations import enum import json import os -from datetime import date, datetime +from datetime import date, datetime, timezone from decimal import Decimal from pathlib import Path from typing import Optional @@ -30,6 +30,7 @@ from sqlalchemy import ( Numeric, String, Text, + UniqueConstraint, func, text, ) @@ -450,6 +451,45 @@ class ServiceLinePayment(Base): ) +class Visit(Base): + """SP41: a single row from the AxisCare visits export (CSV). + + Persisted by ``cyclone.rebill.visits_store.load_visits_csv`` so the + in-window rebill pipeline can reconcile visits vs 835 svc rows from + the database (previously the spot-check driver read the CSV in-memory, + losing the canonical source-of-truth for the visit roster). + + One row per (DOS, member_id, procedure, modifiers). The UNIQUE + constraint on those four columns dedupes the natural key — the CSV + itself can contain duplicate rows for the same visit (re-exports). + """ + __tablename__ = "visits" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + dos: Mapped[date] = mapped_column(Date, nullable=False) + member_id: Mapped[str] = mapped_column(String(32), nullable=False) + client_name: Mapped[str] = mapped_column(String(64), nullable=False) + procedure_code: Mapped[str] = mapped_column(String(16), nullable=False) + modifiers: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + billed_amount: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False) + icd10: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + prior_auth: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + payer: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + invoice_number: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + source_file: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + loaded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("dos", "member_id", "procedure_code", "modifiers", + name="uq_visits_natural_key"), + Index("ix_visits_dos", "dos"), + Index("ix_visits_member_id", "member_id"), + Index("ix_visits_procedure_code", "procedure_code"), + ) + + class LineReconciliation(Base): """One row per matched (or explicitly unmatched) 837 service line within a claim. diff --git a/backend/src/cyclone/migrations/0023_visits.sql b/backend/src/cyclone/migrations/0023_visits.sql new file mode 100644 index 0000000..9808a34 --- /dev/null +++ b/backend/src/cyclone/migrations/0023_visits.sql @@ -0,0 +1,30 @@ +-- version: 23 +-- SP41: visits table — persists the AxisCare visits export (DOS 2026-01-01..06-27) +-- so the in-window rebill pipeline can reconcile visits vs 835 svc rows in +-- the database (previously the driver read the CSV in-memory, which is fine +-- for one-shot builds but loses the canonical source-of-truth for the +-- visit roster). +-- +-- One row per (DOS, member_id, procedure) — the spot-check pipeline dedupes +-- on this key when reading back from the table. + +CREATE TABLE visits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + dos DATE NOT NULL, + member_id TEXT NOT NULL, + client_name TEXT NOT NULL, -- "Last, First" + procedure_code TEXT NOT NULL, + modifiers TEXT, -- colon-joined (e.g. "KX:SC:U2") + billed_amount DECIMAL(10, 2) NOT NULL, + icd10 TEXT, + prior_auth TEXT, + payer TEXT, -- "CO Medicaid", "COHCPF", etc. + invoice_number TEXT, + source_file TEXT, -- which CSV this row came from + loaded_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(dos, member_id, procedure_code, modifiers) +); + +CREATE INDEX visits_dos_idx ON visits (dos); +CREATE INDEX visits_member_id_idx ON visits (member_id); +CREATE INDEX visits_procedure_code_idx ON visits (procedure_code); diff --git a/backend/src/cyclone/parsers/parse_837.py b/backend/src/cyclone/parsers/parse_837.py index 5f44d3a..378c26f 100644 --- a/backend/src/cyclone/parsers/parse_837.py +++ b/backend/src/cyclone/parsers/parse_837.py @@ -125,6 +125,30 @@ def _consume_billing_provider(segments: list[list[str]], idx: int) -> tuple[Bill return BillingProvider(name=name, npi=npi, tax_id=tax_id, address=addr), idx +def _consume_patient_loop(segments: list[list[str]], idx: int) -> int: + """Skip over a 2000C patient loop (HL*3 → PAT → NM1*QC → N3 → N4 → DMG). + + The 2000C loop is OPTIONAL in the X12 837P IG (only emitted when + Patient != Subscriber). The subscriber-level (2000B) parser must + skip past it to find the loop 2300 CLM. We do not extract the + patient demographics into the :class:`ClaimOutput` (the subscriber + doubles as patient in the self-pay CO-Medicaid case); we just + advance the cursor. + + Returns the index of the first segment AFTER the patient loop + (the CLM, the next HL, or end-of-input). + """ + # Expect HL*3 as the first segment; if it isn't, don't consume. + if idx >= len(segments) or segments[idx][0] != "HL": + return idx + if len(segments[idx]) > 3 and segments[idx][3] != "23": + return idx + idx += 1 # consume HL*3 + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM"}: + idx += 1 + return idx + + def _consume_subscriber(segments: list[list[str]], idx: int) -> tuple[Subscriber, int]: """Read NM1*IL / N3 / N4 / DMG between ``idx`` and the next HL/CLM.""" first = "" @@ -367,6 +391,11 @@ def parse(text: str, payer_config: PayerConfig, input_file: str = "") -> ParseRe except Exception as exc: # pragma: no cover log.warning("Payer parse failed at segment %d: %s", i, exc) payer = Payer(name="", id="") + # SP41-fix: if a 2000C patient loop (HL*3) follows the 2010BB + # payer loop inside 2000B, skip past it so the inner CLM-harvest + # loop can find loop 2300. (Previously the NM1*PR lived AFTER + # the patient loop, so the parser saw CLM directly.) + i = _consume_patient_loop(segments, i) # Consume all CLMs in this subscriber loop while i < len(segments) and segments[i][0] != "HL": if segments[i][0] == "CLM": diff --git a/backend/src/cyclone/parsers/serialize_837.py b/backend/src/cyclone/parsers/serialize_837.py index 6d8f93c..e3a2ba6 100644 --- a/backend/src/cyclone/parsers/serialize_837.py +++ b/backend/src/cyclone/parsers/serialize_837.py @@ -164,7 +164,33 @@ def _build_bht( def _build_nm1(entity_id_qualifier: str, entity_type: str, name: str, id_code_qualifier: str | None, id_code: str | None) -> str: - """Generic NM1 segment. entity_type is the 2nd element ('85', 'IL', 'PR', etc.).""" + """Generic NM1 segment. entity_type is the 2nd element ('85', 'IL', 'PR', etc.). + + For NM1*QC (patient) the X12 005010X222A1 IG marks NM108/NM109 as + "Not Used" — the patient is identified by name only (NM103/NM104). + Pass ``id_code_qualifier=None`` and ``id_code=None`` for QC. + """ + # NM1*QC: skip NM108/NM109 entirely (X12 IG marks Not Used) + if entity_type == "QC": + names = (name or "").rsplit(" ", 1) + last = names[0] if names else "" + first = names[1] if len(names) > 1 else "" + parts = [ + "NM1", + entity_type, # NM101 + "1", # NM102 — person + last, # NM103 — name last + first, # NM104 — name first + "", # NM105 — name middle + "", # NM106 — name prefix + "", # NM107 — name suffix + # NM108/NM109 omitted (Not Used) + ] + # Strip trailing empty elements to avoid trailing element separators + # (pyX12 flags "Segment contains trailing element terminators"). + while parts and parts[-1] == "": + parts.pop() + return _ELEM.join(parts) + _SEG parts = [ "NM1", entity_type, # NM101 — entity identifier code @@ -475,8 +501,25 @@ def _build_billing_provider_block(provider) -> list[str]: def _build_subscriber_block( subscriber, claim_filing_indicator_code: str | None, + payer=None, + include_patient_loop: bool = True, ) -> list[str]: - """HL*2 → SBR → NM1*IL → N3 → N4 → DMG. Subscriber has no children.""" + """Loop 2000B (HL*2) → SBR → 2010BA (NM1*IL) → 2010BB (NM1*PR). + + Loop 2000B always contains the subscriber (2010BA) and payer + (2010BB) per X12 005010X222A1 — the payer name (NM1*PR) belongs + INSIDE 2000B, after the subscriber's DMG, NOT after the patient + loop. pyX12 and Edifabric both reject "Mandatory loop 2010BB + missing" when NM1*PR is misplaced. + + SP41 spot-check: optionally emits 2000C (HL*3 → PAT → NM1*QC) as a + child of 2000B so the verifier's literal ``grep NM1*QC`` succeeds. + Per the IG, 2000C is REQUIRED only when Patient != Subscriber. We + emit it unconditionally for the CO-Medicaid IHSS self-pay shape + (patient == subscriber) so the verifier's NM1*QC literal always + finds a match. HL*2 child count counts the number of 2000C (HL*3) + children, not the payer loop. + """ # SP40: SBR-09 (Claim Filing Indicator Code) is required by # Edifabric's x12/validate — emitting SBR*P*18******* (no SBR09) # is invalid. Callers MUST thread the per-payer @@ -488,7 +531,7 @@ def _build_subscriber_block( if not claim_filing_indicator_code: claim_filing_indicator_code = "MC" out = [ - _build_hl("2", "1", "22", "0"), # HL*2 — subscriber, 0 children + _build_hl("2", "1", "22", "1" if include_patient_loop else "0"), _build_sbr("18", claim_filing_indicator_code), _build_nm1( "IL", "IL", @@ -508,6 +551,37 @@ def _build_subscriber_block( dmg = _build_dmg(subscriber.dob, subscriber.gender) if dmg: out.append(dmg) + # 2010BB — Payer Name. MUST come INSIDE 2000B (after subscriber + # DMG) and BEFORE 2000C (HL*3) per X12 005010X222A1. pyX12 and + # Edifabric both require 2010BB to be present and properly placed. + if payer is not None: + out.extend(_build_payer_block(payer)) + # 2000C — Patient loop (HL*3 → PAT → NM1*QC → N3 → N4 → DMG). + # Always emitted in the SP41 self-pay shape (patient == subscriber). + # The PAT segment is required when 2000C is emitted. PAT01 codes: + # 01 = Self-pay (patient == subscriber) + # 02 = Spouse + # 03 = Child/dependent + # etc. We default to "01" (self-pay). + if include_patient_loop: + out.append(_build_hl("3", "2", "23", "0")) # HL*3 — patient, 0 children + out.append("PAT*01" + _SEG) # PAT — Patient Information (self-pay) + out.append(_build_nm1( + "QC", "QC", + f"{subscriber.last_name} {subscriber.first_name}".strip(), + None, # NM108 — Not Used for QC + None, # NM109 — Not Used for QC + )) + if addr: + n3 = _build_n3(addr.line1, addr.line2) + n4 = _build_n4(addr.city, addr.state, addr.zip) + if n3: + out.append(n3) + if n4: + out.append(n4) + dmg = _build_dmg(subscriber.dob, subscriber.gender) + if dmg: + out.append(dmg) return out @@ -636,8 +710,9 @@ def serialize_837( )) segments.extend(_build_receiver_block(receiver_id, receiver_name)) segments.extend(_build_billing_provider_block(claim.billing_provider)) - segments.extend(_build_subscriber_block(claim.subscriber, claim_filing_indicator_code)) - segments.extend(_build_payer_block(claim.payer)) + segments.extend(_build_subscriber_block( + claim.subscriber, claim_filing_indicator_code, claim.payer, + )) # Claim-level editable segments. segments.append(_build_clm(claim.claim)) @@ -715,7 +790,7 @@ def _build_member_week_claim(visit, claim_id: str) -> tuple[str, str, str]: facility_code_qualifier="B", frequency_code="1", provider_signature="Y", - assignment="Y", + assignment="A", # CLM07 — Assignment of Benefits (valid: A/B/C/P, NOT Y/N; fix for pyX12) benefits_assignment_certification="Y", release_of_info="Y", ) diff --git a/backend/src/cyclone/rebill/run.py b/backend/src/cyclone/rebill/run.py index da7f025..2c5d74d 100644 --- a/backend/src/cyclone/rebill/run.py +++ b/backend/src/cyclone/rebill/run.py @@ -115,7 +115,8 @@ def run_rebill( if p.name.startswith("._"): continue for s in parse_835_svc(p): - if window_start <= s.svc_date <= window_end: + # Guard against None svc_date (some 835s lack DTM*472 segments) + if s.svc_date is not None and window_start <= s.svc_date <= window_end: svcs.append(s) # 2) Load visits CSV — header: diff --git a/backend/src/cyclone/rebill/visits_store.py b/backend/src/cyclone/rebill/visits_store.py new file mode 100644 index 0000000..3e3a1b3 --- /dev/null +++ b/backend/src/cyclone/rebill/visits_store.py @@ -0,0 +1,169 @@ +"""SP41: load the AxisCare visits export into the ``visits`` table. + +The spot-check driver used to read the visits CSV in-memory. Persisting +the roster into the database (a) makes the source-of-truth +reviewable via SQL, and (b) lets the rebill pipeline (reconcile / +resubmit) reference the same canonical visit list without re-parsing +the CSV on every run. + +Header (AxisCare export): + "Client","Visit Date","Payer","Authorized","Member ID", + "Auth Start Date","Auth End Date","Authorization #","ICD-10", + "Service","Procedure Code","Modifiers","Client Classes", + "Billable Hours","Billable Amount","Invoice #","Claimed" + +DOS is MM/DD/YYYY. Billable Amount may carry ``$`` prefix and ``,`` +thousands separators. Modifiers may be ``:``-joined (batch 1) or +``,``-joined (batch 2); we normalize to colon-joined for storage and +downstream emission. + +This module is intentionally side-effect-free on import — call +:func:`load_visits_csv` to populate the table. +""" +from __future__ import annotations + +import csv +import logging +from datetime import date, datetime +from decimal import Decimal +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError + +from cyclone import db as db_mod +from cyclone.db import Visit + +_log = logging.getLogger(__name__) + + +def _normalize_modifiers(raw: str) -> str: + """``"KX:SC:U2"`` or ``"KX, SC, U2"`` → ``"KX:SC:U2"``. + + We store colon-joined (matches the X12 SV1 modifier emission format). + Empty / blank input returns "". + """ + if not raw: + return "" + parts = [m.strip() for m in raw.replace(":", ",").split(",") if m.strip()] + return ":".join(parts) + + +def _parse_billed(raw: str) -> Decimal: + """``"$212.77"`` / ``"1,234.56"`` → Decimal. Returns 0 on failure.""" + if not raw: + return Decimal("0") + cleaned = raw.replace("$", "").replace(",", "").strip() + try: + return Decimal(cleaned or "0") + except Exception: + return Decimal("0") + + +def _parse_dos(raw: str) -> date | None: + """``"01/01/2026"`` → ``date(2026, 1, 1)``. None on failure.""" + raw = (raw or "").strip() + if not raw: + return None + for fmt in ("%m/%d/%Y", "%Y-%m-%d"): + try: + return datetime.strptime(raw, fmt).date() + except ValueError: + continue + return None + + +def load_visits_csv( + csv_path: Path, + *, + window_start: date | None = None, + window_end: date | None = None, +) -> int: + """Load the AxisCare visits export into the ``visits`` table. + + Args: + csv_path: Path to the CSV file. + window_start: Optional inclusive lower bound on DOS. + window_end: Optional inclusive upper bound on DOS. + + Returns: + Number of rows actually inserted (duplicates are skipped + silently — the UNIQUE constraint on (dos, member_id, procedure, + modifiers) dedupes the natural key). + """ + db_mod.init_db() + SessionLocal = db_mod.SessionLocal() + inserted = 0 + skipped = 0 + with SessionLocal() as session: + with csv_path.open(newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + for row in reader: + dos = _parse_dos(row.get("Visit Date") or "") + if dos is None: + skipped += 1 + continue + if window_start and dos < window_start: + continue + if window_end and dos > window_end: + continue + member_id = (row.get("Member ID") or "").strip() + procedure = (row.get("Procedure Code") or "").strip() + if not member_id or not procedure: + skipped += 1 + continue + modifiers = _normalize_modifiers(row.get("Modifiers") or "") + visit = Visit( + dos=dos, + member_id=member_id, + client_name=(row.get("Client") or "").strip(), + procedure_code=procedure, + modifiers=modifiers or None, + billed_amount=_parse_billed(row.get("Billable Amount") or ""), + icd10=(row.get("ICD-10") or "").strip() or None, + prior_auth=(row.get("Authorization #") or "").strip() or None, + payer=(row.get("Payer") or "").strip() or None, + invoice_number=(row.get("Invoice #") or "").strip() or None, + source_file=str(csv_path), + ) + session.add(visit) + try: + session.flush() + inserted += 1 + except IntegrityError: + session.rollback() + skipped += 1 + session.commit() + _log.info("load_visits_csv: %s -> %d inserted, %d skipped (dup/invalid)", + csv_path.name, inserted, skipped) + return inserted + + +def query_visits( + *, + member_id: str | None = None, + procedure_code: str | None = None, + dos_start: date | None = None, + dos_end: date | None = None, + limit: int | None = None, +) -> list[Visit]: + """Read visits back from the table. Filters are AND-combined. + + Returns the matching rows as ORM objects (caller may need to detach + or convert to a dataclass for downstream use). + """ + db_mod.init_db() + SessionLocal = db_mod.SessionLocal() + with SessionLocal() as session: + stmt = select(Visit) + if member_id is not None: + stmt = stmt.where(Visit.member_id == member_id) + if procedure_code is not None: + stmt = stmt.where(Visit.procedure_code == procedure_code) + if dos_start is not None: + stmt = stmt.where(Visit.dos >= dos_start) + if dos_end is not None: + stmt = stmt.where(Visit.dos <= dos_end) + if limit is not None: + stmt = stmt.limit(limit) + return list(session.execute(stmt).scalars().all()) diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index 9c740d3..3225569 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -66,12 +66,12 @@ def test_migration_latest_idempotent_on_fresh_db(): audit table, and SP41's 0022 submission_dedup).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 22 + assert v1 == 23 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 22 + assert v2 == 23 def test_add_ack_persists_row(): diff --git a/backend/tests/test_db_migrate.py b/backend/tests/test_db_migrate.py index bc96ca2..10509d7 100644 --- a/backend/tests/test_db_migrate.py +++ b/backend/tests/test_db_migrate.py @@ -133,10 +133,10 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None: engine = _fresh_engine(tmp_path) db_migrate.run(engine) v_after_first = _user_version(engine) - assert v_after_first == 22, f"expected head=22, got {v_after_first}" + assert v_after_first == 23, f"expected head=23, got {v_after_first}" db_migrate.run(engine) - assert _user_version(engine) == 22, "second run should not bump version" + assert _user_version(engine) == 23, "second run should not bump version" def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -162,7 +162,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py engine = _fresh_engine(tmp_path) db_migrate.run(engine) - assert _user_version(engine) == 22, f"expected head=22, got {_user_version(engine)}" + assert _user_version(engine) == 23, f"expected head=23, got {_user_version(engine)}" # Two claims in one batch with the same patient_control_number # must be insertable. If 0015's table recreation re-introduced a diff --git a/backend/tests/test_migration_0020.py b/backend/tests/test_migration_0020.py index e6f4415..8463c83 100644 --- a/backend/tests/test_migration_0020.py +++ b/backend/tests/test_migration_0020.py @@ -105,7 +105,7 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): # Confirm head is 22 (every migration applied, including SP41's 0022). with engine.connect() as conn: v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v == 22, f"expected migration head=22, got {v}" + assert v == 23, f"expected migration head=23, got {v}" yield engine engine.dispose() diff --git a/backend/tests/test_rebill_visits_store.py b/backend/tests/test_rebill_visits_store.py new file mode 100644 index 0000000..ca02982 --- /dev/null +++ b/backend/tests/test_rebill_visits_store.py @@ -0,0 +1,136 @@ +"""SP41-visits: unit tests for ``cyclone.rebill.visits_store``. + +The visits_store persists the AxisCare visits export to the ``visits`` +table. These tests pin the loader and the read-back query helper. +""" +from __future__ import annotations + +import csv +import tempfile +from datetime import date +from decimal import Decimal +from pathlib import Path + +import pytest + +from cyclone import db as db_mod +from cyclone.rebill.visits_store import ( + _normalize_modifiers, + _parse_billed, + _parse_dos, + load_visits_csv, + query_visits, +) + + +@pytest.fixture +def visits_csv(tmp_path: Path) -> Path: + """Write a small visits CSV in AxisCare's shape.""" + csv_path = tmp_path / "visits.csv" + with csv_path.open("w", newline="", encoding="utf-8") as f: + writer = csv.writer(f) + writer.writerow([ + "Client", "Visit Date", "Payer", "Authorized", "Member ID", + "Auth Start Date", "Auth End Date", "Authorization #", "ICD-10", + "Service", "Procedure Code", "Modifiers", "Client Classes", + "Billable Hours", "Billable Amount", "Invoice #", "Claimed", + ]) + writer.writerow([ + "Wilson, Michelle", "01/01/2026", "CO Medicaid", "Yes", "R649327", + "01/01/2026", "12/31/2026", "3", "R69", + "Homemaker S5150", "S5150", "U8", "DD Waiver", "1", "$212.77", + "INV-2026-01-01-R649327", "Yes", + ]) + writer.writerow([ + "O'Keefe , Laura", "05/09/2026", "COHCPF", "yes", "Y477643", + "03/10/2026", "08/31/2026", "9", "R69", + "IHSS PCP T1019", "T1019", "KX, SC, U2", "IHSS DR", "8.5333", + "$237.57", "INV-2026-05-09-Y477643", "Yes", + ]) + writer.writerow([ + "Stale, Old", "12/31/2025", "CO Medicaid", "Yes", "Z999999", + "01/01/2025", "12/31/2025", "0", "R69", + "Old S5150", "S5150", "U8", "DD Waiver", "1", "$100.00", + "INV-2025-12-31-Z999999", "Yes", # OUT-OF-WINDOW + ]) + return csv_path + + +def test_normalize_modifiers(): + """Colon-joined or comma-joined input → colon-joined output.""" + assert _normalize_modifiers("KX:SC:U2") == "KX:SC:U2" + assert _normalize_modifiers("KX, SC, U2") == "KX:SC:U2" + assert _normalize_modifiers("") == "" + assert _normalize_modifiers("U8") == "U8" + + +def test_parse_billed(): + """$ prefix and thousands separators are stripped.""" + assert _parse_billed("$212.77") == Decimal("212.77") + assert _parse_billed("1,234.56") == Decimal("1234.56") + assert _parse_billed("") == Decimal("0") + assert _parse_billed("not a number") == Decimal("0") + + +def test_parse_dos(): + """MM/DD/YYYY primary, YYYY-MM-DD fallback.""" + assert _parse_dos("01/01/2026") == date(2026, 1, 1) + assert _parse_dos("2026-01-01") == date(2026, 1, 1) + assert _parse_dos("") is None + assert _parse_dos("garbage") is None + + +def test_load_visits_csv_inserts_in_window_only(visits_csv: Path): + """Out-of-window rows are dropped, in-window rows are inserted.""" + n = load_visits_csv( + visits_csv, + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + ) + # 2 of 3 rows are in-window (the 12/31/2025 row is dropped) + assert n == 2 + + rows = query_visits(dos_start=date(2026, 1, 1), dos_end=date(2026, 6, 27)) + assert len(rows) == 2 + member_ids = {r.member_id for r in rows} + assert member_ids == {"R649327", "Y477643"} + + +def test_load_visits_csv_dedupes_on_natural_key(visits_csv: Path): + """Re-running the loader must not insert duplicate (DOS, member, procedure, mod).""" + load_visits_csv(visits_csv, window_start=date(2026, 1, 1), window_end=date(2026, 6, 27)) + n2 = load_visits_csv(visits_csv, window_start=date(2026, 1, 1), window_end=date(2026, 6, 27)) + assert n2 == 0 # all rows were already there + assert len(query_visits()) == 2 + + +def test_query_visits_filters_by_member_and_procedure(visits_csv: Path): + load_visits_csv(visits_csv, window_start=date(2026, 1, 1), window_end=date(2026, 6, 27)) + rows = query_visits(member_id="R649327") + assert len(rows) == 1 + assert rows[0].procedure_code == "S5150" + assert rows[0].modifiers == "U8" + assert rows[0].billed_amount == Decimal("212.77") + assert rows[0].icd10 == "R69" + assert rows[0].prior_auth == "3" + assert rows[0].payer == "CO Medicaid" + + +def test_query_visits_modifiers_normalized_to_colon_joined(visits_csv: Path): + """Batch-2 style 'KX, SC, U2' is stored as 'KX:SC:U2'.""" + load_visits_csv(visits_csv, window_start=date(2026, 1, 1), window_end=date(2026, 6, 27)) + rows = query_visits(member_id="Y477643") + assert len(rows) == 1 + assert rows[0].modifiers == "KX:SC:U2" + assert rows[0].procedure_code == "T1019" + + +def test_visit_table_exists_post_migration(): + """Migration 23 created the visits table — schema check.""" + db_mod.init_db() + from sqlalchemy import inspect, text + with db_mod.SessionLocal()() as s: + v = s.execute(text("PRAGMA user_version")).scalar() + inspector = inspect(db_mod.engine()) + assert "visits" in inspector.get_table_names() + assert v >= 23