Merged in feature/active-rates (pull request #1004)
Feature/active rates * initial commit * Merged in feature/FixplaceholderIssue (pull request #977) Remove curly braces from echo statements in dev->stg * Remove curly brances from echo statements in dev->stg * Removed curly braces in echo statements in feature-> dev gate Approved-by: Sujit Deokar * Merged in feature/standardized-services (pull request #958) Feature/standardized services * service term standardization * prompt update * prompt updates for standardization * only service standardization * new file * add supporting files and test scripts for standardization work Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * remove old files * Merge remote-tracking branch 'origin/dev' into feature/standardized-services * final fixes * Merged dev into feature/standardized-services * additional features * removed unwanted files * remove unwanted files * Merge branch 'dev' into feature/standardized-services * Merge remote-tracking branch 'origin/dev' into feature/standardized-services * reversed vendor changes * Merged dev into feature/standardized-services * addressed PR comments * Merge branch 'dev' into feature/standardized-services * Merged dev into feature/standardized-services * addressed 3 remaining comments inPR * black formatting * incorporated feedback from AI c… * prompt added for amendment intent * amendment intent language * update field name * amendment intent tag * prompt update * Merge remote-tracking branch 'origin/dev' into feature/active-rates * Merge remote-tracking branch 'origin/dev' into feature/active-rates * Merge branch 'dev' into feature/active-rates * exhibit standardization updates * use AARETE_DERIVED_EXHIBIT_TITLE for intent * active rates stuff * Merge branch 'dev' into feature/active-rates * prompt update * amendment intent types * active rates logic update * unit tests * Merged dev into feature/active-rates * logic updates * Merge remote-tracking branch 'origin/dev' into feature/active-rates * Merge remote-tracking branch 'origin/dev' into feature/active-rates * Merged dev into feature/active-rates Approved-by: Katon Minhas
This commit is contained in:
Karan Desai
committed by
Katon Minhas
Katon Minhas
parent
2306334b40
commit
830d228fb3
@@ -29,6 +29,8 @@ FIELD_FORMAT_MAPPING = {
|
||||
"CONTRACT_AMENDMENT_NUM": "str",
|
||||
"FILENAME_AMENDMENT_NUM": "str", # Default: not in original mapping
|
||||
"AARETE_DERIVED_AMENDMENT_NUM": "str",
|
||||
"AMENDMENT_INTENT": "str",
|
||||
"AMENDMENT_INTENT_LANGUAGE": "list[str]",
|
||||
"CLIENT_NAME": "str",
|
||||
"CLIENT_STATE": "list[str]",
|
||||
"PAYER_NAME": "str",
|
||||
|
||||
+32
-8
@@ -48,6 +48,10 @@ from src.parent_child import main as parent_child_main
|
||||
from src.pipelines.shared.postprocessing.service_term_standardization import (
|
||||
standardize_service_terms,
|
||||
)
|
||||
from src.pipelines.shared.postprocessing.active_rates import (
|
||||
run_active_rates_postprocessing,
|
||||
write_active_rates_output,
|
||||
)
|
||||
from src.document_classification import main as dtc_main
|
||||
from src.qc_qa.pipeline import (
|
||||
generate_statistics,
|
||||
@@ -553,14 +557,34 @@ def main(client: str = "saas", testing=False, test_params={}):
|
||||
)
|
||||
else:
|
||||
logging.info("Service Term standardization complete.")
|
||||
if config.WRITE_TO_S3:
|
||||
io_utils.write_s3(
|
||||
FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full"
|
||||
)
|
||||
else:
|
||||
io_utils.write_local(
|
||||
FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full"
|
||||
)
|
||||
|
||||
# ── Active Rates postprocessing ───────────────────────────────
|
||||
# Merges PC columns (GROUP/QC_RANK/QC_FLAG), standardises exhibit
|
||||
# titles, derives amendment intent, then assigns MEMBER_TYPE /
|
||||
# FAMILY_TAG / ACTIVE_RATE.
|
||||
logging.info("=" * 80)
|
||||
logging.info("Running Active Rates postprocessing...")
|
||||
logging.info("=" * 80)
|
||||
with timing_utils.timed_block(
|
||||
"active_rates_postprocessing", log_level="INFO"
|
||||
):
|
||||
try:
|
||||
FINAL_RESULT_DF_CC = run_active_rates_postprocessing(
|
||||
FINAL_RESULT_DF_CC, pc_details_df
|
||||
)
|
||||
write_active_rates_output(
|
||||
FINAL_RESULT_DF_CC,
|
||||
run_timestamp,
|
||||
write_to_s3=config.WRITE_TO_S3,
|
||||
)
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"Active Rates postprocessing failed: {e}", exc_info=True
|
||||
)
|
||||
logging.warning(
|
||||
"Continuing without Active Rates columns (ACTIVE_RATE, "
|
||||
"MEMBER_TYPE, FAMILY_TAG, AARETE_DERIVED_EXHIBIT_TITLE)."
|
||||
)
|
||||
|
||||
if not per_file_usage_df.empty and not batch_summary_df.empty:
|
||||
logging.info("Exporting usage and cost tracking data locally...")
|
||||
|
||||
@@ -0,0 +1,541 @@
|
||||
"""
|
||||
Active Rates postprocessing pipeline.
|
||||
|
||||
Orchestrates the full active-rates flow that runs after parent-child mapping:
|
||||
|
||||
1. merge_parent_child_columns — join GROUP / QC_RANK / QC_FLAG onto the main output
|
||||
2. _standardize_and_derive_intent
|
||||
a. standardize_exhibit_titles → AARETE_DERIVED_EXHIBIT_TITLE (2-pass LLM)
|
||||
b. add_amendment_intent → AMENDMENT_INTENT (LLM per exhibit)
|
||||
3. add_active_rate_flags — deterministic logic → MEMBER_TYPE / FAMILY_TAG / ACTIVE_RATE
|
||||
|
||||
Public entry point: run_active_rates_postprocessing(main_df, pc_df)
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src import config
|
||||
|
||||
# PC columns merged into the main output from the parent-child details df
|
||||
_PC_MERGE_COLS = ["FILE_NAME", "GROUP", "QC_RANK", "QC_FLAG"]
|
||||
|
||||
# Columns written to the "focused cols" sheet of the active-rates Excel output
|
||||
_FOCUSED_COLS = [
|
||||
"FILE_NAME",
|
||||
"EXHIBIT_TITLE",
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE",
|
||||
"AARETE_DERIVED_SERVICE_TERM",
|
||||
"AMENDMENT_INTENT_LANGUAGE",
|
||||
"AMENDMENT_INTENT",
|
||||
"AMENDMENT_INTENT_REASON",
|
||||
"GROUP",
|
||||
"QC_RANK",
|
||||
"ACTIVE_RATE",
|
||||
]
|
||||
|
||||
|
||||
# ── Step 1: Merge PC columns ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def merge_parent_child_columns(
|
||||
main_df: pd.DataFrame, pc_df: pd.DataFrame
|
||||
) -> pd.DataFrame:
|
||||
"""Left-join GROUP, QC_RANK, QC_FLAG from pc_df into main_df on FILE_NAME.
|
||||
|
||||
pc_df is the details DataFrame returned by parent_child_main.main(). It has
|
||||
one row per FILE_NAME; every row in main_df for a given file receives the same
|
||||
PC classification columns.
|
||||
"""
|
||||
if pc_df is None or pc_df.empty:
|
||||
logging.warning("PC details df is empty; skipping PC column merge.")
|
||||
return main_df
|
||||
|
||||
available = [c for c in _PC_MERGE_COLS if c in pc_df.columns]
|
||||
merge_cols = [c for c in available if c != "FILE_NAME"]
|
||||
|
||||
if "FILE_NAME" not in available or not merge_cols:
|
||||
logging.warning(
|
||||
"PC details df missing FILE_NAME or all merge columns; skipping merge. "
|
||||
f"Available columns: {pc_df.columns.tolist()}"
|
||||
)
|
||||
return main_df
|
||||
|
||||
# Drop pre-existing PC columns from main_df to avoid _x/_y suffix conflicts
|
||||
cols_to_drop = [c for c in merge_cols if c in main_df.columns]
|
||||
if cols_to_drop:
|
||||
main_df = main_df.drop(columns=cols_to_drop)
|
||||
|
||||
# Deduplicate pc_df on FILE_NAME (one assignment per file)
|
||||
pc_subset = pc_df[available].drop_duplicates(subset=["FILE_NAME"])
|
||||
|
||||
merged = main_df.merge(pc_subset, on="FILE_NAME", how="left")
|
||||
logging.info(
|
||||
f"Merged PC columns {merge_cols} into main output "
|
||||
f"({len(merged):,} rows, "
|
||||
f"{pc_subset['FILE_NAME'].nunique()} unique files in PC output)"
|
||||
)
|
||||
return merged
|
||||
|
||||
|
||||
# ── Step 2: Exhibit title standardisation + amendment intent ─────────────────
|
||||
|
||||
|
||||
def _standardize_and_derive_intent(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Run exhibit title standardisation then amendment intent derivation.
|
||||
|
||||
Combines the logic from:
|
||||
- src/pipelines/shared/postprocessing/exhibit_title_standardization.py
|
||||
(EXHIBIT_TITLE → AARETE_DERIVED_EXHIBIT_TITLE, grouped by GROUP)
|
||||
- postprocessing_funcs.add_amendment_intent
|
||||
(AARETE_DERIVED_EXHIBIT_TITLE + AMENDMENT_INTENT_LANGUAGE → AMENDMENT_INTENT)
|
||||
"""
|
||||
from src.pipelines.shared.postprocessing.exhibit_title_standardization import (
|
||||
standardize_exhibit_titles,
|
||||
)
|
||||
from src.pipelines.shared.postprocessing.postprocessing_funcs import (
|
||||
add_amendment_intent,
|
||||
)
|
||||
|
||||
logging.info("Active Rates — Step 2a: Standardizing exhibit titles...")
|
||||
df = standardize_exhibit_titles(df)
|
||||
|
||||
logging.info("Active Rates — Step 2b: Deriving amendment intent (LLM)...")
|
||||
df = add_amendment_intent(df)
|
||||
|
||||
return df
|
||||
|
||||
|
||||
# ── Step 3: Active rate flags ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def add_active_rate_flags(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Add MEMBER_TYPE, FAMILY_TAG, and ACTIVE_RATE to df.
|
||||
|
||||
Required columns (all uppercase, consistent with the main pipeline):
|
||||
QC_RANK, QC_FLAG, GROUP, FILE_NAME,
|
||||
AARETE_DERIVED_EXHIBIT_TITLE, AMENDMENT_INTENT, AMENDMENT_INTENT_LANGUAGE
|
||||
"""
|
||||
required = {"QC_RANK", "QC_FLAG", "GROUP", "AARETE_DERIVED_EXHIBIT_TITLE"}
|
||||
missing = required - set(df.columns)
|
||||
if missing:
|
||||
logging.warning(
|
||||
f"add_active_rate_flags: missing required columns {missing}; skipping."
|
||||
)
|
||||
return df
|
||||
|
||||
df = df.copy()
|
||||
|
||||
if "AMENDMENT_INTENT_REASON" not in df.columns:
|
||||
df["AMENDMENT_INTENT_REASON"] = None
|
||||
|
||||
# ── Rank helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
def _clean_rank(raw) -> str | None:
|
||||
"""Normalise QC_RANK: Excel float "1.0" → "1"."""
|
||||
if pd.isna(raw):
|
||||
return None
|
||||
s = str(raw).strip()
|
||||
if not s:
|
||||
return None
|
||||
if s.count(".") == 1:
|
||||
left, right = s.split(".")
|
||||
if right == "0":
|
||||
s = left
|
||||
return s
|
||||
|
||||
def _rank_tuple(s: str | None) -> tuple | None:
|
||||
if s is None:
|
||||
return None
|
||||
try:
|
||||
return tuple(int(x) for x in s.split("."))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
def _effective_rank(rt: tuple | None) -> tuple | None:
|
||||
"""For children (3+ segments) keep only (first, last) to find latest version."""
|
||||
if rt is None:
|
||||
return None
|
||||
if len(rt) <= 2:
|
||||
return rt
|
||||
return (rt[0], rt[-1])
|
||||
|
||||
def _parent_prefix(s: str | None) -> str | None:
|
||||
if s is None:
|
||||
return None
|
||||
return s.split(".")[0]
|
||||
|
||||
df["_rank_clean"] = df["QC_RANK"].apply(_clean_rank)
|
||||
df["_rank_tuple"] = df["_rank_clean"].apply(_rank_tuple)
|
||||
df["_eff_rank"] = df["_rank_tuple"].apply(_effective_rank)
|
||||
df["_parent_pfx"] = df["_rank_clean"].apply(_parent_prefix)
|
||||
|
||||
# ── Normalise AMENDMENT_INTENT_LANGUAGE ───────────────────────────────────
|
||||
|
||||
if "AMENDMENT_INTENT_LANGUAGE" in df.columns:
|
||||
df["AMENDMENT_INTENT_LANGUAGE"] = df["AMENDMENT_INTENT_LANGUAGE"].replace(
|
||||
to_replace=r'(?si)^\s*\[\s*"?\s*N/A\s*"?\s*\]\s*$',
|
||||
value=np.nan,
|
||||
regex=True,
|
||||
)
|
||||
|
||||
# ── Clear amendment fields on parent rows ─────────────────────────────────
|
||||
|
||||
amend_cols = [
|
||||
c for c in ["AMENDMENT_INTENT", "AMENDMENT_INTENT_LANGUAGE"] if c in df.columns
|
||||
]
|
||||
if amend_cols:
|
||||
df.loc[df["QC_FLAG"] == "Parent", amend_cols] = None
|
||||
df.loc[df["QC_FLAG"] == "Parent", "AMENDMENT_INTENT_REASON"] = (
|
||||
"parent_row_cleared"
|
||||
)
|
||||
|
||||
# ── Fallback: fill AMENDMENT_INTENT for files with no language at all ─────
|
||||
# add_amendment_intent (LLM step) sets "N/A" for empty-language rows.
|
||||
# Files with no amendment language are classified as OVERLAY_UPDATE for manual grouping
|
||||
# by service term rather than exhibit title.
|
||||
|
||||
if "AMENDMENT_INTENT_LANGUAGE" in df.columns and "AMENDMENT_INTENT" in df.columns:
|
||||
_files_no_lang = set(
|
||||
df.groupby("FILE_NAME")["AMENDMENT_INTENT_LANGUAGE"]
|
||||
.apply(lambda s: s.isna().all())
|
||||
.pipe(lambda x: x[x].index)
|
||||
)
|
||||
|
||||
_not_parent = (df["QC_FLAG"] != "Parent") if "QC_FLAG" in df.columns else True
|
||||
_eligible = (
|
||||
(
|
||||
df["AMENDMENT_INTENT"].isna()
|
||||
| df["AMENDMENT_INTENT"].isin(["N/A", "n/a"])
|
||||
)
|
||||
& df["FILE_NAME"].isin(_files_no_lang)
|
||||
& _not_parent
|
||||
)
|
||||
|
||||
df.loc[_eligible, "AMENDMENT_INTENT"] = "OVERLAY_UPDATE"
|
||||
df.loc[_eligible, "AMENDMENT_INTENT_REASON"] = "no_language_in_file"
|
||||
|
||||
# ── Classify MEMBER_TYPE ──────────────────────────────────────────────────
|
||||
|
||||
def _classify(row) -> str:
|
||||
pfx = row["_parent_pfx"]
|
||||
if pfx is None:
|
||||
return "unknown"
|
||||
if pfx == "0":
|
||||
return "orphan"
|
||||
rt = row["_rank_tuple"]
|
||||
if rt is None:
|
||||
return "unknown"
|
||||
return "parent" if len(rt) == 1 else "child"
|
||||
|
||||
df["MEMBER_TYPE"] = df.apply(_classify, axis=1)
|
||||
|
||||
# ── FAMILY_TAG ────────────────────────────────────────────────────────────
|
||||
|
||||
eligible_mask = df["MEMBER_TYPE"].isin(["parent", "child"])
|
||||
|
||||
df["FAMILY_TAG"] = pd.NA
|
||||
df.loc[eligible_mask, "FAMILY_TAG"] = (
|
||||
df.loc[eligible_mask, "GROUP"].astype(str)
|
||||
+ "__"
|
||||
+ df.loc[eligible_mask, "_parent_pfx"]
|
||||
)
|
||||
|
||||
# Parent-only groups: all files in a GROUP are parents → collapse FAMILY_TAG to GROUP
|
||||
_parent_only_groups = {
|
||||
grp_name
|
||||
for grp_name, grp_df in df[eligible_mask].groupby("GROUP", dropna=False)
|
||||
if (grp_df["MEMBER_TYPE"] == "parent").all()
|
||||
}
|
||||
if _parent_only_groups:
|
||||
_po_mask = eligible_mask & df["GROUP"].isin(_parent_only_groups)
|
||||
df.loc[_po_mask, "FAMILY_TAG"] = df.loc[_po_mask, "GROUP"].astype(str)
|
||||
|
||||
# ── ACTIVE_RATE ───────────────────────────────────────────────────────────
|
||||
|
||||
df["ACTIVE_RATE"] = pd.NA
|
||||
|
||||
# For PARTIAL_REPLACEMENT / OVERLAY_UPDATE, group by service term instead of exhibit title
|
||||
df["_exhibit_key"] = df["AARETE_DERIVED_EXHIBIT_TITLE"]
|
||||
if "AMENDMENT_INTENT" in df.columns and "AARETE_DERIVED_SERVICE_TERM" in df.columns:
|
||||
_po_mask = (
|
||||
df["AMENDMENT_INTENT"].isin(["PARTIAL_REPLACEMENT", "OVERLAY_UPDATE"])
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].notna()
|
||||
)
|
||||
df.loc[_po_mask, "_exhibit_key"] = df.loc[
|
||||
_po_mask, "AARETE_DERIVED_SERVICE_TERM"
|
||||
]
|
||||
|
||||
# UNCLEAR_REVIEW rows are left for manual review — exclude from rank assignment
|
||||
if "AMENDMENT_INTENT" in df.columns:
|
||||
_active_rate_eligible = eligible_mask & (
|
||||
df["AMENDMENT_INTENT"] != "UNCLEAR_REVIEW"
|
||||
)
|
||||
else:
|
||||
_active_rate_eligible = eligible_mask
|
||||
|
||||
# Assign Y to the highest-ranked row per (FAMILY_TAG, exhibit key), N to the rest
|
||||
for (family, exhibit), grp in df[_active_rate_eligible].groupby(
|
||||
["FAMILY_TAG", "_exhibit_key"], sort=False, dropna=False
|
||||
):
|
||||
if pd.isna(exhibit):
|
||||
continue
|
||||
max_rank = max(grp["_eff_rank"])
|
||||
is_latest = grp["_eff_rank"] == max_rank
|
||||
df.loc[grp.index[is_latest], "ACTIVE_RATE"] = "Y"
|
||||
df.loc[grp.index[~is_latest], "ACTIVE_RATE"] = "N"
|
||||
|
||||
# Additive override: an ADDITIVE amendment with no later FULL_REPLACEMENT stays active.
|
||||
# Cross-family: also check FULL_REPLACEMENT rows in sibling families within the same
|
||||
# GROUP. If the ADDITIVE is the highest-ranked row in the group, no FULL_REPLACEMENT
|
||||
# will exceed it and it stays Y — preserving the "youngest member is ADDITIVE →
|
||||
# older member not suppressed" invariant.
|
||||
if "AMENDMENT_INTENT" in df.columns:
|
||||
# Pre-build all FULL_REPLACEMENT ranks per (GROUP, _exhibit_key) for cross-family lookup.
|
||||
_fr_ranks_by_group_exhibit: dict[tuple, set] = {}
|
||||
for (grp_name, exhibit), sub in df[_active_rate_eligible].groupby(
|
||||
["GROUP", "_exhibit_key"], sort=False, dropna=False
|
||||
):
|
||||
if pd.isna(exhibit):
|
||||
continue
|
||||
ranks = {
|
||||
r
|
||||
for r in sub.loc[
|
||||
sub["AMENDMENT_INTENT"] == "FULL_REPLACEMENT", "_eff_rank"
|
||||
]
|
||||
if r is not None
|
||||
}
|
||||
if ranks:
|
||||
_fr_ranks_by_group_exhibit[(grp_name, exhibit)] = ranks
|
||||
|
||||
for (family, exhibit), grp in df[_active_rate_eligible].groupby(
|
||||
["FAMILY_TAG", "_exhibit_key"], sort=False, dropna=False
|
||||
):
|
||||
if pd.isna(exhibit):
|
||||
continue
|
||||
add_idx = grp.index[grp["AMENDMENT_INTENT"] == "ADDITIVE"]
|
||||
if add_idx.empty:
|
||||
continue
|
||||
replace_ranks = {
|
||||
r
|
||||
for r in grp.loc[
|
||||
grp["AMENDMENT_INTENT"] == "FULL_REPLACEMENT", "_eff_rank"
|
||||
]
|
||||
if r is not None
|
||||
}
|
||||
# Widen to include FULL_REPLACEMENT rows in sibling families.
|
||||
grp_name = df.loc[grp.index[0], "GROUP"]
|
||||
replace_ranks |= _fr_ranks_by_group_exhibit.get((grp_name, exhibit), set())
|
||||
for idx in add_idx:
|
||||
add_rank = df.loc[idx, "_eff_rank"]
|
||||
if add_rank is None:
|
||||
continue
|
||||
if not any(r > add_rank for r in replace_ranks):
|
||||
df.loc[idx, "ACTIVE_RATE"] = "Y"
|
||||
|
||||
# Orphan active rate: highest QC_RANK within (GROUP, exhibit key) wins
|
||||
# UNCLEAR_REVIEW orphans are excluded from rank assignment
|
||||
_orphan_unclear_excluded = (
|
||||
(df["AMENDMENT_INTENT"] != "UNCLEAR_REVIEW")
|
||||
if "AMENDMENT_INTENT" in df.columns
|
||||
else pd.Series(True, index=df.index)
|
||||
)
|
||||
_orphan_exhibit_mask = (
|
||||
(df["MEMBER_TYPE"] == "orphan")
|
||||
& df["_exhibit_key"].notna()
|
||||
& _orphan_unclear_excluded
|
||||
)
|
||||
for (group, exhibit), grp in df[_orphan_exhibit_mask].groupby(
|
||||
["GROUP", "_exhibit_key"], sort=False, dropna=False
|
||||
):
|
||||
if pd.isna(exhibit):
|
||||
continue
|
||||
valid_ranks = [r for r in grp["_rank_tuple"] if r is not None]
|
||||
if not valid_ranks:
|
||||
continue
|
||||
max_rank = max(valid_ranks)
|
||||
is_latest = grp["_rank_tuple"].apply(lambda r: r == max_rank)
|
||||
df.loc[grp.index[is_latest], "ACTIVE_RATE"] = "Y"
|
||||
df.loc[grp.index[~is_latest], "ACTIVE_RATE"] = "N"
|
||||
|
||||
# Propagate within (GROUP, FILE_NAME, exhibit key): if any row is Y, all are Y
|
||||
_all_flagged_mask = (_active_rate_eligible | _orphan_exhibit_mask) & df[
|
||||
"_exhibit_key"
|
||||
].notna()
|
||||
for (group, file_name, exhibit), grp in df[_all_flagged_mask].groupby(
|
||||
["GROUP", "FILE_NAME", "_exhibit_key"], sort=False
|
||||
):
|
||||
group_flag = "Y" if (grp["ACTIVE_RATE"] == "Y").any() else "N"
|
||||
df.loc[grp.index, "ACTIVE_RATE"] = group_flag
|
||||
|
||||
# ── PARTIAL_REPLACEMENT: parent gets N for service terms covered by children ─
|
||||
# When children partially replace the parent, the parent's rates for those
|
||||
# specific service terms are no longer active. This runs after propagation so
|
||||
# it cannot be flipped back to Y.
|
||||
if "AMENDMENT_INTENT" in df.columns and "AARETE_DERIVED_SERVICE_TERM" in df.columns:
|
||||
_pr_child_terms = (
|
||||
df[
|
||||
(df["MEMBER_TYPE"] == "child")
|
||||
& (df["AMENDMENT_INTENT"] == "PARTIAL_REPLACEMENT")
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].notna()
|
||||
]
|
||||
.groupby("FAMILY_TAG")["AARETE_DERIVED_SERVICE_TERM"]
|
||||
.apply(set)
|
||||
)
|
||||
for family_tag, service_terms in _pr_child_terms.items():
|
||||
_parent_covered = (
|
||||
(df["MEMBER_TYPE"] == "parent")
|
||||
& (df["FAMILY_TAG"] == family_tag)
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].isin(service_terms)
|
||||
)
|
||||
df.loc[_parent_covered, "ACTIVE_RATE"] = "N"
|
||||
|
||||
# Cross-family: same PARTIAL_REPLACEMENT child service terms also suppress
|
||||
# parents in sibling families within the same GROUP.
|
||||
_pr_terms_by_group = (
|
||||
df[
|
||||
(df["MEMBER_TYPE"] == "child")
|
||||
& (df["AMENDMENT_INTENT"] == "PARTIAL_REPLACEMENT")
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].notna()
|
||||
]
|
||||
.groupby("GROUP")["AARETE_DERIVED_SERVICE_TERM"]
|
||||
.apply(set)
|
||||
)
|
||||
for group, service_terms in _pr_terms_by_group.items():
|
||||
_cross_parents = (
|
||||
(df["MEMBER_TYPE"] == "parent")
|
||||
& (df["GROUP"] == group)
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].isin(service_terms)
|
||||
)
|
||||
df.loc[_cross_parents, "ACTIVE_RATE"] = "N"
|
||||
|
||||
# ── FULL_REPLACEMENT: suppress parents in sibling families for same exhibit title ─
|
||||
# A FULL_REPLACEMENT child in family B supersedes the same exhibit title on parents
|
||||
# in ALL sibling families within the GROUP, not just its own family.
|
||||
# ADDITIVE children never trigger this block, so parents in ADDITIVE-only families
|
||||
# are unaffected.
|
||||
if "AMENDMENT_INTENT" in df.columns:
|
||||
_fr_exhibits_by_group = (
|
||||
df[
|
||||
(df["MEMBER_TYPE"] == "child")
|
||||
& (df["AMENDMENT_INTENT"] == "FULL_REPLACEMENT")
|
||||
& df["_exhibit_key"].notna()
|
||||
]
|
||||
.groupby("GROUP")["_exhibit_key"]
|
||||
.apply(set)
|
||||
)
|
||||
for group, exhibits in _fr_exhibits_by_group.items():
|
||||
_fr_cross_parents = (
|
||||
(df["MEMBER_TYPE"] == "parent")
|
||||
& (df["GROUP"] == group)
|
||||
& df["_exhibit_key"].isin(exhibits)
|
||||
)
|
||||
df.loc[_fr_cross_parents, "ACTIVE_RATE"] = "N"
|
||||
|
||||
# ── Cross-family parent deduplication ─────────────────────────────────────
|
||||
# Within a GROUP, if multiple parent families share a service term, only the
|
||||
# highest-ranked parent stays active; lower-ranked parents are suppressed.
|
||||
if "AARETE_DERIVED_SERVICE_TERM" in df.columns:
|
||||
_multi_parent_mask = (
|
||||
(df["MEMBER_TYPE"] == "parent")
|
||||
& df["AARETE_DERIVED_SERVICE_TERM"].notna()
|
||||
& df["ACTIVE_RATE"].notna()
|
||||
)
|
||||
for (group, svc_term), grp in df[_multi_parent_mask].groupby(
|
||||
["GROUP", "AARETE_DERIVED_SERVICE_TERM"], sort=False, dropna=False
|
||||
):
|
||||
if pd.isna(svc_term) or grp["FAMILY_TAG"].nunique() <= 1:
|
||||
continue
|
||||
valid_ranks = [r for r in grp["_eff_rank"] if r is not None]
|
||||
if not valid_ranks:
|
||||
continue
|
||||
max_rank = max(valid_ranks)
|
||||
is_not_max = grp["_eff_rank"].apply(
|
||||
lambda r: r is not None and r < max_rank
|
||||
)
|
||||
df.loc[grp.index[is_not_max], "ACTIVE_RATE"] = "N"
|
||||
|
||||
# Drop internal temp columns
|
||||
df = df.drop(
|
||||
columns=[
|
||||
"_rank_clean",
|
||||
"_rank_tuple",
|
||||
"_eff_rank",
|
||||
"_parent_pfx",
|
||||
"_exhibit_key",
|
||||
]
|
||||
)
|
||||
|
||||
logging.info(
|
||||
"Active rate flags computed. "
|
||||
f"MEMBER_TYPE breakdown: {df['MEMBER_TYPE'].value_counts(dropna=False).to_dict()} | "
|
||||
f"ACTIVE_RATE breakdown: {df['ACTIVE_RATE'].value_counts(dropna=False).to_dict()}"
|
||||
)
|
||||
return df
|
||||
|
||||
|
||||
# ── Output writer ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def write_active_rates_output(
|
||||
df: pd.DataFrame, run_timestamp: str, write_to_s3: bool = False
|
||||
) -> str:
|
||||
"""Write active rates Excel: Sheet1 = full data, 'focused cols' = key columns.
|
||||
|
||||
Returns the path of the written file.
|
||||
"""
|
||||
output_dir = os.path.join(
|
||||
config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "active-rates"
|
||||
)
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
output_path = os.path.join(output_dir, f"{config.BATCH_ID}-ACTIVE-RATES.xlsx")
|
||||
|
||||
focused_cols_present = [c for c in _FOCUSED_COLS if c in df.columns]
|
||||
|
||||
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
|
||||
df.to_excel(writer, sheet_name="Sheet1", index=False)
|
||||
df[focused_cols_present].to_excel(
|
||||
writer, sheet_name="focused cols", index=False
|
||||
)
|
||||
|
||||
logging.info(f"Active rates output written to: {output_path}")
|
||||
|
||||
if write_to_s3:
|
||||
from src.utils import io_utils
|
||||
|
||||
io_utils.upload_local_file_to_s3(output_path, run_timestamp, "active-rates")
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
# ── Public orchestration entry point ─────────────────────────────────────────
|
||||
|
||||
|
||||
def run_active_rates_postprocessing(
|
||||
main_df: pd.DataFrame, pc_df: pd.DataFrame
|
||||
) -> pd.DataFrame:
|
||||
"""Run the full active-rates postprocessing chain on main_df.
|
||||
|
||||
Steps:
|
||||
1. Merge GROUP / QC_RANK / QC_FLAG from pc_df
|
||||
2a. Standardize exhibit titles → AARETE_DERIVED_EXHIBIT_TITLE
|
||||
2b. Derive amendment intent → AMENDMENT_INTENT
|
||||
3. Compute active rate flags → MEMBER_TYPE / FAMILY_TAG / ACTIVE_RATE
|
||||
|
||||
Returns the enriched DataFrame (main_df is not mutated).
|
||||
"""
|
||||
logging.info("=" * 80)
|
||||
logging.info("Active Rates postprocessing: starting...")
|
||||
logging.info("=" * 80)
|
||||
|
||||
main_df = merge_parent_child_columns(main_df, pc_df)
|
||||
main_df = _standardize_and_derive_intent(main_df)
|
||||
main_df = add_active_rate_flags(main_df)
|
||||
|
||||
logging.info("=" * 80)
|
||||
logging.info("Active Rates postprocessing: complete.")
|
||||
logging.info("=" * 80)
|
||||
return main_df
|
||||
@@ -0,0 +1,446 @@
|
||||
"""
|
||||
Exhibit Title Standardization
|
||||
|
||||
After all files have been processed and combined, this module standardizes the
|
||||
exhibit_title field by mapping each unique value to a canonical title via LLM.
|
||||
The result is stored in a new AARETE_DERIVED_EXHIBIT_TITLE column inserted immediately
|
||||
after exhibit_title.
|
||||
|
||||
Two-pass design:
|
||||
Pass 1 — Taxonomy derivation
|
||||
All unique terms are sent to the LLM (in chunks when needed) with the sole
|
||||
purpose of producing a stable canonical title list. No mapping happens here.
|
||||
If chunks are used, each chunk contributes candidate titles which are then
|
||||
consolidated into one final taxonomy via a short deduplication call.
|
||||
|
||||
Pass 2 — Mapping
|
||||
Every unique term is mapped to exactly one canonical title from the taxonomy
|
||||
derived in Pass 1. Mapping is batched so very large vocabularies stay within
|
||||
token limits.
|
||||
|
||||
This guarantees the canonical title vocabulary is determined by the full dataset —
|
||||
not just the first N terms — before any row is mapped.
|
||||
|
||||
Grouping mode (optional):
|
||||
When a Group column is present, the two-pass standardization runs independently
|
||||
for each group so that canonical titles are derived from the specific mix of each
|
||||
group. Rows with no Group value are standardized globally as a fallback.
|
||||
|
||||
On any LLM failure the column falls back to the original exhibit_title value so
|
||||
downstream processing is never blocked.
|
||||
|
||||
Standalone usage:
|
||||
python -m src.pipelines.shared.postprocessing.exhibit_title_standardization
|
||||
<input_path> [sheet_name]
|
||||
|
||||
input_path — CSV or Excel file containing exhibit_title and Group columns.
|
||||
sheet_name — optional: name of the sheet to read (default: first sheet).
|
||||
Ignored for CSV files.
|
||||
"""
|
||||
|
||||
import difflib
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
import src.prompts.prompt_templates as prompt_templates
|
||||
import src.utils.json_utils as json_utils
|
||||
import src.utils.llm_utils as llm_utils
|
||||
|
||||
_COL = "exhibit_title"
|
||||
_OUT_COL = "AARETE_DERIVED_EXHIBIT_TITLE"
|
||||
_GROUP_COL = "GROUP"
|
||||
|
||||
# ── Tuneable constants ────────────────────────────────────────────────────────
|
||||
_MODEL_ID = "haiku_latest"
|
||||
_MAX_TOKENS = 8192
|
||||
_TAXONOMY_CHUNK_SIZE = 400
|
||||
_CONSOLIDATION_CHUNK_SIZE = 250
|
||||
_MAPPING_BATCH_SIZE = 100
|
||||
|
||||
|
||||
# ── LLM helper ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _call_llm(prompt: str) -> str:
|
||||
try:
|
||||
return llm_utils.invoke_claude(
|
||||
prompt=prompt,
|
||||
model_id=_MODEL_ID,
|
||||
filename="exhibit_title_standardization",
|
||||
max_tokens=_MAX_TOKENS,
|
||||
instruction=prompt_templates.EXHIBIT_TITLE_STANDARDIZATION_INSTRUCTION(),
|
||||
usage_label="exhibit_title_standardization",
|
||||
)
|
||||
except Exception as exc:
|
||||
logging.error(
|
||||
f"exhibit_title standardization: LLM call failed — {exc}", exc_info=True
|
||||
)
|
||||
return ""
|
||||
|
||||
|
||||
# ── Pass 1 — Taxonomy derivation ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def _derive_taxonomy(unique_terms: list[str]) -> list[str]:
|
||||
chunks = [
|
||||
unique_terms[i : i + _TAXONOMY_CHUNK_SIZE]
|
||||
for i in range(0, len(unique_terms), _TAXONOMY_CHUNK_SIZE)
|
||||
]
|
||||
|
||||
all_candidates: list[str] = []
|
||||
for idx, chunk in enumerate(chunks):
|
||||
response = _call_llm(
|
||||
prompt_templates.EXHIBIT_TITLE_TAXONOMY_CHUNK(
|
||||
json.dumps(chunk, ensure_ascii=False)
|
||||
)
|
||||
)
|
||||
if not response:
|
||||
logging.warning(f"Pass 1: empty response for chunk {idx + 1}; skipping")
|
||||
continue
|
||||
try:
|
||||
candidates = [
|
||||
str(v).strip()
|
||||
for v in json_utils.parse_json_list(response)
|
||||
if str(v).strip()
|
||||
]
|
||||
except (ValueError, TypeError):
|
||||
logging.warning(f"Pass 1: failed to parse list response (chunk {idx + 1})")
|
||||
candidates = []
|
||||
all_candidates.extend(candidates)
|
||||
|
||||
if not all_candidates:
|
||||
logging.warning(
|
||||
"Pass 1: no canonical titles derived; falling back to original terms"
|
||||
)
|
||||
return []
|
||||
|
||||
if len(chunks) == 1:
|
||||
return sorted(set(all_candidates))
|
||||
|
||||
deduped = sorted(set(all_candidates))
|
||||
consolidated: list[str] = []
|
||||
con_chunks = [
|
||||
deduped[i : i + _CONSOLIDATION_CHUNK_SIZE]
|
||||
for i in range(0, len(deduped), _CONSOLIDATION_CHUNK_SIZE)
|
||||
]
|
||||
for idx, con_chunk in enumerate(con_chunks):
|
||||
response = _call_llm(
|
||||
prompt_templates.EXHIBIT_TITLE_CONSOLIDATION(
|
||||
json.dumps(con_chunk, ensure_ascii=False)
|
||||
)
|
||||
)
|
||||
if response:
|
||||
try:
|
||||
consolidated.extend(
|
||||
str(v).strip()
|
||||
for v in json_utils.parse_json_list(response)
|
||||
if str(v).strip()
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
logging.warning(
|
||||
f"Pass 1: failed to parse consolidation response (chunk {idx + 1})"
|
||||
)
|
||||
consolidated.extend(con_chunk)
|
||||
else:
|
||||
consolidated.extend(con_chunk)
|
||||
|
||||
if len(con_chunks) > 1 and consolidated:
|
||||
response = _call_llm(
|
||||
prompt_templates.EXHIBIT_TITLE_CONSOLIDATION(
|
||||
json.dumps(sorted(set(consolidated)), ensure_ascii=False)
|
||||
)
|
||||
)
|
||||
if response:
|
||||
try:
|
||||
merged = [
|
||||
str(v).strip()
|
||||
for v in json_utils.parse_json_list(response)
|
||||
if str(v).strip()
|
||||
]
|
||||
if merged:
|
||||
consolidated = merged
|
||||
except (ValueError, TypeError):
|
||||
logging.warning("Pass 1: failed to parse final merge response")
|
||||
|
||||
return sorted(set(consolidated)) if consolidated else sorted(set(all_candidates))
|
||||
|
||||
|
||||
# ── Pass 2 — Mapping ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _map_terms_to_taxonomy(
|
||||
unique_terms: list[str], taxonomy: list[str]
|
||||
) -> dict[str, str]:
|
||||
mapping: dict[str, str] = {}
|
||||
categories_json = json.dumps(taxonomy, ensure_ascii=False)
|
||||
batches = [
|
||||
unique_terms[i : i + _MAPPING_BATCH_SIZE]
|
||||
for i in range(0, len(unique_terms), _MAPPING_BATCH_SIZE)
|
||||
]
|
||||
|
||||
for idx, batch in enumerate(batches):
|
||||
response = _call_llm(
|
||||
prompt_templates.EXHIBIT_TITLE_MAPPING_BATCH(
|
||||
categories_json, json.dumps(batch, ensure_ascii=False)
|
||||
)
|
||||
)
|
||||
if response:
|
||||
try:
|
||||
batch_mapping = {
|
||||
str(k): str(v)
|
||||
for k, v in json_utils.parse_json_dict(response).items()
|
||||
}
|
||||
except (ValueError, TypeError):
|
||||
logging.warning(
|
||||
f"Pass 2: failed to parse mapping response (batch {idx + 1})"
|
||||
)
|
||||
batch_mapping = {}
|
||||
else:
|
||||
batch_mapping = {}
|
||||
|
||||
for term in batch:
|
||||
mapping[term] = batch_mapping.get(term, term)
|
||||
|
||||
unmapped = [t for t in batch if t not in batch_mapping]
|
||||
if unmapped:
|
||||
logging.warning(
|
||||
f"Pass 2: batch {idx + 1} — {len(unmapped)} term(s) not mapped by LLM, "
|
||||
"falling back to original"
|
||||
)
|
||||
|
||||
return mapping
|
||||
|
||||
|
||||
# ── Public API ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_standardization_mapping(unique_terms: list[str]) -> dict[str, str]:
|
||||
if not unique_terms:
|
||||
return {}
|
||||
taxonomy = _derive_taxonomy(unique_terms)
|
||||
if not taxonomy:
|
||||
return {t: t for t in unique_terms}
|
||||
return _map_terms_to_taxonomy(unique_terms, taxonomy)
|
||||
|
||||
|
||||
def standardize_exhibit_titles(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Add AARETE_DERIVED_EXHIBIT_TITLE column immediately after exhibit_title.
|
||||
|
||||
When a Group column is present, standardization runs independently per group
|
||||
so canonical titles are derived from the specific mix of each group.
|
||||
Rows with no Group value are standardized globally as a fallback.
|
||||
|
||||
The source column is resolved case-insensitively, so both 'exhibit_title'
|
||||
(standalone script) and 'EXHIBIT_TITLE' (main pipeline) are accepted.
|
||||
|
||||
Returns df unchanged if no exhibit_title column is found.
|
||||
"""
|
||||
col_lower_map = {c.lower(): c for c in df.columns}
|
||||
actual_col = col_lower_map.get(_COL.lower())
|
||||
if actual_col is None:
|
||||
logging.warning(
|
||||
f"exhibit_title standardization: '{_COL}' column not found; skipping."
|
||||
)
|
||||
return df
|
||||
|
||||
renamed = actual_col != _COL
|
||||
if renamed:
|
||||
df = df.rename(columns={actual_col: _COL})
|
||||
|
||||
if _OUT_COL in df.columns:
|
||||
df = df.drop(columns=[_OUT_COL])
|
||||
|
||||
has_group = _GROUP_COL in df.columns
|
||||
|
||||
if has_group:
|
||||
row_groups = df[_GROUP_COL].where(
|
||||
df[_GROUP_COL].notna() & (df[_GROUP_COL].astype(str).str.strip() != "")
|
||||
)
|
||||
|
||||
ungrouped_count = row_groups.isna().sum()
|
||||
if ungrouped_count > 0:
|
||||
logging.info(
|
||||
f"exhibit_title standardization: {ungrouped_count} rows have no Group value; "
|
||||
"using global fallback"
|
||||
)
|
||||
|
||||
combined_mapping: dict[str, str] = {}
|
||||
|
||||
unique_groups = row_groups.dropna().unique().tolist()
|
||||
for group_key in unique_groups:
|
||||
group_terms = [
|
||||
t
|
||||
for t in df.loc[row_groups == group_key, _COL]
|
||||
.dropna()
|
||||
.unique()
|
||||
.tolist()
|
||||
if isinstance(t, str) and t.strip()
|
||||
]
|
||||
if not group_terms:
|
||||
continue
|
||||
logging.info(
|
||||
f" Standardizing group '{group_key}' ({len(group_terms)} unique titles)..."
|
||||
)
|
||||
combined_mapping.update(build_standardization_mapping(group_terms))
|
||||
|
||||
ungrouped_terms = [
|
||||
t
|
||||
for t in df.loc[row_groups.isna(), _COL].dropna().unique().tolist()
|
||||
if isinstance(t, str) and t.strip() and t not in combined_mapping
|
||||
]
|
||||
if ungrouped_terms:
|
||||
logging.info(
|
||||
f" Standardizing ungrouped fallback ({len(ungrouped_terms)} unique titles)..."
|
||||
)
|
||||
combined_mapping.update(build_standardization_mapping(ungrouped_terms))
|
||||
|
||||
mapping = combined_mapping
|
||||
else:
|
||||
logging.info("Global mode: no Group column found")
|
||||
unique_terms = [
|
||||
t
|
||||
for t in df[_COL].dropna().unique().tolist()
|
||||
if isinstance(t, str) and t.strip()
|
||||
]
|
||||
mapping = build_standardization_mapping(unique_terms)
|
||||
|
||||
df[_OUT_COL] = df[_COL].apply(
|
||||
lambda v: (
|
||||
mapping.get(str(v), str(v))
|
||||
if pd.notna(v) and str(v).strip()
|
||||
else (v if pd.isna(v) else "")
|
||||
)
|
||||
)
|
||||
|
||||
cols = df.columns.tolist()
|
||||
col_idx = cols.index(_COL)
|
||||
cols.remove(_OUT_COL)
|
||||
cols.insert(col_idx + 1, _OUT_COL)
|
||||
df = df[cols]
|
||||
|
||||
if renamed:
|
||||
df = df.rename(columns={_COL: actual_col})
|
||||
|
||||
return df
|
||||
|
||||
|
||||
# ── CLI ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def main():
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)-8s %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
)
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
logging.error(__doc__)
|
||||
sys.exit(1)
|
||||
|
||||
input_path = sys.argv[1]
|
||||
sheet_name = sys.argv[2] if len(sys.argv) > 2 else 0
|
||||
|
||||
p = Path(input_path)
|
||||
if not p.exists():
|
||||
logging.error(f"File not found: {input_path}")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
if p.suffix.lower() in {".xlsx", ".xls"}:
|
||||
df = pd.read_excel(input_path, sheet_name=sheet_name)
|
||||
logging.info(
|
||||
f"Loaded sheet '{sheet_name}' from {input_path}: {len(df):,} rows"
|
||||
)
|
||||
else:
|
||||
df = pd.read_csv(input_path)
|
||||
logging.info(f"Loaded {len(df):,} rows from {input_path}")
|
||||
except Exception as exc:
|
||||
logging.error(f"Could not load file — {exc}")
|
||||
sys.exit(1)
|
||||
|
||||
if _COL not in df.columns:
|
||||
logging.error(f"'{_COL}' column not found in the provided file.")
|
||||
close = difflib.get_close_matches(_COL, df.columns.tolist(), n=3, cutoff=0.4)
|
||||
if close:
|
||||
logging.error(f"Did you mean one of: {close}")
|
||||
else:
|
||||
logging.error(f"Columns present: {df.columns.tolist()}")
|
||||
sys.exit(1)
|
||||
|
||||
has_group = _GROUP_COL in df.columns
|
||||
unique_before = df[_COL].dropna().nunique()
|
||||
logging.info(
|
||||
f"Unique exhibit_title values before standardization: {unique_before:,}"
|
||||
)
|
||||
if has_group:
|
||||
logging.info(
|
||||
f"Grouping mode: {_GROUP_COL} column found ({df[_GROUP_COL].dropna().nunique()} unique groups)"
|
||||
)
|
||||
else:
|
||||
logging.info("Global mode: no Group column found")
|
||||
|
||||
logging.info("Running standardize_exhibit_titles...")
|
||||
df_out = standardize_exhibit_titles(df)
|
||||
|
||||
if _OUT_COL not in df_out.columns:
|
||||
logging.error(f"'{_OUT_COL}' column was not added — check logs above.")
|
||||
sys.exit(1)
|
||||
|
||||
mapping_df = (
|
||||
df_out[[_COL, _OUT_COL]]
|
||||
.drop_duplicates()
|
||||
.sort_values(_COL)
|
||||
.reset_index(drop=True)
|
||||
)
|
||||
|
||||
changed = mapping_df[
|
||||
mapping_df[_COL].astype(str) != mapping_df[_OUT_COL].astype(str)
|
||||
]
|
||||
unchanged = mapping_df[
|
||||
mapping_df[_COL].astype(str) == mapping_df[_OUT_COL].astype(str)
|
||||
]
|
||||
|
||||
logging.info("=" * 70)
|
||||
logging.info("STANDARDIZATION SUMMARY")
|
||||
logging.info("=" * 70)
|
||||
logging.info(f" Total unique mappings : {len(mapping_df):,}")
|
||||
logging.info(f" Changed : {len(changed):,}")
|
||||
logging.info(f" Unchanged (pass-thru) : {len(unchanged):,}")
|
||||
|
||||
if not changed.empty:
|
||||
logging.info("CHANGED MAPPINGS (first 50):")
|
||||
for _, row in changed.head(50).iterrows():
|
||||
raw = str(row[_COL])[:53]
|
||||
canonical = str(row[_OUT_COL])
|
||||
logging.info(f" {raw:<55} → {canonical}")
|
||||
|
||||
if not unchanged.empty:
|
||||
logging.info("UNCHANGED TITLES (first 20):")
|
||||
for _, row in unchanged.head(20).iterrows():
|
||||
logging.info(f" {row[_COL]!r}")
|
||||
|
||||
out_dir = p.parent
|
||||
stem = p.stem
|
||||
suffix = p.suffix.lower()
|
||||
out_ext = suffix if suffix in {".xlsx", ".xls"} else ".csv"
|
||||
|
||||
output_file = out_dir / f"{stem}_standardized_exhibits{out_ext}"
|
||||
mapping_file = out_dir / f"{stem}_standardized_exhibits_mapping.csv"
|
||||
|
||||
if out_ext in {".xlsx", ".xls"}:
|
||||
df_out.to_excel(output_file, index=False)
|
||||
else:
|
||||
df_out.to_csv(output_file, index=False)
|
||||
logging.info(f"Full output written to : {output_file}")
|
||||
|
||||
mapping_df.to_csv(mapping_file, index=False)
|
||||
logging.info(f"Mapping table written to : {mapping_file}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -127,6 +127,9 @@ def standard_postprocess(df, constants: Constants):
|
||||
# Derive int from contract text
|
||||
df = postprocessing_funcs.add_aarete_derived_amendment_num(df)
|
||||
|
||||
# Classify each unique exhibit's amendment intent (Add/Remove/Modify/Replace)
|
||||
df = postprocessing_funcs.add_amendment_intent(df)
|
||||
|
||||
# Process PATIENT_AGE_RANGE into PATIENT_AGE_MIN and PATIENT_AGE_MAX
|
||||
df = postprocessing_funcs.process_patient_age_range(df)
|
||||
|
||||
|
||||
@@ -733,6 +733,152 @@ def add_aarete_derived_amendment_num(df: pd.DataFrame) -> pd.DataFrame:
|
||||
return df
|
||||
|
||||
|
||||
_VALID_INTENT_TAGS = {
|
||||
"FULL_REPLACEMENT",
|
||||
"PARTIAL_REPLACEMENT",
|
||||
"ADDITIVE",
|
||||
"UNCLEAR_REVIEW",
|
||||
}
|
||||
|
||||
|
||||
def add_amendment_intent(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""
|
||||
Add AMENDMENT_INTENT column by classifying each unique AARETE_DERIVED_EXHIBIT_TITLE
|
||||
as FULL_REPLACEMENT, PARTIAL_REPLACEMENT, ADDITIVE, or UNCLEAR_REVIEW,
|
||||
using the AMENDMENT_INTENT_LANGUAGE field as context.
|
||||
|
||||
Called once per unique exhibit title — not once per row.
|
||||
"""
|
||||
if (
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE" not in df.columns
|
||||
or "AMENDMENT_INTENT_LANGUAGE" not in df.columns
|
||||
):
|
||||
return df
|
||||
|
||||
filename = (
|
||||
df["FILE_NAME"].iloc[0] if "FILE_NAME" in df.columns and not df.empty else ""
|
||||
)
|
||||
|
||||
has_qc_flag = "QC_FLAG" in df.columns
|
||||
has_file = "FILE_NAME" in df.columns
|
||||
|
||||
# intent_map key: (file_name, exhibit_title) — each file is classified independently
|
||||
# so a file with no amendment language is never assigned an intent derived from a
|
||||
# sibling file in the same group.
|
||||
intent_map: dict = {}
|
||||
reason_map: dict = {}
|
||||
|
||||
pair_cols = (
|
||||
["FILE_NAME", "AARETE_DERIVED_EXHIBIT_TITLE"]
|
||||
if has_file
|
||||
else ["AARETE_DERIVED_EXHIBIT_TITLE"]
|
||||
)
|
||||
pairs = (
|
||||
df[pair_cols].dropna(subset=["AARETE_DERIVED_EXHIBIT_TITLE"]).drop_duplicates()
|
||||
)
|
||||
|
||||
for row in pairs.itertuples(index=False, name=None):
|
||||
if has_file:
|
||||
file_name, exhibit_title = row
|
||||
title_mask = (df["FILE_NAME"] == file_name) & (
|
||||
df["AARETE_DERIVED_EXHIBIT_TITLE"] == exhibit_title
|
||||
)
|
||||
map_key = (file_name, exhibit_title)
|
||||
else:
|
||||
(exhibit_title,) = row
|
||||
file_name = None
|
||||
title_mask = df["AARETE_DERIVED_EXHIBIT_TITLE"] == exhibit_title
|
||||
map_key = exhibit_title
|
||||
|
||||
if string_utils.is_empty(str(exhibit_title)):
|
||||
continue
|
||||
|
||||
# Skip LLM for exhibit titles that belong exclusively to parent contracts.
|
||||
# Parent rows have amendment fields cleared downstream; running the LLM on them
|
||||
# wastes tokens and can produce misleading classifications.
|
||||
if has_qc_flag and (df.loc[title_mask, "QC_FLAG"] == "Parent").all():
|
||||
intent_map[map_key] = None
|
||||
reason_map[map_key] = "parent_only_exhibit"
|
||||
continue
|
||||
|
||||
candidate_languages = df.loc[title_mask, "AMENDMENT_INTENT_LANGUAGE"]
|
||||
candidate_languages = candidate_languages[
|
||||
~candidate_languages.astype(str).str.strip().isin({"", "N/A", "n/a", "nan"})
|
||||
& ~candidate_languages.isna()
|
||||
]
|
||||
amendment_language = (
|
||||
candidate_languages.iloc[0] if len(candidate_languages) > 0 else ""
|
||||
)
|
||||
|
||||
if string_utils.is_empty(str(amendment_language)) or str(
|
||||
amendment_language
|
||||
).strip() in {"N/A", "n/a", "nan"}:
|
||||
intent_map[map_key] = "N/A"
|
||||
reason_map[map_key] = "no_language"
|
||||
continue
|
||||
|
||||
try:
|
||||
prompt = prompt_templates.AMENDMENT_INTENT(
|
||||
str(exhibit_title), str(amendment_language)
|
||||
)
|
||||
raw = llm_utils.invoke_claude(
|
||||
prompt,
|
||||
"sonnet_latest",
|
||||
filename,
|
||||
max_tokens=150,
|
||||
usage_label="AMENDMENT_INTENT",
|
||||
)
|
||||
parsed = json_utils.parse_json_dict(raw)
|
||||
candidate = str(parsed.get("intent", "")).strip().upper()
|
||||
if candidate in _VALID_INTENT_TAGS:
|
||||
intent_map[map_key] = candidate
|
||||
reason_map[map_key] = "llm"
|
||||
else:
|
||||
logging.warning(
|
||||
f"AMENDMENT_INTENT returned unrecognized tag '{candidate}' for exhibit "
|
||||
f"'{exhibit_title}' (file={file_name!r}); raw={raw!r}"
|
||||
)
|
||||
intent_map[map_key] = "N/A"
|
||||
reason_map[map_key] = "llm_invalid_tag"
|
||||
except Exception as e:
|
||||
logging.warning(
|
||||
f"AMENDMENT_INTENT classification failed for exhibit '{exhibit_title}' "
|
||||
f"(file={file_name!r}): {e}"
|
||||
)
|
||||
intent_map[map_key] = "N/A"
|
||||
reason_map[map_key] = "llm_error"
|
||||
|
||||
if has_file:
|
||||
df["AMENDMENT_INTENT"] = pd.Series(
|
||||
[
|
||||
intent_map.get((f, et), "N/A") or "N/A"
|
||||
for f, et in zip(df["FILE_NAME"], df["AARETE_DERIVED_EXHIBIT_TITLE"])
|
||||
],
|
||||
index=df.index,
|
||||
)
|
||||
df["AMENDMENT_INTENT_REASON"] = pd.Series(
|
||||
[
|
||||
reason_map.get((f, et))
|
||||
for f, et in zip(df["FILE_NAME"], df["AARETE_DERIVED_EXHIBIT_TITLE"])
|
||||
],
|
||||
index=df.index,
|
||||
)
|
||||
else:
|
||||
df["AMENDMENT_INTENT"] = (
|
||||
df["AARETE_DERIVED_EXHIBIT_TITLE"].map(intent_map).fillna("N/A")
|
||||
)
|
||||
df["AMENDMENT_INTENT_REASON"] = df["AARETE_DERIVED_EXHIBIT_TITLE"].map(
|
||||
reason_map
|
||||
)
|
||||
|
||||
if has_qc_flag:
|
||||
df.loc[df["QC_FLAG"] == "Parent", "AMENDMENT_INTENT"] = None
|
||||
df.loc[df["QC_FLAG"] == "Parent", "AMENDMENT_INTENT_REASON"] = (
|
||||
"parent_row_cleared"
|
||||
)
|
||||
return df
|
||||
|
||||
|
||||
def add_aarete_derived_program_product(df):
|
||||
"""No-op pass-through: AARETE_DERIVED_PROGRAM/PRODUCT are now derived upstream
|
||||
in one_to_n_cleaning (one_to_n_funcs.add_aarete_derived_program_product) before
|
||||
|
||||
@@ -221,28 +221,37 @@ def update_provider_name(
|
||||
EMBEDDING_MODEL_ID=rag_config.EMBEDDING_MODEL_ID,
|
||||
BEDROCK_REGION=rag_config.BEDROCK_REGION,
|
||||
)
|
||||
hsc_provider_name = prompt_hsc_single_field(
|
||||
one_to_one_fields.get_field("AARETE_DERIVED_PROVIDER_NAME"),
|
||||
retriever,
|
||||
reranker,
|
||||
enhanced_config, # Use enhanced config instead of default
|
||||
text_dict,
|
||||
constants,
|
||||
filename,
|
||||
)[
|
||||
1
|
||||
] # Get the field value from the returned tuple
|
||||
if not string_utils.is_empty(hsc_provider_name):
|
||||
answers_dict["AARETE_DERIVED_PROVIDER_NAME"] = (
|
||||
hsc_provider_name[0]
|
||||
if isinstance(hsc_provider_name, list)
|
||||
else hsc_provider_name
|
||||
)
|
||||
logging.info(
|
||||
f"AARETE_DERIVED_PROVIDER_NAME extracted via enhanced RAG: {answers_dict['AARETE_DERIVED_PROVIDER_NAME']} | Filename: {filename}"
|
||||
)
|
||||
aarete_field = next(
|
||||
(
|
||||
f
|
||||
for f in one_to_one_fields.fields
|
||||
if f.field_name == "AARETE_DERIVED_PROVIDER_NAME"
|
||||
),
|
||||
None,
|
||||
)
|
||||
if aarete_field is not None:
|
||||
hsc_provider_name = prompt_hsc_single_field(
|
||||
aarete_field,
|
||||
retriever,
|
||||
reranker,
|
||||
enhanced_config, # Use enhanced config instead of default
|
||||
text_dict,
|
||||
constants,
|
||||
filename,
|
||||
)[
|
||||
1
|
||||
] # Get the field value from the returned tuple
|
||||
if not string_utils.is_empty(hsc_provider_name):
|
||||
answers_dict["AARETE_DERIVED_PROVIDER_NAME"] = (
|
||||
hsc_provider_name[0]
|
||||
if isinstance(hsc_provider_name, list)
|
||||
else hsc_provider_name
|
||||
)
|
||||
logging.info(
|
||||
f"AARETE_DERIVED_PROVIDER_NAME extracted via enhanced RAG: {answers_dict['AARETE_DERIVED_PROVIDER_NAME']} | Filename: {filename}"
|
||||
)
|
||||
|
||||
if string_utils.is_empty(answers_dict["PROVIDER_NAME"]):
|
||||
if string_utils.is_empty(answers_dict.get("PROVIDER_NAME")):
|
||||
answers_dict["PROVIDER_NAME"] = answers_dict.get(
|
||||
"AARETE_DERIVED_PROVIDER_NAME", "N/A"
|
||||
)
|
||||
|
||||
@@ -180,6 +180,13 @@
|
||||
"prompt": "Extract the amendment number or identifier if this document is an amendment to a contract. The amendment identifier can be numeric, letters only, or alphanumeric. Follow these guidelines:\n1. Look for phrases like 'Amendment No. X', 'X Amendment', 'Amendment X to the Agreement', 'Xth Amendment' in the document title, preamble, or header sections.\n2. Return the amendment number/identifier exactly as it appears, only if it consists of numbers, letters, or alphanumeric characters. (e.g., for 'Amendment No. 3' return '3', for 'Amendment A' return 'A', for 'Amendment 3A' return '3A').\n3. If the amendment number is written as a word (e.g., 'First Amendment'), convert it to a number (e.g., '1').\n4. If the amendment identifier is alphanumeric, return the whole value (e.g., for '2A' return '2A').\n5. If the amendment number uses Roman numerals (e.g., 'Amendment IV'), convert it to a number (e.g., '4').\n6. If the amendment number includes decimal points (e.g., 'Amendment 2.1'), return the full numeric value (e.g., '2.1').\n7. If the amendment identifier is letters only (e.g., 'Amendment AB'), return the letters exactly as they appear (e.g., 'AB',for 'Amendment XYZ' return 'XYZ').\n8. If multiple amendment numbers appear, return the one most clearly associated with the current document.\n9. If the document is an amendment but the number/identifier cannot be determined, return 'UNKNOWN'.\n10. If the document is not an amendment, return 'N/A'.\n11. If the amendment number contains any leading zeroes (e.g., '001'), drop all leading zeros (e.g., '1').",
|
||||
"retrieval_question": "What is the amendment number, letter, or alphanumeric identifier of this document? Amendment No., Amendment Number, Amendment Letter, Amendment Code, ordinal amendment, numbered amendment, lettered amendment, coded amendment? Amendment to agreement, amendment to contract."
|
||||
},
|
||||
{
|
||||
"field_name": "AMENDMENT_INTENT_LANGUAGE",
|
||||
"relationship": "one_to_one",
|
||||
"field_type": "smart_chunked",
|
||||
"prompt": "First, determine whether this document is an amendment, addendum, or modification to a contract. If it is NOT an amendment, return {\"AMENDMENT_INTENT_LANGUAGE\": \"N/A\"}. If it IS an amendment, extract ALL changes it makes to the original contract — do not omit any. Amendments typically enumerate their changes (e.g., \"1. Section 4.2 is amended to read...\", \"2. Exhibit A is replaced...\"). For each change, copy the FULL verbatim text exactly as it appears in the document — including all new or replacement contract language that the amendment introduces. Never truncate, summarize, paraphrase, or substitute a placeholder (such as '[full text of section]', '[new language]', or any bracketed description) in place of actual text. If new or replacement language is provided in the document, it must be included in full. Return {\"AMENDMENT_INTENT_LANGUAGE\": [\"1. <full verbatim text of change 1>\", \"2. <full verbatim text of change 2>\", ...]}. If the document is an amendment but lists no enumerated changes, return {\"AMENDMENT_INTENT_LANGUAGE\": [\"1. N/A\"]}.",
|
||||
"retrieval_question": "What changes does this amendment make to the original contract? Numbered or lettered list of modifications, amendments, revisions, replacements, deletions, additions to sections, exhibits, or provisions. Amendment intent, amendment terms, amendment changes, section is amended to read, exhibit is replaced, hereby modified, is deleted and replaced, is added."
|
||||
},
|
||||
{
|
||||
"field_name": "PROVIDER_STATE",
|
||||
"relationship": "one_to_one",
|
||||
|
||||
@@ -3021,6 +3021,91 @@ Briefly explain your reasoning, then put your final answer in JSON dictionary fo
|
||||
}}"""
|
||||
|
||||
|
||||
def AMENDMENT_INTENT_LANGUAGE(context: str) -> Tuple[str, Callable[[str], list]]:
|
||||
"""Returns prompt and parser for extracting numbered amendment changes.
|
||||
|
||||
Args:
|
||||
context: Contract text (typically first few pages where amendments list changes)
|
||||
|
||||
Returns:
|
||||
Tuple of (prompt_string, parser_function) where parser returns list of dicts.
|
||||
"""
|
||||
prompt = f"""The following is the text of a contract amendment.
|
||||
|
||||
## START AMENDMENT TEXT ##
|
||||
{context}
|
||||
## END AMENDMENT TEXT ##
|
||||
|
||||
{AMENDMENT_INTENT_LANGUAGE_INSTRUCTION()}"""
|
||||
|
||||
return (prompt, _json_list_parser)
|
||||
|
||||
|
||||
def AMENDMENT_INTENT_LANGUAGE_INSTRUCTION() -> str:
|
||||
return (
|
||||
"[OBJECTIVE]\n"
|
||||
"Extract the numbered or lettered list of changes this amendment makes to the original contract. "
|
||||
"Amendments typically enumerate their changes at the beginning of the document (e.g., '1. Section 4.2 is amended to read...', '2. Exhibit A is replaced...'). "
|
||||
"For each change, capture the change identifier and a concise description of what is being modified.\n\n"
|
||||
"[OUTPUT FORMAT]\n"
|
||||
"Briefly explain your findings, then return your final answer as a valid JSON list of dictionaries, "
|
||||
"one entry per numbered/lettered change:\n"
|
||||
'[\n {"change_number": "1", "description": "Section X is amended to replace ..."},\n'
|
||||
' {"change_number": "2", "description": "Exhibit A is deleted and replaced ..."}\n]\n'
|
||||
'If no numbered amendment changes are found (e.g. the document is not an amendment or lists no changes), return [{"change_number": "N/A", "description": "N/A"}].'
|
||||
)
|
||||
|
||||
|
||||
def AMENDMENT_INTENT_INSTRUCTION() -> str:
|
||||
return (
|
||||
"[OBJECTIVE]\n"
|
||||
"Given the amendment language for the specified exhibit title, classify the intent of the amendment "
|
||||
"as exactly one of: FULL_REPLACEMENT, PARTIAL_REPLACEMENT, ADDITIVE, UNCLEAR_REVIEW.\n\n"
|
||||
"[DEFINITIONS]\n"
|
||||
"- FULL_REPLACEMENT: The exhibit or contract section is deleted and replaced in its entirety. "
|
||||
"Triggered by phrases like 'deleted and replaced in its entirety,' 'superseded,' 'substituted therefore,' "
|
||||
"or similar language indicating a complete swap. "
|
||||
"This applies even when multiple exhibits are listed together under the same replacement language "
|
||||
"(e.g., 'Exhibit A and A-1 are deleted and replaced in their entirety' — both exhibits are FULL_REPLACEMENT).\n"
|
||||
"- PARTIAL_REPLACEMENT: Only a section, part, or portion of the exhibit is changed, or the change is "
|
||||
"limited to specific services or references. "
|
||||
"Triggered by phrases like 'section,' 'part,' 'portion,' 'with respect to,' limited service references, "
|
||||
"or similar language indicating a targeted modification.\n"
|
||||
"- ADDITIVE: New content is introduced without replacing existing content. "
|
||||
"Triggered by phrases like 'add,' 'addendum,' 'in addition to,' or similar language indicating a pure addition.\n"
|
||||
"- UNCLEAR_REVIEW: Use ONLY when FULL_REPLACEMENT, PARTIAL_REPLACEMENT, and ADDITIVE have each been explicitly "
|
||||
"ruled out — meaning the exhibit-specific language is so fundamentally contradictory or absent that none of the "
|
||||
"three can be supported even weakly. This is not a fallback for uncertainty; it is a classification of last resort "
|
||||
"that should almost never be needed. "
|
||||
"Do NOT use UNCLEAR_REVIEW for any of the following: "
|
||||
"(1) the overall amendment covers multiple exhibits — classify each on its own language; "
|
||||
"(2) language is informal or imprecise but a dominant intent is discernible; "
|
||||
"(3) the exhibit is not named explicitly but context makes intent reasonably inferable; "
|
||||
"(4) you are uncertain between two concrete options — commit to the best-supported one. "
|
||||
"Before choosing UNCLEAR_REVIEW, ask yourself: 'Can I justify FULL_REPLACEMENT, PARTIAL_REPLACEMENT, or ADDITIVE "
|
||||
"even partially?' If yes, choose that classification instead.\n\n"
|
||||
"[CLASSIFICATION APPROACH]\n"
|
||||
"Focus on the specific exhibit named in the 'Exhibit' field. Identify the language that directly describes what "
|
||||
"happens to that exhibit. Classify based on that targeted language alone, ignoring actions for other exhibits. "
|
||||
"Treat UNCLEAR_REVIEW as valid only after you have actively tried and failed to fit the language into each of the "
|
||||
"three concrete categories. Doubt, imprecision, or partial ambiguity are not grounds for UNCLEAR_REVIEW.\n\n"
|
||||
"[OUTPUT FORMAT]\n"
|
||||
'Respond with a JSON object with two fields: "reasoning" (1-2 sentences explaining your classification) '
|
||||
'and "intent" (exactly one of: FULL_REPLACEMENT, PARTIAL_REPLACEMENT, ADDITIVE, UNCLEAR_REVIEW).\n'
|
||||
"IMPORTANT: You MUST always return exactly one of FULL_REPLACEMENT, PARTIAL_REPLACEMENT, ADDITIVE, or UNCLEAR_REVIEW. "
|
||||
"No other values are valid.\n"
|
||||
'Example: {"reasoning": "The amendment states the exhibit is deleted and replaced in its entirety.", "intent": "FULL_REPLACEMENT"}'
|
||||
)
|
||||
|
||||
|
||||
def AMENDMENT_INTENT(exhibit_title: str, amendment_language: str) -> str:
|
||||
return (
|
||||
f"Amendment changes:\n{amendment_language}\n\n"
|
||||
f"Exhibit: {exhibit_title}\n\n"
|
||||
f"{AMENDMENT_INTENT_INSTRUCTION()}"
|
||||
)
|
||||
|
||||
|
||||
#####################################################################################
|
||||
############################### PREPROCESSING PROMPTS ###############################
|
||||
#####################################################################################
|
||||
@@ -4266,3 +4351,192 @@ def PRODUCT_TO_LOB(
|
||||
if payer_state and payer_state.strip():
|
||||
context += f"\nPayer State: {payer_state.strip()}"
|
||||
return (context, _json_list_parser)
|
||||
|
||||
|
||||
#####################################################################################
|
||||
########################### EXHIBIT TITLE STANDARDIZATION ###########################
|
||||
#####################################################################################
|
||||
|
||||
|
||||
def EXHIBIT_TITLE_STANDARDIZATION_INSTRUCTION() -> str:
|
||||
return (
|
||||
"You are a contract data analyst specializing in normalizing exhibit and schedule "
|
||||
"titles extracted from legal and business agreements. Your task is to standardize "
|
||||
"raw exhibit_title strings into clean, consistent canonical names so that downstream "
|
||||
"analysis can reliably group and compare exhibits across documents."
|
||||
)
|
||||
|
||||
|
||||
def EXHIBIT_TITLE_TAXONOMY_CHUNK(terms_json: str) -> str:
|
||||
"""Pass-1 prompt: derive a canonical taxonomy from a chunk of unique exhibit titles."""
|
||||
return f"""\
|
||||
Below is a sample of unique exhibit_title values from a contract document group.
|
||||
|
||||
Produce a concise list of CANONICAL standardized exhibit titles that cover every title in
|
||||
this list.
|
||||
|
||||
Rules:
|
||||
1. TITLE CASE — all output labels must be in Title Case. Fix ALL-CAPS raw titles.
|
||||
Example: "EXHIBIT A - PRICING SCHEDULE" → "Exhibit A - Pricing Schedule"
|
||||
|
||||
2. STRIP REDUNDANT PREFIXES — if all titles share a common prefix (e.g. "Exhibit",
|
||||
"Schedule", "Attachment"), keep it only when it distinguishes variants.
|
||||
Example: "Exhibit A - Services", "Exhibit B - Pricing" → keep as-is (letter distinguishes)
|
||||
Example: "Exhibit - Pricing", "EXHIBIT - PRICING SCHEDULE" → "Pricing Schedule" (prefix adds nothing)
|
||||
|
||||
3. NORMALIZE FILLER — remove words that add no meaning such as "And", "The", "For",
|
||||
"Regarding" at the start of a title core.
|
||||
Example: "Exhibit for the Services" → "Services"
|
||||
|
||||
4. MERGE NEAR-DUPLICATES — if two titles clearly refer to the same exhibit (same letter,
|
||||
same subject, minor casing or punctuation difference), output one canonical form.
|
||||
Example: "Exhibit A-Rates", "Exhibit A - Rate Schedule" → one canonical entry
|
||||
|
||||
5. PRESERVE MEANINGFUL VARIANTS — lettered or numbered variants are distinct.
|
||||
Example: "Exhibit A", "Exhibit B" are separate canonical names.
|
||||
Example: "Schedule 1", "Schedule 2" are separate canonical names.
|
||||
|
||||
6. PASS-THROUGH — if a raw title is already clean and well-formed, output it with only
|
||||
minimal adjustments (Title Case, strip trailing punctuation).
|
||||
|
||||
7. SECTION IDENTIFIER TITLES — if a raw title begins with a section keyword followed by
|
||||
an identifier, the canonical form depends on the keyword:
|
||||
|
||||
• EXHIBIT or SCHEDULE: output ONLY the bare identifier (letter, number, or compound
|
||||
code). Completely drop the "Exhibit" / "Schedule" keyword prefix AND any description,
|
||||
regardless of how the description is attached (newline, dash, colon, same line, etc.).
|
||||
Example: "EXHIBIT C\\nPROGRAMS TO WHICH THIS AGREEMENT APPLIES" → "C"
|
||||
Example: "EXHIBIT C - PROGRAMS TO WHICH THIS AGREEMENT APPLIES" → "C"
|
||||
Example: "EXHIBIT C-3\\nMEDICARE ADVANTAGE PPO AND HMO PROVISIONS" → "C-3"
|
||||
Example: "SCHEDULE C-2\\nMEDICARE ADVANTAGE PAYMENT RATES" → "C-2"
|
||||
Example: "SCHEDULE 1" → "1"
|
||||
|
||||
When a title contains multiple Exhibit/Schedule identifiers across lines (e.g. a
|
||||
top-level exhibit and a nested schedule/attachment), return the MOST SPECIFIC
|
||||
(lowest-level) Exhibit or Schedule identifier — i.e. prefer a compound code
|
||||
(letter-hyphen-number) over a plain letter, and a deeper reference over a shallower one.
|
||||
Example: "Exhibit C\\nAttachment 1 to Schedule C-3\\nEPO PAYMENT RATES" → "C-3"
|
||||
(C-3 is more specific than C; ignore non-Exhibit/Schedule keywords like Attachment)
|
||||
|
||||
• All other keywords (ATTACHMENT, ARTICLE, ADDENDUM, APPENDIX, SECTION, and any
|
||||
similar structural keyword): the canonical form is keyword + identifier in Title Case.
|
||||
Any text after the identifier — whether separated by a colon, newline, dash, or any
|
||||
other punctuation — must be completely stripped. Do NOT carry over any description
|
||||
words into the output.
|
||||
Strip qualifier words between the keyword and identifier such as "NO.", "NO",
|
||||
"NUMBER", "#", or similar — they add no meaning.
|
||||
Example: "ATTACHMENT B\\nCOMPENSATION" → "Attachment B"
|
||||
Example: "ARTICLE I\\nDefinitions" → "Article I"
|
||||
Example: "ADDENDUM 3:\\nADDITIONAL TERMS AND CONDITIONS" → "Addendum 3"
|
||||
Example: "AMENDMENT NO. 1\\nTO\\nHOSPITAL SERVICE AGREEMENT" → "Amendment 1"
|
||||
Example: "ADDENDUM NUMBER 2\\nPRICING TERMS" → "Addendum 2"
|
||||
|
||||
MULTI-LINE BOILERPLATE — Lines after the identifier that begin with prepositions
|
||||
such as "TO", "BETWEEN", "AND", "FOR", or "WITH" (typically introducing agreement
|
||||
names or party names) are boilerplate and must be stripped entirely.
|
||||
Example: "ADDENDUM I\\nTO FACILITY SERVICES AGREEMENT\\nBETWEEN PARTY A\\nAND PARTY B" → "Addendum I"
|
||||
Example: "ATTACHMENT B\\nTO THE MASTER AGREEMENT\\nBETWEEN PARTY A AND PARTY B" → "Attachment B"
|
||||
|
||||
COMPOUND STRUCTURAL HEADERS — If the title begins with a modifier phrase rather
|
||||
than a primary structural keyword (e.g. "AMENDMENT TO", "SUPPLEMENT TO",
|
||||
"RESTATEMENT OF"), identify the primary structural keyword and its identifier from
|
||||
the subsequent lines and use only those as the canonical form. Strip all lines
|
||||
that follow the identifier, including boilerplate preposition lines.
|
||||
Example: "AMENDMENT TO\\nADDENDUM I\\nTO PARTICIPATING NETWORK AGREEMENT\\nBETWEEN PARTY A" → "Addendum I"
|
||||
Example: "SUPPLEMENT TO\\nAPPENDIX 2\\nSERVICES DESCRIPTION" → "Appendix 2"
|
||||
|
||||
Return ONLY a valid JSON array of canonical exhibit title strings — no explanation, no markdown.
|
||||
|
||||
EXHIBIT_TITLES:
|
||||
{terms_json}"""
|
||||
|
||||
|
||||
def EXHIBIT_TITLE_CONSOLIDATION(categories_json: str) -> str:
|
||||
"""Pass-1 consolidation prompt: deduplicate candidate canonical titles across chunks."""
|
||||
return f"""\
|
||||
The following candidate canonical exhibit titles were proposed from different subsets of the
|
||||
same document group. Many are duplicates or near-duplicates.
|
||||
|
||||
Merge and deduplicate into one final, minimal set:
|
||||
- Keep the most descriptive but concise form (prefer "Pricing Schedule" over "Pricing" or
|
||||
"Full Pricing Schedule and Rate Table").
|
||||
- Keep lettered/numbered variants separate ("Exhibit A" is distinct from "Exhibit B").
|
||||
- All labels must be in Title Case.
|
||||
- Drop near-duplicates that refer to the same exhibit.
|
||||
|
||||
Return ONLY a valid JSON array of the final canonical title strings — no explanation, no markdown.
|
||||
|
||||
CANDIDATE TITLES:
|
||||
{categories_json}"""
|
||||
|
||||
|
||||
def EXHIBIT_TITLE_MAPPING_BATCH(categories_json: str, terms_json: str) -> str:
|
||||
"""Pass-2 prompt: map each raw exhibit title to its canonical form."""
|
||||
return f"""\
|
||||
Map each raw exhibit_title below to exactly one canonical title from the established list.
|
||||
Do NOT create new canonical titles.
|
||||
|
||||
Mapping rules:
|
||||
1. TITLE CASE — normalize ALL-CAPS raw titles before matching.
|
||||
2. IGNORE MINOR DIFFERENCES — punctuation, spacing, and filler word variations should map
|
||||
to the same canonical title.
|
||||
3. PRESERVE LETTER/NUMBER VARIANTS — "Exhibit A" must map to the "Exhibit A" canonical
|
||||
entry, not "Exhibit B".
|
||||
4. PASS-THROUGH — if a raw title is already in or very close to its canonical form, map it
|
||||
directly.
|
||||
5. CONSISTENCY — within this batch, if two raw titles represent the same exhibit, map them
|
||||
both to the same canonical title.
|
||||
|
||||
6. SECTION IDENTIFIER TITLES — if a raw title begins with a section keyword followed by
|
||||
an identifier, the canonical form depends on the keyword:
|
||||
|
||||
• EXHIBIT or SCHEDULE: map to ONLY the bare identifier (letter, number, or compound
|
||||
code). Completely drop the "Exhibit" / "Schedule" keyword prefix AND any description,
|
||||
regardless of how the description is attached (newline, dash, colon, same line, etc.).
|
||||
Example: "EXHIBIT C\\nPROGRAMS TO WHICH THIS AGREEMENT APPLIES" → "C"
|
||||
Example: "EXHIBIT C - PROGRAMS TO WHICH THIS AGREEMENT APPLIES" → "C"
|
||||
Example: "EXHIBIT C-3\\nMEDICARE ADVANTAGE PPO AND HMO PROVISIONS" → "C-3"
|
||||
Example: "SCHEDULE C-2\\nMEDICARE ADVANTAGE PAYMENT RATES" → "C-2"
|
||||
Example: "SCHEDULE 1" → "1"
|
||||
|
||||
When a title contains multiple Exhibit/Schedule identifiers across lines (e.g. a
|
||||
top-level exhibit and a nested schedule/attachment), map to the MOST SPECIFIC
|
||||
(lowest-level) Exhibit or Schedule identifier — i.e. prefer a compound code
|
||||
(letter-hyphen-number) over a plain letter, and a deeper reference over a shallower one.
|
||||
Example: "Exhibit C\\nAttachment 1 to Schedule C-3\\nEPO PAYMENT RATES" → "C-3"
|
||||
(C-3 is more specific than C; ignore non-Exhibit/Schedule keywords like Attachment)
|
||||
|
||||
• All other keywords (ATTACHMENT, ARTICLE, ADDENDUM, APPENDIX, SECTION, and any
|
||||
similar structural keyword): map to keyword + identifier in Title Case. Any text
|
||||
after the identifier — whether separated by a colon, newline, dash, or any other
|
||||
punctuation — must be completely stripped. Do NOT carry over any description words.
|
||||
Strip qualifier words between the keyword and identifier such as "NO.", "NO",
|
||||
"NUMBER", "#", or similar — they add no meaning.
|
||||
Example: "ATTACHMENT B\\nCOMPENSATION" → "Attachment B"
|
||||
Example: "ARTICLE I\\nDefinitions" → "Article I"
|
||||
Example: "ADDENDUM 3:\\nADDITIONAL TERMS AND CONDITIONS" → "Addendum 3"
|
||||
Example: "AMENDMENT NO. 1\\nTO\\nHOSPITAL SERVICE AGREEMENT" → "Amendment 1"
|
||||
Example: "ADDENDUM NUMBER 2\\nPRICING TERMS" → "Addendum 2"
|
||||
|
||||
MULTI-LINE BOILERPLATE — Lines after the identifier that begin with prepositions
|
||||
such as "TO", "BETWEEN", "AND", "FOR", or "WITH" (typically introducing agreement
|
||||
names or party names) are boilerplate and must be stripped entirely.
|
||||
Example: "ADDENDUM I\\nTO FACILITY SERVICES AGREEMENT\\nBETWEEN PARTY A\\nAND PARTY B" → "Addendum I"
|
||||
Example: "ATTACHMENT B\\nTO THE MASTER AGREEMENT\\nBETWEEN PARTY A AND PARTY B" → "Attachment B"
|
||||
|
||||
COMPOUND STRUCTURAL HEADERS — If the title begins with a modifier phrase rather
|
||||
than a primary structural keyword (e.g. "AMENDMENT TO", "SUPPLEMENT TO",
|
||||
"RESTATEMENT OF"), identify the primary structural keyword and its identifier from
|
||||
the subsequent lines and use only those as the canonical form. Strip all lines
|
||||
that follow the identifier, including boilerplate preposition lines.
|
||||
Example: "AMENDMENT TO\\nADDENDUM I\\nTO PARTICIPATING NETWORK AGREEMENT\\nBETWEEN PARTY A" → "Addendum I"
|
||||
Example: "SUPPLEMENT TO\\nAPPENDIX 2\\nSERVICES DESCRIPTION" → "Appendix 2"
|
||||
|
||||
ESTABLISHED CANONICAL TITLES:
|
||||
{categories_json}
|
||||
|
||||
EXHIBIT_TITLES:
|
||||
{terms_json}
|
||||
|
||||
Return ONLY a valid JSON object where each key is the original exhibit_title string
|
||||
(exactly as given) and each value is its canonical title — no explanation, no markdown."""
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.pipelines.shared.postprocessing.active_rates import add_active_rate_flags
|
||||
from src.pipelines.shared.postprocessing.postprocessing_funcs import (
|
||||
add_amendment_intent,
|
||||
)
|
||||
|
||||
|
||||
def _df(**cols) -> pd.DataFrame:
|
||||
"""Build a DataFrame with required active-rates columns plus any extras."""
|
||||
n = len(next(iter(cols.values())))
|
||||
base = {
|
||||
"FILE_NAME": [f"f{i}.pdf" for i in range(n)],
|
||||
"QC_FLAG": ["Child"] * n,
|
||||
"GROUP": ["G1"] * n,
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A"] * n,
|
||||
}
|
||||
base.update(cols)
|
||||
return pd.DataFrame(base)
|
||||
|
||||
|
||||
# ── add_active_rate_flags ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAddActiveRateFlagsMissingColumns(unittest.TestCase):
|
||||
def test_missing_required_columns_returns_unchanged(self):
|
||||
df = pd.DataFrame({"FILE_NAME": ["f1.pdf"], "QC_RANK": ["1"]})
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertNotIn("MEMBER_TYPE", result.columns)
|
||||
self.assertNotIn("ACTIVE_RATE", result.columns)
|
||||
|
||||
def test_no_temp_columns_in_output(self):
|
||||
df = _df(QC_RANK=["1"], QC_FLAG=["Child"])
|
||||
result = add_active_rate_flags(df)
|
||||
for col in [
|
||||
"_rank_clean",
|
||||
"_rank_tuple",
|
||||
"_eff_rank",
|
||||
"_parent_pfx",
|
||||
"_exhibit_key",
|
||||
]:
|
||||
self.assertNotIn(col, result.columns)
|
||||
|
||||
|
||||
class TestMemberTypeClassification(unittest.TestCase):
|
||||
def _run(self, ranks, flags=None):
|
||||
n = len(ranks)
|
||||
df = _df(
|
||||
QC_RANK=ranks,
|
||||
QC_FLAG=flags or ["Child"] * n,
|
||||
)
|
||||
return add_active_rate_flags(df)
|
||||
|
||||
def test_parent_rank(self):
|
||||
result = self._run(["1"])
|
||||
self.assertEqual(result["MEMBER_TYPE"].iloc[0], "parent")
|
||||
|
||||
def test_child_rank(self):
|
||||
result = self._run(["1.1"])
|
||||
self.assertEqual(result["MEMBER_TYPE"].iloc[0], "child")
|
||||
|
||||
def test_orphan_rank(self):
|
||||
result = self._run(["0.1"])
|
||||
self.assertEqual(result["MEMBER_TYPE"].iloc[0], "orphan")
|
||||
|
||||
def test_unknown_when_rank_is_null(self):
|
||||
result = self._run([None])
|
||||
self.assertEqual(result["MEMBER_TYPE"].iloc[0], "unknown")
|
||||
|
||||
def test_excel_float_rank_normalised(self):
|
||||
# "1.0" should be treated as parent rank "1"
|
||||
result = self._run(["1.0"])
|
||||
self.assertEqual(result["MEMBER_TYPE"].iloc[0], "parent")
|
||||
|
||||
|
||||
class TestFamilyTag(unittest.TestCase):
|
||||
def test_child_family_tag_uses_parent_prefix(self):
|
||||
df = _df(QC_RANK=["1.1"], GROUP=["G1"])
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertEqual(result["FAMILY_TAG"].iloc[0], "G1__1")
|
||||
|
||||
def test_parent_family_tag_with_child_present(self):
|
||||
# When a group has both parent and child, FAMILY_TAG = GROUP__prefix
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["1", "1.1"],
|
||||
GROUP=["G1", "G1"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertTrue((result["FAMILY_TAG"] == "G1__1").all())
|
||||
|
||||
def test_sole_parent_in_group_collapses_family_tag_to_group(self):
|
||||
# Single parent with no children → parent-only group → FAMILY_TAG = GROUP
|
||||
df = _df(QC_RANK=["1"], GROUP=["G1"])
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertEqual(result["FAMILY_TAG"].iloc[0], "G1")
|
||||
|
||||
def test_orphan_and_unknown_have_no_family_tag(self):
|
||||
df = _df(QC_RANK=["0.1", None], GROUP=["G1", "G1"])
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertTrue(pd.isna(result["FAMILY_TAG"].iloc[0]))
|
||||
self.assertTrue(pd.isna(result["FAMILY_TAG"].iloc[1]))
|
||||
|
||||
def test_parent_only_group_collapses_family_tag_to_group(self):
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["1", "2"],
|
||||
QC_FLAG=["Child", "Child"],
|
||||
GROUP=["G1", "G1"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
# All rows in G1 are parents → FAMILY_TAG should be "G1", not "G1__1"
|
||||
self.assertTrue((result["FAMILY_TAG"] == "G1").all())
|
||||
|
||||
|
||||
class TestActiveRateHighestRankWins(unittest.TestCase):
|
||||
def test_highest_rank_gets_Y(self):
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["1", "1.1"],
|
||||
QC_FLAG=["Child", "Child"],
|
||||
GROUP=["G1", "G1"],
|
||||
AARETE_DERIVED_EXHIBIT_TITLE=["Ex A", "Ex A"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
rank_1_row = result[result["QC_RANK"] == "1"].iloc[0]
|
||||
rank_11_row = result[result["QC_RANK"] == "1.1"].iloc[0]
|
||||
self.assertEqual(rank_11_row["ACTIVE_RATE"], "Y")
|
||||
self.assertEqual(rank_1_row["ACTIVE_RATE"], "N")
|
||||
|
||||
def test_different_exhibits_each_get_own_Y(self):
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["1", "2"],
|
||||
QC_FLAG=["Child", "Child"],
|
||||
GROUP=["G1", "G1"],
|
||||
AARETE_DERIVED_EXHIBIT_TITLE=["Ex A", "Ex B"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertTrue((result["ACTIVE_RATE"] == "Y").all())
|
||||
|
||||
|
||||
class TestAdditiveOverride(unittest.TestCase):
|
||||
def _make_three_rows(self, third_intent):
|
||||
return _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf", "f3.pdf"],
|
||||
QC_RANK=["1", "1.1", "1.2"],
|
||||
QC_FLAG=["Child", "Child", "Child"],
|
||||
GROUP=["G1", "G1", "G1"],
|
||||
AARETE_DERIVED_EXHIBIT_TITLE=["Ex A", "Ex A", "Ex A"],
|
||||
AMENDMENT_INTENT=[None, "ADDITIVE", third_intent],
|
||||
)
|
||||
|
||||
def test_additive_stays_Y_when_no_full_replacement(self):
|
||||
df = self._make_three_rows(None) # no FULL_REPLACEMENT in the group
|
||||
result = add_active_rate_flags(df)
|
||||
additive_row = result[result["QC_RANK"] == "1.1"].iloc[0]
|
||||
self.assertEqual(additive_row["ACTIVE_RATE"], "Y")
|
||||
|
||||
def test_additive_stays_N_when_later_full_replacement_exists(self):
|
||||
df = self._make_three_rows("FULL_REPLACEMENT")
|
||||
result = add_active_rate_flags(df)
|
||||
additive_row = result[result["QC_RANK"] == "1.1"].iloc[0]
|
||||
self.assertEqual(additive_row["ACTIVE_RATE"], "N")
|
||||
|
||||
|
||||
class TestOrphanActiveRate(unittest.TestCase):
|
||||
def test_highest_orphan_rank_gets_Y(self):
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["0.1", "0.2"],
|
||||
QC_FLAG=["Child", "Child"],
|
||||
GROUP=["G1", "G1"],
|
||||
AARETE_DERIVED_EXHIBIT_TITLE=["Ex A", "Ex A"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
orphan_low = result[result["QC_RANK"] == "0.1"].iloc[0]
|
||||
orphan_high = result[result["QC_RANK"] == "0.2"].iloc[0]
|
||||
self.assertEqual(orphan_high["ACTIVE_RATE"], "Y")
|
||||
self.assertEqual(orphan_low["ACTIVE_RATE"], "N")
|
||||
|
||||
|
||||
class TestActiveRatePropagation(unittest.TestCase):
|
||||
def test_all_rows_in_same_file_get_Y_if_any_is_Y(self):
|
||||
# Two rows in the same file, same exhibit — child is Y, parent would be N
|
||||
# but propagation should make parent Y too.
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["f1.pdf", "f1.pdf"],
|
||||
"QC_RANK": ["1", "1.1"],
|
||||
"QC_FLAG": ["Child", "Child"],
|
||||
"GROUP": ["G1", "G1"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A", "Ex A"],
|
||||
}
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertTrue((result["ACTIVE_RATE"] == "Y").all())
|
||||
|
||||
|
||||
class TestUnclearReviewExcluded(unittest.TestCase):
|
||||
def test_unclear_review_rows_have_no_active_rate(self):
|
||||
df = _df(
|
||||
FILE_NAME=["f1.pdf", "f2.pdf"],
|
||||
QC_RANK=["1", "1.1"],
|
||||
QC_FLAG=["Child", "Child"],
|
||||
GROUP=["G1", "G1"],
|
||||
AARETE_DERIVED_EXHIBIT_TITLE=["Ex A", "Ex A"],
|
||||
AMENDMENT_INTENT=["UNCLEAR_REVIEW", "UNCLEAR_REVIEW"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertTrue(result["ACTIVE_RATE"].isna().all())
|
||||
|
||||
|
||||
class TestPartialReplacementParent(unittest.TestCase):
|
||||
def test_parent_active_rate_set_to_N_for_covered_service_terms(self):
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["f_parent.pdf", "f_child.pdf"],
|
||||
"QC_RANK": ["1", "1.1"],
|
||||
"QC_FLAG": ["Child", "Child"],
|
||||
"GROUP": ["G1", "G1"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A", "Ex A"],
|
||||
"AARETE_DERIVED_SERVICE_TERM": ["Svc X", "Svc X"],
|
||||
"AMENDMENT_INTENT": [None, "PARTIAL_REPLACEMENT"],
|
||||
}
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
parent_row = result[result["FILE_NAME"] == "f_parent.pdf"].iloc[0]
|
||||
child_row = result[result["FILE_NAME"] == "f_child.pdf"].iloc[0]
|
||||
self.assertEqual(parent_row["ACTIVE_RATE"], "N")
|
||||
self.assertEqual(child_row["ACTIVE_RATE"], "Y")
|
||||
|
||||
|
||||
class TestAmendmentFieldClearedOnParents(unittest.TestCase):
|
||||
def test_parent_qc_flag_clears_amendment_intent(self):
|
||||
df = _df(
|
||||
QC_RANK=["1"],
|
||||
QC_FLAG=["Parent"],
|
||||
AMENDMENT_INTENT=["FULL_REPLACEMENT"],
|
||||
AMENDMENT_INTENT_LANGUAGE=["some language"],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertIsNone(result["AMENDMENT_INTENT"].iloc[0])
|
||||
self.assertIsNone(result["AMENDMENT_INTENT_LANGUAGE"].iloc[0])
|
||||
|
||||
|
||||
class TestOverlayUpdateFallback(unittest.TestCase):
|
||||
def test_files_with_no_language_get_overlay_update(self):
|
||||
df = _df(
|
||||
QC_RANK=["1.1"],
|
||||
QC_FLAG=["Child"],
|
||||
AMENDMENT_INTENT=["N/A"],
|
||||
AMENDMENT_INTENT_LANGUAGE=[None],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertEqual(result["AMENDMENT_INTENT"].iloc[0], "OVERLAY_UPDATE")
|
||||
|
||||
def test_files_with_language_keep_existing_intent(self):
|
||||
df = _df(
|
||||
QC_RANK=["1.1"],
|
||||
QC_FLAG=["Child"],
|
||||
AMENDMENT_INTENT=["FULL_REPLACEMENT"],
|
||||
AMENDMENT_INTENT_LANGUAGE=["This replaces all prior rates."],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
self.assertEqual(result["AMENDMENT_INTENT"].iloc[0], "FULL_REPLACEMENT")
|
||||
|
||||
|
||||
class TestAmendmentLanguageNormalisation(unittest.TestCase):
|
||||
def test_bracketed_na_string_becomes_nan(self):
|
||||
df = _df(
|
||||
QC_RANK=["1.1"],
|
||||
QC_FLAG=["Child"],
|
||||
AMENDMENT_INTENT=["N/A"],
|
||||
AMENDMENT_INTENT_LANGUAGE=['["N/A"]'],
|
||||
)
|
||||
result = add_active_rate_flags(df)
|
||||
# ["N/A"] is treated as no language → fallback OVERLAY_UPDATE applied
|
||||
self.assertEqual(result["AMENDMENT_INTENT"].iloc[0], "OVERLAY_UPDATE")
|
||||
|
||||
|
||||
# ── add_amendment_intent ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAddAmendmentIntent(unittest.TestCase):
|
||||
def test_returns_unchanged_when_missing_exhibit_title_column(self):
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"AMENDMENT_INTENT_LANGUAGE": ["some text"],
|
||||
"FILE_NAME": ["f1.pdf"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
self.assertNotIn("AMENDMENT_INTENT", result.columns)
|
||||
|
||||
def test_returns_unchanged_when_missing_language_column(self):
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A"],
|
||||
"FILE_NAME": ["f1.pdf"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
self.assertNotIn("AMENDMENT_INTENT", result.columns)
|
||||
|
||||
@patch(
|
||||
"src.pipelines.shared.postprocessing.postprocessing_funcs.llm_utils.invoke_claude"
|
||||
)
|
||||
def test_skips_llm_for_parent_only_exhibits(self, mock_invoke):
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["f1.pdf", "f2.pdf"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A", "Ex A"],
|
||||
"AMENDMENT_INTENT_LANGUAGE": ["some language", "more language"],
|
||||
"QC_FLAG": ["Parent", "Parent"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
mock_invoke.assert_not_called()
|
||||
# Parent rows should have None
|
||||
self.assertTrue(result["AMENDMENT_INTENT"].isna().all())
|
||||
|
||||
@patch(
|
||||
"src.pipelines.shared.postprocessing.postprocessing_funcs.llm_utils.invoke_claude"
|
||||
)
|
||||
def test_parent_rows_cleared_after_llm(self, mock_invoke):
|
||||
mock_invoke.return_value = '{"intent": "FULL_REPLACEMENT"}'
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["f1.pdf", "f1.pdf", "f1.pdf"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A", "Ex A", "Ex A"],
|
||||
"AMENDMENT_INTENT_LANGUAGE": [
|
||||
"replaces all prior",
|
||||
"replaces all prior",
|
||||
"replaces all prior",
|
||||
],
|
||||
"QC_FLAG": ["Child", "Child", "Parent"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
child_intents = result[result["QC_FLAG"] == "Child"]["AMENDMENT_INTENT"]
|
||||
parent_intents = result[result["QC_FLAG"] == "Parent"]["AMENDMENT_INTENT"]
|
||||
self.assertTrue((child_intents == "FULL_REPLACEMENT").all())
|
||||
self.assertTrue(parent_intents.isna().all())
|
||||
|
||||
@patch(
|
||||
"src.pipelines.shared.postprocessing.postprocessing_funcs.llm_utils.invoke_claude"
|
||||
)
|
||||
def test_empty_language_yields_na_intent(self, mock_invoke):
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["f1.pdf"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A"],
|
||||
"AMENDMENT_INTENT_LANGUAGE": [None],
|
||||
"QC_FLAG": ["Child"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
mock_invoke.assert_not_called()
|
||||
self.assertEqual(result["AMENDMENT_INTENT"].iloc[0], "N/A")
|
||||
|
||||
@patch(
|
||||
"src.pipelines.shared.postprocessing.postprocessing_funcs.llm_utils.invoke_claude"
|
||||
)
|
||||
def test_no_language_child_not_contaminated_by_sibling_in_same_group(
|
||||
self, mock_invoke
|
||||
):
|
||||
# parent_file has language → LLM returns FULL_REPLACEMENT
|
||||
# child_file has no language at all → must get N/A, not inherit from sibling
|
||||
mock_invoke.return_value = '{"intent": "FULL_REPLACEMENT"}'
|
||||
df = pd.DataFrame(
|
||||
{
|
||||
"FILE_NAME": ["parent_file.pdf", "parent_file.pdf", "child_file.pdf"],
|
||||
"AARETE_DERIVED_EXHIBIT_TITLE": ["Ex A", "Ex A", "Ex A"],
|
||||
"AMENDMENT_INTENT_LANGUAGE": [
|
||||
"replaces all prior rates",
|
||||
"replaces all prior rates",
|
||||
None,
|
||||
],
|
||||
"QC_FLAG": ["Child", "Child", "Child"],
|
||||
"GROUP": ["G1", "G1", "G1"],
|
||||
}
|
||||
)
|
||||
result = add_amendment_intent(df)
|
||||
parent_intents = result[result["FILE_NAME"] == "parent_file.pdf"][
|
||||
"AMENDMENT_INTENT"
|
||||
]
|
||||
child_intents = result[result["FILE_NAME"] == "child_file.pdf"][
|
||||
"AMENDMENT_INTENT"
|
||||
]
|
||||
self.assertTrue((parent_intents == "FULL_REPLACEMENT").all())
|
||||
self.assertEqual(child_intents.iloc[0], "N/A")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user