import sys import time _T0 = time.perf_counter() # absolute zero — for measuring full window-open path def _ms(t): return (time.perf_counter() - t) * 1000 import os import copy import sqlite3 import numpy as np import pandas as pd from datetime import datetime print(f"[timing] imports (stdlib + pandas + numpy): {_ms(_T0):7.1f} ms") _T_AFTER_DATA_IMPORTS = time.perf_counter() from PySide6.QtWidgets import ( QApplication, QMainWindow, QTreeView, QVBoxLayout, QHBoxLayout, QPushButton, QWidget, QStackedWidget, QComboBox, QLabel, QCheckBox, QMessageBox, QDialog, QListWidget, QDialogButtonBox, QMenu, QAbstractItemView, QRadioButton, QButtonGroup, QTabWidget, QTabBar ) from PySide6.QtGui import QStandardItemModel, QStandardItem, QColor, QBrush, QFont from PySide6.QtCore import Qt, QTimer, QAbstractItemModel, QModelIndex, QEvent, Signal print(f"[timing] PySide6 imports: {_ms(_T_AFTER_DATA_IMPORTS):7.1f} ms") # ============================================================ # MRR MOVEMENT CATEGORY CODES # ============================================================ # # 2 Add On, New SOW / Amendment # 21 Add On, New SOW / Amendment (incr. quantity) # 22 Add On, New SOW / Amendment (incr. pricing) # 23 Add On, BCO (standard) # 24 Add On, BCO (incr. quantity) # 25 Upgrade at Renewal # 101 Overages / Underages # 100 Accounting Driven # 1 New Customer # 11 New Customer (BCO) # 12 New Customer (Amendment) [manual] # 13 New Customer on Ramp [manual] # 14 New Customer, Returning [manual] # -2 Downgrade, New SOW / Amendment # -21 Downgrade, New SOW / Amendment (decr. quantity) # -22 Downgrade, New SOW / Amendment (decr. pricing) # -23 Downgrade, BCO (standard) # -24 Downgrade, BCO (decr. quantity) # -25 Downgrade at Renewal # -1 Lost customer, voluntary # -3 Lost customer, involuntary [manual] # Custom Qt item data roles for category cells ROLE_CUSTOMER_ID = Qt.UserRole + 1 ROLE_CONTRACT_ID = Qt.UserRole + 2 ROLE_MONTH = Qt.UserRole + 3 ROLE_IS_CATEGORY_CELL = Qt.UserRole + 4 ROLE_CATEGORY_CODE = Qt.UserRole + 5 ROLE_REF_NO = Qt.UserRole + 6 ROLE_IS_UNASSIGNED_REF = Qt.UserRole + 7 ROLE_ASSIGNED_CONTRACT_ID = Qt.UserRole + 8 # current contract_id for pre-selection in dialog ROLE_CAN_CLEAR = Qt.UserRole + 9 # True only when a manual DB assignment exists and can be cleared # Sentinel values produced by heal_and_fill when source data lacks a real id. # Used to decide whether a row is interactive (i.e. has a real customer/contract). MISSING_SENTINELS = {"Missing ID", "Missing Info", "Uncategorized"} # Ordered list of all valid category codes with descriptions ALL_CATEGORIES = [ (1, "New Customer"), (11, "New Customer (BCO)"), (12, "New Customer (Amendment)"), (13, "New Customer on Ramp"), (14, "New Customer, Returning"), (2, "Add On, New SOW / Amendment"), (21, "Add On (increase quantity)"), (22, "Add On (increase pricing)"), (23, "Add On, BCO (standard)"), (24, "Add On, BCO (increase quantity)"), (25, "Upgrade at Renewal"), (100, "Accounting Driven"), (101, "Overages / Underages"), (-1, "Lost Customer, Voluntary"), (-2, "Downgrade, New SOW / Amendment"), (-21, "Downgrade (decrease quantity)"), (-22, "Downgrade (decrease pricing)"), (-23, "Downgrade, BCO (standard)"), (-24, "Downgrade, BCO (decrease quantity)"), (-25, "Downgrade at Renewal"), (-3, "Lost Customer, Involuntary"), ] # ============================================================ # DATABASE & UTILS # ============================================================ def init_database(): db_path = "data_source/mrr_data.db" os.makedirs("data_source", exist_ok=True) conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='contracts'") exists = cursor.fetchone() if not exists: print("--- [CHECK] Database not found. Creating new DB from CSV sources... ---") try: df_l = pd.read_csv("data_source/ledger.csv", low_memory=False) df_c = pd.read_csv("data_source/contracts.csv", low_memory=False) df_l = clean_headers(df_l) df_c = clean_headers(df_c) df_l.to_sql("ledger", conn, index=False) df_c.to_sql("contracts", conn, index=False) conn.execute("CREATE TABLE manual_categorization (customer_name TEXT, contract_name TEXT, line_number TEXT, category TEXT, updated_at TIMESTAMP)") conn.execute("CREATE TABLE write_locks (lock_key TEXT PRIMARY KEY, user_name TEXT, timestamp DATETIME)") conn.commit() print("--- [CHECK] Database initialized successfully. ---") except Exception as e: print(f"--- [ERROR] Initial Database Creation Failed: {e} ---") else: print("--- [CHECK] Existing database found. Skipping CSV ingestion. ---") # Non-destructive top-up: load contractsbilling.csv into its own table if missing. # Lets older databases pick up the new billing source without a rebuild. cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='contracts_billing'") if not cursor.fetchone(): billing_csv = "data_source/contractsbilling.csv" if os.path.exists(billing_csv): print("--- [CHECK] Loading contractsbilling.csv into contracts_billing table... ---") try: df_b = pd.read_csv(billing_csv, low_memory=False) df_b = clean_headers(df_b) df_b.to_sql("contracts_billing", conn, index=False) conn.commit() print(f"--- [CHECK] Loaded {len(df_b)} billing rows. ---") except Exception as e: print(f"--- [ERROR] Billing data load failed: {e} ---") else: print(f"--- [WARN] {billing_csv} not found; backlog view will be empty. ---") return conn def clean_headers(df): df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_") return df def ensure_category_overrides_table(): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute(""" CREATE TABLE IF NOT EXISTS category_overrides ( customer_id TEXT, contract_id TEXT, month TEXT, category INTEGER, updated_at TIMESTAMP, PRIMARY KEY (customer_id, contract_id, month) ) """) conn.commit() conn.close() def load_category_overrides() -> dict: """Return {(customer_id, contract_id): {month: category_code}} for all saved overrides.""" conn = sqlite3.connect("data_source/mrr_data.db") try: df = pd.read_sql("SELECT customer_id, contract_id, month, category FROM category_overrides", conn) except Exception: return {} finally: conn.close() result = {} for _, row in df.iterrows(): key = (str(row["customer_id"]), str(row["contract_id"])) result.setdefault(key, {})[row["month"]] = int(row["category"]) return result def save_category_override(customer_id: str, contract_id: str, month: str, category: int): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute( "INSERT OR REPLACE INTO category_overrides (customer_id, contract_id, month, category, updated_at) VALUES (?, ?, ?, ?, ?)", (customer_id, contract_id, month, category, datetime.now()) ) conn.commit() conn.close() def delete_category_override(customer_id: str, contract_id: str, month: str): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute( "DELETE FROM category_overrides WHERE customer_id=? AND contract_id=? AND month=?", (customer_id, contract_id, month) ) conn.commit() conn.close() # ── Ledger → Contract manual assignments ───────────────────────────────────── def ensure_ledger_assignments_table(): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute(""" CREATE TABLE IF NOT EXISTS ledger_assignments ( reference_no TEXT, customer_id TEXT, contract_id TEXT, contract_name TEXT, updated_at TIMESTAMP, PRIMARY KEY (reference_no, customer_id) ) """) conn.commit() conn.close() def load_ledger_assignments() -> dict: """Return {(reference_no, customer_id): (contract_id, contract_name)}.""" conn = sqlite3.connect("data_source/mrr_data.db") try: df = pd.read_sql( "SELECT reference_no, customer_id, contract_id, contract_name FROM ledger_assignments", conn ) except Exception: return {} finally: conn.close() return { (str(r["reference_no"]), str(r["customer_id"])): (str(r["contract_id"]), str(r["contract_name"])) for _, r in df.iterrows() } def delete_ledger_assignment(reference_no: str, customer_id: str): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute( "DELETE FROM ledger_assignments WHERE reference_no=? AND customer_id=?", (reference_no, customer_id) ) conn.commit() conn.close() def save_ledger_assignment(reference_no: str, customer_id: str, contract_id: str, contract_name: str): conn = sqlite3.connect("data_source/mrr_data.db") conn.execute( "INSERT OR REPLACE INTO ledger_assignments " "(reference_no, customer_id, contract_id, contract_name, updated_at) VALUES (?, ?, ?, ?, ?)", (reference_no, customer_id, contract_id, contract_name, datetime.now()) ) conn.commit() conn.close() def apply_ledger_assignments(df_l: pd.DataFrame, assignments: dict) -> pd.DataFrame: """ Patch contract_id / contract_name on raw ledger rows that have a saved manual assignment. Must run BEFORE build_ledger_pivot so the pivot groups the row under its new contract. The reference number and customer_id keys stored in the assignments dict come from the pivot (post heal_and_fill), where nulls have been collapsed to "Missing Info" / "Missing ID". We normalise the raw DataFrame the same way before matching, otherwise NaN-bearing rows silently fail to match (pandas.Series.astype(str) preserves NaN). """ if not assignments: return df_l ref_col = "reference_no." if "reference_no." in df_l.columns else "reference_no" if ref_col not in df_l.columns or "customer_id" not in df_l.columns: return df_l def _norm(series, null_token): s = series.fillna(null_token).astype(str).str.strip() return s.replace(["", "nan", "None", "NaN", ""], null_token) df_l["_ref_norm"] = _norm(df_l[ref_col], "Missing Info") df_l["_cust_norm"] = _norm(df_l["customer_id"], "Missing ID") # Build a small DataFrame of assignments and merge it onto df_l in one pass, # instead of running a boolean mask scan per assignment. assign_df = pd.DataFrame( [(str(r), str(c), cid, cname) for (r, c), (cid, cname) in assignments.items()], columns=["_ref_norm", "_cust_norm", "_new_contract_id", "_new_contract_name"], ) df_l = df_l.merge(assign_df, on=["_ref_norm", "_cust_norm"], how="left") matched = df_l["_new_contract_id"].notna() df_l.loc[matched, "contract_id"] = df_l.loc[matched, "_new_contract_id"] df_l.loc[matched, "contract_name"] = df_l.loc[matched, "_new_contract_name"] return df_l.drop(columns=["_ref_norm", "_cust_norm", "_new_contract_id", "_new_contract_name"]) # ============================================================ # DATA PROCESSING & MOM CALCULATION # ============================================================ def heal_and_fill(df, index_cols): for id_col in ["customer_id", "contract_id"]: if id_col in df.columns: df[id_col] = df[id_col].astype(str).replace(["", "nan", "None"], "Missing ID") if "customer_id" in df.columns and "customer_name" in df.columns: valid_cust_names = df[df["customer_name"].notna() & (df["customer_name"] != "")].groupby("customer_id")["customer_name"].first() df["customer_name"] = df["customer_name"].fillna(df["customer_id"].map(valid_cust_names)) if "contract_id" in df.columns and "contract_name" in df.columns: valid_cont_names = df[df["contract_name"].notna() & (df["contract_name"] != "")].groupby("contract_id")["contract_name"].first() df["contract_name"] = df["contract_name"].fillna(df["contract_id"].map(valid_cont_names)) for col in index_cols: if col not in df.columns: df[col] = "Uncategorized" if "contract" in col else "Missing Info" elif "contract" in col: df[col] = df[col].fillna("Uncategorized").replace(["", "nan", "None"], "Uncategorized") else: df[col] = df[col].fillna("Missing Info").replace(["", "nan", "None"], "Missing Info") return df # Items excluded from contracts processing (one-off / non-MRR SKUs). # Used by the legacy/full contracts pipeline that feeds discrepancy highlighting # and the categorisation engine. The per-tab views below use their own filters. _EXCLUDED_CONTRACT_ITEMS = ( 'I590', 'I900', 'I901', 'I902', 'I904', 'I905', 'L214', 'L217', 'L231', 'L310', 'L311', 'L198', ) _CONTRACT_INDEX_COLS = ["customer_id", "customer_name", "contract_id", "contract_name", "line_number"] # Contracts View sub-tabs. Each tab shows a pivot filtered to the listed items # (or, for MS, the inverse of the listed exclusion set). _CONTRACT_TAB_PS_ITEMS = ('I900', 'I901', 'I902', 'I903', 'I904', 'I905', 'L231', 'L198') _CONTRACT_TAB_MS_EXCLUDED_ITEMS = ( 'I590', 'I900', 'I901', 'I902', 'I903', 'I904', 'I905', 'L214', 'L217', 'L231', 'L310', 'L311', 'L198', 'L401', ) _CONTRACT_TAB_CLOUD_ITEMS = ('L203', 'L204', 'L401') _CONTRACT_TAB_OTHER_RESELL_ITEMS = ('L214', 'L217', 'L310', 'L311') # Professional Services items — source rows for the Backlog (project backlog) report. _PS_ITEMS_GENERAL = ('I900', 'I901', 'I902', 'I903', 'I905', 'L231') _PS_ITEMS_RECURRING = ('I904',) def build_backlog_data(df: pd.DataFrame, items=_PS_ITEMS_GENERAL): """ Source: contracts_billing table (billing schedule, not revenue). Filter to the given PS item codes, drop status='Terminated' rows, parse month/amount, and aggregate to one row per (customer_id, customer_name, contract_id, contract_name, month). Also returns a set of (customer_id, contract_id) keys for contracts that still have unbilled PS rows (billed=False) — i.e. contracts with remaining scheduled billings. """ if df is None or df.empty: return pd.DataFrame(), set() if 'item' in df.columns: df = df[df['item'].astype(str).isin(items)] if 'status' in df.columns: df = df[df['status'].astype(str).str.strip().str.lower() != 'terminated'] if df.empty: return df.iloc[0:0], set() df = df.copy() parsed_dates = pd.to_datetime(df["scheduled_billing_date"], errors="coerce") df["month"] = parsed_dates.dt.to_period("M").astype(str) df["amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0) df = df.dropna(subset=["month"]) df = heal_and_fill(df, ["customer_id", "customer_name", "contract_id", "contract_name"]) # `billed` round-trips through SQLite as 0/1 INTEGER, but be defensive in # case the column ever lands as 'True'/'False' strings. billed_str = df["billed"].astype(str).str.strip().str.lower() is_billed = billed_str.isin(['true', '1']) df["_unbilled"] = ~is_billed df["_billed_amt"] = df["amount"].where(is_billed, 0.0) has_remaining = ( df.groupby(["customer_id", "contract_id"])["_unbilled"].any() ) has_remaining_keys = { (str(c), str(ct)) for (c, ct), v in has_remaining.items() if v } monthly = ( df.groupby( ["customer_id", "customer_name", "contract_id", "contract_name", "month"], sort=False, ).agg(amount=("amount", "sum"), billed=("_billed_amt", "sum")).reset_index() ) return monthly, has_remaining_keys def _preprocess_contracts_df(df, item_filter=None): """ Filter out excluded/terminated rows, parse month/amount, heal nullable id columns, and ensure mrr_category and item_description exist as strings. *item_filter* is an optional callable taking the item-column Series and returning a boolean mask of rows to keep. If None, the default exclusion list (_EXCLUDED_CONTRACT_ITEMS) is applied — preserving legacy behaviour. """ if 'item' in df.columns: items = df['item'].astype(str) if item_filter is None: df = df[~items.isin(_EXCLUDED_CONTRACT_ITEMS)] else: df = df[item_filter(items)] if 'status' in df.columns: df = df[df['status'].astype(str).str.strip().str.lower() != 'terminated'] df["month"] = pd.to_datetime(df["scheduled_posting_date"], errors="coerce").dt.to_period("M").astype(str) df["amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0) df = df.dropna(subset=["month"]) df = heal_and_fill(df, _CONTRACT_INDEX_COLS) for col in ["mrr_category", "item_description"]: if col not in df.columns: df[col] = "" else: df[col] = df[col].fillna("").astype(str) return df # Order is significant — drives the tab order in the Contracts View. CONTRACT_TAB_KEYS = ('PS', 'MS', 'Cloud', 'Other Resell') def build_contract_outputs(df): """ Preprocess the contracts frame once and return both the pivot (used by the Contracts View) and the detail DataFrame (used by the categorisation engine). Avoids re-running _preprocess_contracts_df twice on the same input. """ df = _preprocess_contracts_df(df) pivot = pd.pivot_table(df, values="amount", index=_CONTRACT_INDEX_COLS, columns="month", aggfunc="sum", fill_value=0).reset_index() detail = ( df.groupby(_CONTRACT_INDEX_COLS + ["mrr_category", "item_description", "month"])["amount"] .sum().reset_index() ) return pivot, detail # Ledger View sub-tabs. MS keeps the historical allowed-account list; PS / Cloud / # Other Resell pull from different revenue accounts. Tab order drives the QTabWidget. _LEDGER_TAB_MS_ACCOUNTS = ( '40101', '40202', '40204', '40205', '40206', '40207', '40209', '40211', '40304', '42006', '43004', '44403', ) _LEDGER_TAB_PS_ACCOUNTS = ( '42001', '42002', '42003', '42004', '42005', '43005', '43006', '43007', '43008', ) _LEDGER_TAB_CLOUD_ACCOUNTS = ('43001', '43002', '43003', '49100') _LEDGER_TAB_OTHER_RESELL_ACCOUNTS = ( '44101', '44102', '44103', '44201', '44202', '44203', '44301', '44302', '44401', '44402', '44404', ) LEDGER_TAB_KEYS = ('PS', 'MS', 'Cloud', 'Other Resell') _LEDGER_TAB_ACCOUNTS = { 'PS': _LEDGER_TAB_PS_ACCOUNTS, 'MS': _LEDGER_TAB_MS_ACCOUNTS, 'Cloud': _LEDGER_TAB_CLOUD_ACCOUNTS, 'Other Resell': _LEDGER_TAB_OTHER_RESELL_ACCOUNTS, } def build_ledger_pivot(df, allowed_accounts=_LEDGER_TAB_MS_ACCOUNTS): df = df.copy() df['base_account'] = df['account'].astype(str).str.split('-').str[0] df = df[df['base_account'].isin(allowed_accounts)].copy() ref_col = "reference_no." if "reference_no." in df.columns else "reference_no" index_cols = ["customer_id", "customer_name", "contract_id", "contract_name", ref_col] df = heal_and_fill(df, index_cols) df = df[df["customer_id"] != "Missing Info"] # drop rows with no identifiable customer df["month"] = pd.to_datetime(df["posting_date"], errors="coerce").dt.to_period("M").astype(str) df["amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0) * (-1) pivot = pd.pivot_table(df, values="amount", index=index_cols, columns="month", aggfunc="sum", fill_value=0) return pivot.reset_index() def calculate_mom_diff(df): meta_cols = [c for c in df.columns if "-" not in str(c)] month_cols = [c for c in df.columns if "-" in str(c)] diff_df = df[meta_cols].copy() for i in range(len(month_cols)): curr_m = month_cols[i] if i == 0: diff_df[curr_m] = df[curr_m] else: prev_m = month_cols[i-1] diff_df[curr_m] = df[curr_m] - df[prev_m] return diff_df # ============================================================ # MRR CATEGORIZATION ENGINE # ============================================================ def build_contract_category_map(detail_df: pd.DataFrame, all_months: list) -> dict: """ Returns a nested dict: category_map[(customer_id, contract_id)][month] = dominant_category_code (int or 0) Logic per (customer_id, contract_id, month): 1. Compute MoM diff per line (comparing to same line's amount in previous month). 2. Classify each line's diff into a bucket via the np.select cascade below. 3. The dominant bucket is whichever has the largest abs() contribution. 4. If no movement → 0. Vectorised end-to-end (~12s → <1s on the production dataset). """ if detail_df is None or detail_df.empty: return {} id_cols = ["customer_id", "contract_id", "line_number", "mrr_category", "item_description"] line_pivot = detail_df.pivot_table( values="amount", index=id_cols, columns="month", aggfunc="sum", fill_value=0, ) # Initialise category_map with 0 for every (customer, contract) × every month. sorted_months = sorted(all_months) cust_contract_pairs = ( line_pivot.index.to_frame(index=False)[["customer_id", "contract_id"]] .drop_duplicates() ) category_map = { (str(c), str(ct)): {m: 0 for m in sorted_months} for c, ct in cust_contract_pairs.itertuples(index=False, name=None) } # Time axis: every month in `all_months`. Missing columns become zeros, so # diff against them naturally yields zero movement. amounts = line_pivot.reindex(columns=sorted_months, fill_value=0) prev = amounts.shift(1, axis=1).fillna(0) diffs = amounts - prev # "New customer" flag: fewer than 12 months of non-zero MRR anywhere in the pivot. lp_months = [c for c in line_pivot.columns if "-" in str(c)] if lp_months: cust_nonzero_months = ( (line_pivot[lp_months].abs() > 0.01) .groupby(level="customer_id").any().sum(axis=1) ) is_new_cust_series = cust_nonzero_months < 12 else: is_new_cust_series = pd.Series(dtype=bool) # "Drops to zero" flag per (customer, contract, month). ct_totals = amounts.groupby(level=["customer_id", "contract_id"]).sum() ct_prev = ct_totals.shift(1, axis=1).fillna(0) drops_df = (ct_totals.abs() < 0.01) & (ct_prev.abs() > 0.01) # Melt to long form and keep only meaningful movements. diffs_long = ( diffs.reset_index() .melt(id_vars=id_cols, value_vars=sorted_months, var_name="month", value_name="diff") ) diffs_long = diffs_long[diffs_long["diff"].abs() >= 0.01] if diffs_long.empty: return category_map # Join the per-row flags. diffs_long["is_new_cust"] = ( diffs_long["customer_id"].map(is_new_cust_series).fillna(True).astype(bool) ) drops_long = drops_df.stack().rename("drops_to_zero") diffs_long = diffs_long.merge( drops_long, left_on=["customer_id", "contract_id", "month"], right_index=True, how="left", ) diffs_long["drops_to_zero"] = diffs_long["drops_to_zero"].fillna(False).astype(bool) # Classification cascade. np.select picks the first matching condition, # so order encodes priority (new-customer ⟶ lost ⟶ renewal ⟶ BCO ⟶ standard). cat = diffs_long["mrr_category"].fillna("").astype(str).str.strip() desc = diffs_long["item_description"].fillna("").astype(str).str.strip().str.upper() is_bco_v = desc.str.startswith("RITM") increasing = diffs_long["diff"] > 0 is_new = diffs_long["is_new_cust"] drops = diffs_long["drops_to_zero"] cat_qty = cat == "Quantity" cat_price = cat == "Pricing" cat_up = cat == "Upgrade at Renewal" cat_down = cat == "Downgrade at Renewal" conditions = [ is_new & is_bco_v, # 11 — new customer (BCO) is_new, # 1 — new customer drops & ~increasing, # -1 — lost customer (voluntary) cat_up, # 25 — upgrade at renewal cat_down, # -25 — downgrade at renewal is_bco_v & increasing & cat_qty, # 24 — BCO add-on, quantity is_bco_v & increasing, # 23 — BCO add-on, standard is_bco_v & ~increasing & cat_qty, # -24 — BCO downgrade, quantity is_bco_v & ~increasing, # -23 — BCO downgrade, standard increasing & cat_qty, # 21 — add-on, quantity increasing & cat_price, # 22 — add-on, pricing increasing, # 2 — add-on, amendment / SOW ~increasing & cat_qty, # -21 — downgrade, quantity ~increasing & cat_price, # -22 — downgrade, pricing # default (decreasing, neither qty nor pricing) → -2 ] codes = [11, 1, -1, 25, -25, 24, 23, -24, -23, 21, 22, 2, -21, -22] diffs_long["code"] = np.select(conditions, codes, default=-2) diffs_long["abs_diff"] = diffs_long["diff"].abs() # Dominant bucket per (customer, contract, month) = code with the largest sum |diff|. # sort=False preserves the row order from melt, so ties resolve to the code that # first appears in the pivot — matching the legacy's dict-insertion tiebreaker. # Round the bucket totals to cents before comparing to neutralise float-precision # noise between pandas' .sum() and the legacy's manual += accumulator. bucket_sums = ( diffs_long.groupby(["customer_id", "contract_id", "month", "code"], sort=False)["abs_diff"] .sum().reset_index() ) bucket_sums["abs_diff"] = bucket_sums["abs_diff"].round(2) dominant_idx = ( bucket_sums.groupby(["customer_id", "contract_id", "month"], sort=False)["abs_diff"].idxmax() ) for row in bucket_sums.loc[dominant_idx].itertuples(index=False): category_map[(str(row.customer_id), str(row.contract_id))][row.month] = int(row.code) return category_map def augment_category_map_from_ledger(category_map: dict, ledger_diff_df: pd.DataFrame, all_months: list) -> dict: """ For (customer_id, contract_id) pairs where the ledger shows a MoM change but the contracts-based category is 0, assign 2 (increase) or -2 (decrease). """ if ledger_diff_df is None or ledger_diff_df.empty: return category_map all_months_set = set(all_months) relevant_months = [c for c in ledger_diff_df.columns if "-" in str(c) and c in all_months_set] if not relevant_months: return category_map # One groupby+sum replaces the per-(cust,contract)×per-month .sum() loop. ct_sums = ledger_diff_df.groupby(["customer_id", "contract_id"], sort=False)[relevant_months].sum() for (cid, ctid), row in ct_sums.iterrows(): key = (str(cid), str(ctid)) if key not in category_map: category_map[key] = {} for month in relevant_months: if category_map[key].get(month, 0) != 0: continue ledger_sum = row[month] if abs(ledger_sum) < 0.01: continue category_map[key][month] = 2 if ledger_sum > 0 else -2 return category_map # ============================================================ # MODEL BUILDER # ============================================================ CATEGORY_ROW_BG = QColor(40, 50, 65) # Dark blue-grey for category rows CATEGORY_ROW_FG = QColor(180, 210, 255) # Light blue text for category codes CATEGORY_ZERO_FG = QColor(90, 100, 110) # Muted grey for zero / no movement CATEGORY_MANUAL_FG = QColor(255, 200, 100) # Amber for manually overridden categories SUMMARY_POS_FG = QColor(100, 220, 140) # Green for positive movements SUMMARY_NEG_FG = QColor(220, 100, 100) # Red for negative movements SUMMARY_TOTAL_BG = QColor(50, 58, 70) # Slightly lifted bg for the total row ASSIGNED_REF_FG = QColor(100, 200, 190) # Teal for manually-assigned (reassignable) refs def _bold(item: QStandardItem) -> QStandardItem: """Set an item's font to bold. Returns the item for chaining.""" f = item.font() f.setBold(True) item.setFont(f) return item def _no_edit(item: QStandardItem) -> QStandardItem: """Make an item non-editable. Returns the item for chaining.""" item.setFlags(item.flags() & ~Qt.ItemIsEditable) return item def _amount_item(val: float, *, fmt: str = ",.0f", threshold: float = 0.01, bold: bool = False, total_bg: bool = False) -> QStandardItem: """ Build a money cell for QStandardItemModel: right-aligned, sign-coloured (green / red), blank for near-zero values, optionally bold and / or with the total-row background. Used across the drill-down dialogs. """ text = format(val, fmt) if abs(val) >= threshold else "" item = _no_edit(QStandardItem(text)) item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) if bold: _bold(item) if total_bg: item.setBackground(QBrush(SUMMARY_TOTAL_BG)) if val > threshold: item.setForeground(QBrush(SUMMARY_POS_FG)) elif val < -threshold: item.setForeground(QBrush(SUMMARY_NEG_FG)) return item def _aggregate_summary_by_category(category_map: dict, contract_totals, month_cols: list) -> dict: """ Sum per-contract MoM amounts into {category_code: {month: total}}. *contract_totals* is a DataFrame (or Series with multi-index) indexed by (customer_id, contract_id) with month columns. Used by both the in-app MRR Summary model and the Excel export. """ cat_amounts: dict[int, dict[str, float]] = {} for (cid, ctid), row in contract_totals.iterrows(): key = (str(cid), str(ctid)) cat_months = category_map.get(key, {}) for month in month_cols: code = cat_months.get(month, 0) if code == 0: continue val = float(row[month]) if abs(val) < 0.01: continue cat_amounts.setdefault(code, {}) cat_amounts[code][month] = cat_amounts[code].get(month, 0.0) + val return cat_amounts # Backlog tree column → info-dict key. Column 0 holds the label. _BACKLOG_COL_KEYS = {1: "tcv", 2: "billed", 3: "remaining"} # ============================================================ # CUSTOM TREE MODEL (replaces QStandardItemModel for the large views) # ============================================================ # # Avoids allocating a QStandardItem per cell — the ledger view alone has # ~25k rows × ~13 visible columns, which previously cost ~1.6s in Qt object # allocation. The custom model holds a lightweight Python Node tree and # computes per-cell role data on demand from cached per-row info. # Pre-built QBrush instances. QBrush is cheap to construct but the views ask # for the brush thousands of times — caching avoids per-call allocations. _BR_CATEGORY_ROW_BG = QBrush(CATEGORY_ROW_BG) _BR_CATEGORY_ROW_FG = QBrush(CATEGORY_ROW_FG) _BR_CATEGORY_ZERO_FG = QBrush(CATEGORY_ZERO_FG) _BR_CATEGORY_MANUAL_FG = QBrush(CATEGORY_MANUAL_FG) _BR_SUMMARY_POS_FG = QBrush(SUMMARY_POS_FG) _BR_SUMMARY_NEG_FG = QBrush(SUMMARY_NEG_FG) _BR_SUMMARY_TOTAL_BG = QBrush(SUMMARY_TOTAL_BG) _BR_ASSIGNED_REF_FG = QBrush(ASSIGNED_REF_FG) _BR_UNASSIGNED_REF_FG = QBrush(QColor(220, 180, 80)) _FONT_CACHE = {} _FONT_SIZE_OFFSET = 2 # Increase default font size by this many points def _qfont(bold=False, italic=False): """Return a cached QFont with the given attributes.""" key = (bold, italic) f = _FONT_CACHE.get(key) if f is None: f = QFont() if bold: f.setBold(True) if italic: f.setItalic(True) f.setPointSize(f.pointSize() + _FONT_SIZE_OFFSET) _FONT_CACHE[key] = f return f class Node: """Tree-model node. Holds a kind tag + free-form info dict for role lookups.""" __slots__ = ("parent", "children", "kind", "info", "row_in_parent") def __init__(self, parent, kind, info): self.parent = parent self.kind = kind self.info = info self.children = [] self.row_in_parent = 0 if parent is not None: self.row_in_parent = len(parent.children) parent.children.append(self) class _MRRModel(QAbstractItemModel): """ Tree model backed by a Node tree. data() dispatches by node.kind. Used by all three views — both the hierarchical contracts/ledger views and the flat MRR summary (whose rows are direct children of root). """ def __init__(self, root, headers, visible_months, manual_keys=None, parent=None): super().__init__(parent) self._root = root self._headers = headers self._visible_months = visible_months self._manual_keys = manual_keys or set() # ── Tree navigation ──────────────────────────────────────────── def rowCount(self, parent=QModelIndex()): if not parent.isValid(): return len(self._root.children) return len(parent.internalPointer().children) def columnCount(self, parent=QModelIndex()): return len(self._headers) def headerData(self, section, orientation, role=Qt.DisplayRole): if role == Qt.DisplayRole and orientation == Qt.Horizontal: return self._headers[section] return None def index(self, row, column, parent=QModelIndex()): if not self.hasIndex(row, column, parent): return QModelIndex() parent_node = self._root if not parent.isValid() else parent.internalPointer() return self.createIndex(row, column, parent_node.children[row]) def parent(self, index): if not index.isValid(): return QModelIndex() parent_node = index.internalPointer().parent if parent_node is None or parent_node is self._root: return QModelIndex() return self.createIndex(parent_node.row_in_parent, 0, parent_node) def flags(self, index): if not index.isValid(): return Qt.NoItemFlags return Qt.ItemIsSelectable | Qt.ItemIsEnabled # never editable # ── data() dispatcher ────────────────────────────────────────── def data(self, index, role=Qt.DisplayRole): if not index.isValid(): return None node = index.internalPointer() col = index.column() kind = node.kind if kind == "customer": return self._customer_data(node.info, col, role) if kind == "contract": return self._contract_data(node.info, col, role) if kind == "category": return self._category_data(node.info, col, role) if kind == "line": return self._line_data(node.info, col, role) if kind == "total": return self._total_data(node.info, col, role) if kind == "summary_row": return self._summary_row_data(node.info, col, role) if kind == "summary_total": return self._summary_total_data(node.info, col, role) if kind == "backlog_customer": return self._backlog_customer_data(node.info, col, role) if kind == "backlog_contract": return self._backlog_contract_data(node.info, col, role) if kind == "backlog_month": return self._backlog_month_data(node.info, col, role) if kind == "backlog_total": return self._backlog_total_data(node.info, col, role) return None # ── Per-kind data lookups ────────────────────────────────────── def _customer_data(self, d, col, role): if role == Qt.BackgroundRole: return d.get("brush") if col == 0: if role == Qt.DisplayRole: return f"{d['cid']} - {d['cname']}" return None if role == Qt.DisplayRole: v = d["month_totals"].get(self._visible_months[col - 1], 0) return f"{v:,.2f}" if v != 0 else "" return None def _contract_data(self, d, col, role): if role == Qt.BackgroundRole: return d.get("brush") if col == 0: if role == Qt.DisplayRole: return f"{d['ctid']} - {d['ctname']}" return None if role == Qt.DisplayRole: v = d["month_totals"].get(self._visible_months[col - 1], 0) return f"{v:,.2f}" if v != 0 else "" return None def _category_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return "Category" if role == Qt.BackgroundRole: return _BR_CATEGORY_ROW_BG if role == Qt.ForegroundRole: return _BR_CATEGORY_ROW_FG if role == Qt.FontRole: return _qfont(bold=True, italic=True) return None month = self._visible_months[col - 1] code = d["codes"].get(month, 0) if role == Qt.DisplayRole: return str(code) if code != 0 else "–" if role == Qt.BackgroundRole: return _BR_CATEGORY_ROW_BG if role == Qt.ForegroundRole: if code == 0: return _BR_CATEGORY_ZERO_FG is_manual = (d["cid"], d["ctid"], month) in self._manual_keys return _BR_CATEGORY_MANUAL_FG if is_manual else _BR_CATEGORY_ROW_FG if role == Qt.FontRole and code != 0: return _qfont(bold=True) if role == Qt.TextAlignmentRole: return Qt.AlignCenter # Custom roles consumed by on_category_clicked: if role == ROLE_IS_CATEGORY_CELL: return True if role == ROLE_CUSTOMER_ID: return d["cid"] if role == ROLE_CONTRACT_ID: return d["ctid"] if role == ROLE_MONTH: return month return None def _line_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return d["label"] if role == Qt.ToolTipRole: return d.get("tooltip") if role == Qt.ForegroundRole: return d.get("label_fg") if role == Qt.FontRole and d.get("italic"): return _qfont(italic=True) else: if role == Qt.DisplayRole: v = d["month_values"].get(self._visible_months[col - 1], 0) return f"{v:,.2f}" if v != 0 else "" # Stamp the interactive-row roles on EVERY column so any clicked cell works. if d.get("is_interactive"): if role == ROLE_IS_UNASSIGNED_REF: return True if role == ROLE_CUSTOMER_ID: return d["cid"] if role == ROLE_REF_NO: return d["ref_text"] if role == ROLE_ASSIGNED_CONTRACT_ID: return d.get("assigned_ctid") if role == ROLE_CAN_CLEAR: return d.get("can_clear", False) return None def _total_data(self, d, col, role): if role == Qt.BackgroundRole: return _BR_SUMMARY_TOTAL_BG if role == Qt.FontRole: return _qfont(bold=True) if col == 0: if role == Qt.DisplayRole: return "Total" return None if role == Qt.DisplayRole: v = d["month_sums"].get(self._visible_months[col - 1], 0) return f"{v:,.2f}" if v != 0 else "" return None def _summary_row_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return d["label"] return None month = self._visible_months[col - 1] val = d["amounts"].get(month, 0.0) if role == Qt.DisplayRole: return f"{val:,.0f}" if abs(val) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignRight | Qt.AlignVCenter if role == Qt.ForegroundRole: if val > 0.01: return _BR_SUMMARY_POS_FG if val < -0.01: return _BR_SUMMARY_NEG_FG if role == ROLE_CATEGORY_CODE: return d["code"] if role == ROLE_MONTH: return month return None def _summary_total_data(self, d, col, role): if role == Qt.BackgroundRole: return _BR_SUMMARY_TOTAL_BG if role == Qt.FontRole: return _qfont(bold=True) if col == 0: if role == Qt.DisplayRole: return "Net MRR Change" return None month = self._visible_months[col - 1] val = d["totals_by_month"].get(month, 0.0) if role == Qt.DisplayRole: return f"{val:,.0f}" if abs(val) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignRight | Qt.AlignVCenter if role == Qt.ForegroundRole: if val > 0.01: return _BR_SUMMARY_POS_FG if val < -0.01: return _BR_SUMMARY_NEG_FG return None # ── Backlog view (Customer → Contract → Month) ───────────────── # Columns: 0=Name, 1=TCV, 2=Invoice Amount, 3=Remaining Balance def _backlog_customer_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return f"{d['cid']} - {d['cname']}" return None key = _BACKLOG_COL_KEYS.get(col) if key is None: return None if role == Qt.DisplayRole: v = d.get(key, 0.0) return f"{v:,.2f}" if abs(v) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None def _backlog_contract_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return f"{d['ctid']} - {d['ctname']}" return None key = _BACKLOG_COL_KEYS.get(col) if key is None: return None if role == Qt.DisplayRole: v = d.get(key, 0.0) return f"{v:,.2f}" if abs(v) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None def _backlog_total_data(self, d, col, role): if role == Qt.BackgroundRole: return _BR_SUMMARY_TOTAL_BG if role == Qt.FontRole: return _qfont(bold=True) if col == 0: if role == Qt.DisplayRole: return "Total" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None key = _BACKLOG_COL_KEYS.get(col) if key is None: return None if role == Qt.DisplayRole: v = d.get(key, 0.0) return f"{v:,.2f}" if abs(v) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None def _backlog_month_data(self, d, col, role): if col == 0: if role == Qt.DisplayRole: return d.get("month", "") if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None if col == 2: if role == Qt.DisplayRole: v = d.get("amount", 0.0) return f"{v:,.2f}" if abs(v) >= 0.01 else "" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter if col == 3: if role == Qt.DisplayRole: v = d.get("remaining", 0.0) return f"{v:,.2f}" if role == Qt.TextAlignmentRole: return Qt.AlignLeft | Qt.AlignVCenter return None def build_qt_model(pivot_df, selected_highlight_month, visible_months, is_ledger=False, is_diff_view=False, reference_totals=None, category_map=None, manual_keys=None, assigned_refs=None): root = Node(None, "root", {}) headers = ["Name"] + list(visible_months) if pivot_df.empty: return _MRRModel(root, headers, visible_months, manual_keys) cid_col, cname_col = pivot_df.columns[0], pivot_df.columns[1] ctid_col, ctname_col = pivot_df.columns[2], pivot_df.columns[3] ref_col = pivot_df.columns[4] pivot_df = pivot_df.sort_values([cname_col, ctname_col], kind="stable") for (c_id, c_name), c_group in pivot_df.groupby([cid_col, cname_col], sort=False): if is_diff_view: hv = c_group[selected_highlight_month].sum() if selected_highlight_month in c_group.columns else 0 if abs(hv) < 0.01: continue color = None if is_ledger and reference_totals: color = reference_totals.get(c_id) or reference_totals.get(c_name) brush = QBrush(color) if color else None c_totals = {m: float(c_group[m].sum()) if m in c_group.columns else 0 for m in visible_months} customer_node = Node(root, "customer", { "cid": c_id, "cname": c_name, "month_totals": c_totals, "brush": brush, }) for (ct_id, ct_name), sub_group in c_group.groupby([ctid_col, ctname_col], sort=False): if is_diff_view: sh = sub_group[selected_highlight_month].sum() if selected_highlight_month in sub_group.columns else 0 if abs(sh) < 0.01: continue # Category row (ledger view only) is a sibling of the contract row. if is_ledger and category_map is not None: key = (str(c_id), str(ct_id)) Node(customer_node, "category", { "cid": str(c_id), "ctid": str(ct_id), "codes": category_map.get(key, {}), }) ct_totals = {m: float(sub_group[m].sum()) if m in sub_group.columns else 0 for m in visible_months} contract_node = Node(customer_node, "contract", { "ctid": ct_id, "ctname": ct_name, "month_totals": ct_totals, "brush": brush, }) is_interactive = is_ledger and str(c_id) not in MISSING_SENTINELS is_unassigned_ref_row = is_interactive and str(ct_id) == "Uncategorized" for _, row in sub_group.iterrows(): if is_diff_view: lh = row[selected_highlight_month] if selected_highlight_month in row.index else 0 if abs(lh) < 0.01: continue label = f"Ref {row[ref_col]}" if is_ledger else f"Line {row[ref_col]}" ref_text = str(row[ref_col]) has_assignment = ( is_interactive and assigned_refs is not None and (ref_text, str(c_id)) in assigned_refs ) if is_unassigned_ref_row: label_fg, italic = _BR_UNASSIGNED_REF_FG, True tooltip = "No contract assigned. Right-click or double-click to assign." elif has_assignment: label_fg, italic = _BR_ASSIGNED_REF_FG, True tooltip = "Manually assigned. Right-click or double-click to change or clear." elif is_interactive: label_fg, italic = None, False tooltip = "Right-click or double-click to move to a different contract." else: label_fg, italic, tooltip = None, False, None line_values = {m: float(row[m]) if m in row.index else 0 for m in visible_months} Node(contract_node, "line", { "label": label, "ref_text": ref_text, "month_values": line_values, "cid": str(c_id), "ctid": str(ct_id), "is_interactive": is_interactive, "assigned_ctid": (str(ct_id) if (is_interactive and not is_unassigned_ref_row) else None), "can_clear": has_assignment, "label_fg": label_fg, "italic": italic, "tooltip": tooltip, }) total_sums = {m: float(pivot_df[m].sum()) if m in pivot_df.columns else 0 for m in visible_months} Node(root, "total", {"month_sums": total_sums}) return _MRRModel(root, headers, visible_months, manual_keys) def build_category_summary_model(category_map, p_l_diff, visible_months): root = Node(None, "root", {}) headers = ["Category"] + list(visible_months) if p_l_diff is None or p_l_diff.empty or not visible_months: return _MRRModel(root, headers, visible_months) cid_col = p_l_diff.columns[0] ctid_col = p_l_diff.columns[2] month_cols = [m for m in visible_months if m in p_l_diff.columns] contract_totals = p_l_diff.groupby([cid_col, ctid_col])[month_cols].sum() cat_amounts = _aggregate_summary_by_category(category_map, contract_totals, month_cols) if not cat_amounts: return _MRRModel(root, headers, visible_months) totals_by_month: dict[str, float] = {} for code, label in ALL_CATEGORIES: if code not in cat_amounts: continue amounts = cat_amounts[code] Node(root, "summary_row", { "code": code, "label": f"{code} — {label}", "amounts": amounts, }) for m in month_cols: totals_by_month[m] = totals_by_month.get(m, 0.0) + amounts.get(m, 0.0) Node(root, "summary_total", {"totals_by_month": totals_by_month}) return _MRRModel(root, headers, visible_months) def build_backlog_model(backlog_df, has_remaining_keys=None, outstanding_only=False): """ Build the Backlog tree model: Customer → Contract → Month. For each contract, TCV is the sum of all PS-line amounts on it. Months under the contract are listed chronologically; the Remaining Balance column shows TCV decreasing by each month's invoice amount until 0. When *outstanding_only* is True, hide any contract whose key is not in *has_remaining_keys* — i.e. contracts with no scheduled_posting_date in the future. Customers left with no visible contracts are dropped. """ headers = ["Name", "TCV", "Billed to Date", "Remaining Balance"] root = Node(None, "root", {}) if backlog_df is None or backlog_df.empty: return _MRRModel(root, headers, []) df = backlog_df.sort_values( ["customer_name", "contract_name", "month"], kind="stable" ) keys = has_remaining_keys or set() total_tcv = 0.0 total_billed = 0.0 total_remaining = 0.0 for (c_id, c_name), c_group in df.groupby( ["customer_id", "customer_name"], sort=False ): # Build contracts first so we can skip empty customers when filtering. contracts = [] for (ct_id, ct_name), ct_group in c_group.groupby( ["contract_id", "contract_name"], sort=False ): if outstanding_only and (str(c_id), str(ct_id)) not in keys: continue tcv = float(ct_group["amount"].sum()) billed_to_date = float(ct_group["billed"].sum()) remaining = tcv - billed_to_date month_rows = ct_group.groupby("month", sort=True).agg( amount=("amount", "sum"), billed=("billed", "sum") ) contracts.append((ct_id, ct_name, tcv, billed_to_date, remaining, month_rows)) if not contracts: continue cust_tcv = sum(t for _, _, t, _, _, _ in contracts) cust_billed = sum(b for _, _, _, b, _, _ in contracts) cust_remaining = sum(r for _, _, _, _, r, _ in contracts) total_tcv += cust_tcv total_billed += cust_billed total_remaining += cust_remaining cust_node = Node(root, "backlog_customer", { "cid": c_id, "cname": c_name, "tcv": cust_tcv, "billed": cust_billed, "remaining": cust_remaining, }) for ct_id, ct_name, tcv, billed_to_date, remaining, month_rows in contracts: contract_node = Node(cust_node, "backlog_contract", { "ctid": ct_id, "ctname": ct_name, "tcv": tcv, "billed": billed_to_date, "remaining": remaining, }) running = tcv for month, row in month_rows.iterrows(): billed_m = float(row["billed"]) running -= billed_m Node(contract_node, "backlog_month", { "month": str(month), "amount": billed_m, "remaining": running, }) if root.children: Node(root, "backlog_total", { "tcv": total_tcv, "billed": total_billed, "remaining": total_remaining, }) return _MRRModel(root, headers, []) # ============================================================ # DRILL-DOWN DIALOG (double-click a summary cell) # ============================================================ class DrillDownDialog(QDialog): """ First-level drill-down from an MRR Summary cell: shows the list of customers / contracts contributing to that category × month, with monthly diff values across the visible range. Double-clicking a number cell opens a second-level dialog (TransactionBreakdownDialog) with the underlying ledger postings for that specific (customer, contract, month). """ def __init__(self, category_code: int, clicked_month: str, category_map: dict, p_l_diff, visible_months: list, df_ledger, parent=None): super().__init__(parent) label = next((lbl for c, lbl in ALL_CATEGORIES if c == category_code), str(category_code)) self.setWindowTitle(f"{category_code} — {label} · {clicked_month}") self.resize(1000, 420) self._df_ledger = df_ledger layout = QVBoxLayout(self) layout.addWidget(QLabel( f"Customers in {category_code} — {label} for {clicked_month}" f", amounts shown for all selected months. " f"Double-click a number to see the underlying ledger transactions." )) cid_col = p_l_diff.columns[0] cname_col = p_l_diff.columns[1] ctid_col = p_l_diff.columns[2] ctname_col = p_l_diff.columns[3] month_cols = [m for m in visible_months if m in p_l_diff.columns] contract_totals = p_l_diff.groupby([cid_col, ctid_col])[month_cols].sum() name_df = p_l_diff.groupby([cid_col, ctid_col])[[cname_col, ctname_col]].first() model = QStandardItemModel() model.setHorizontalHeaderLabels(["Customer / Contract"] + visible_months) totals: dict[str, float] = {} any_rows = False # Collect matching rows, resolve names, then sort by customer name matching = [] for (cid, ctid), row in contract_totals.iterrows(): key = (str(cid), str(ctid)) if category_map.get(key, {}).get(clicked_month) != category_code: continue val_in_month = float(row[clicked_month]) if clicked_month in row.index else 0.0 if abs(val_in_month) < 0.01: continue try: c_name = name_df.loc[(cid, ctid), cname_col] ct_name = name_df.loc[(cid, ctid), ctname_col] except KeyError: c_name, ct_name = str(cid), str(ctid) matching.append((str(c_name), str(ct_name), row, str(cid), str(ctid))) matching.sort(key=lambda x: (x[0].lower(), x[1].lower())) for c_name, ct_name, row, cid, ctid in matching: any_rows = True tooltip = f"{c_name} / {ct_name}" row_items = [_no_edit(QStandardItem(tooltip))] for m in visible_months: val = float(row[m]) if m in row.index else 0.0 cell = _amount_item(val) # Tag the cell so double-click can drill into it. cell.setData(cid, ROLE_CUSTOMER_ID) cell.setData(ctid, ROLE_CONTRACT_ID) cell.setData(m, ROLE_MONTH) cell.setData(tooltip, Qt.ToolTipRole) row_items.append(cell) if m in month_cols: totals[m] = totals.get(m, 0.0) + val model.appendRow(row_items) if not any_rows: model.appendRow([QStandardItem("No contributing customers found.")]) # ── Total row ───────────────────────────────────────────────────── total_label = _no_edit(_bold(QStandardItem("Total"))) total_label.setBackground(QBrush(SUMMARY_TOTAL_BG)) total_row = [total_label] + [ _amount_item(totals.get(m, 0.0), bold=True, total_bg=True) for m in visible_months ] model.appendRow(total_row) tree = QTreeView() tree.setRootIsDecorated(False) tree.setModel(model) tree.resizeColumnToContents(0) tree.doubleClicked.connect(self._on_cell_double_clicked) layout.addWidget(tree) btn = QPushButton("Close") btn.clicked.connect(self.accept) layout.addWidget(btn) def _on_cell_double_clicked(self, index): if not index.isValid() or index.column() == 0: return cid = index.data(ROLE_CUSTOMER_ID) ctid = index.data(ROLE_CONTRACT_ID) month = index.data(ROLE_MONTH) if cid is None or ctid is None or month is None: return TransactionBreakdownDialog(cid, ctid, month, self._df_ledger, parent=self).exec() class TransactionBreakdownDialog(QDialog): """ Second-level drill-down: shows the raw ledger postings for a single (customer, contract, month) cell from DrillDownDialog. Includes the prior month so the MoM diff is visible (clicked - prior = the cell value). """ def __init__(self, customer_id: str, contract_id: str, clicked_month: str, df_ledger, parent=None): super().__init__(parent) try: prior_month = str(pd.Period(clicked_month, freq='M') - 1) except Exception: prior_month = "" ref_col = 'reference_no.' if 'reference_no.' in df_ledger.columns else 'reference_no' title_col = 'account_title' if 'account_title' in df_ledger.columns else None # Filter to the specific (customer, contract) for the two months. df = df_ledger.copy() df['_base_account'] = df['account'].astype(str).str.split('-').str[0] df = df[df['_base_account'].isin(_LEDGER_TAB_MS_ACCOUNTS)] df = df[df['customer_id'].astype(str) == str(customer_id)] df = df[df['contract_id'].astype(str) == str(contract_id)] df['_month'] = pd.to_datetime(df['posting_date'], errors='coerce').dt.to_period('M').astype(str) df['_amount'] = pd.to_numeric(df['amount'], errors='coerce').fillna(0) * (-1) df = df[df['_month'].isin([clicked_month, prior_month])] # Resolve display names from the filtered slice (or fall back to ids). if not df.empty: customer_name = str(df['customer_name'].iloc[0]) contract_name = str(df['contract_name'].iloc[0]) else: customer_name, contract_name = str(customer_id), str(contract_id) self.setWindowTitle(f"{customer_name} / {contract_name} · {clicked_month}") self.resize(1000, 460) clicked_total = df.loc[df['_month'] == clicked_month, '_amount'].sum() prior_total = df.loc[df['_month'] == prior_month, '_amount'].sum() diff = clicked_total - prior_total layout = QVBoxLayout(self) layout.addWidget(QLabel( f"{customer_name} / {contract_name}
" f"{prior_month}: {prior_total:,.0f} → " f"{clicked_month}: {clicked_total:,.0f} " f"(MoM diff: {diff:,.0f})" )) headers = ["Posting Date", "Month", "Reference", "Account"] if title_col is not None: headers.append("Description") headers.append("Amount") model = QStandardItemModel() model.setHorizontalHeaderLabels(headers) if df.empty: model.appendRow([QStandardItem("No ledger postings found for this cell.")]) else: df = df.sort_values(['posting_date']) for _, tx in df.iterrows(): row = [ _no_edit(QStandardItem(str(tx['posting_date']))), _no_edit(QStandardItem(str(tx['_month']))), _no_edit(QStandardItem(str(tx[ref_col]))), _no_edit(QStandardItem(str(tx['account']))), ] if title_col is not None: row.append(_no_edit(QStandardItem(str(tx[title_col])))) row.append(_amount_item(float(tx['_amount']), fmt=",.2f", threshold=0.005)) model.appendRow(row) # Per-month total rows + diff total row. def _total_row(label_text, val): lbl = _no_edit(_bold(QStandardItem(label_text))) lbl.setBackground(QBrush(SUMMARY_TOTAL_BG)) blanks = [] for _ in range(len(headers) - 2): b = _no_edit(QStandardItem("")) b.setBackground(QBrush(SUMMARY_TOTAL_BG)) blanks.append(b) amt = _amount_item(val, fmt=",.2f", threshold=0.005, bold=True, total_bg=True) return [lbl, *blanks, amt] model.appendRow(_total_row(f"Total {prior_month}", prior_total)) model.appendRow(_total_row(f"Total {clicked_month}", clicked_total)) model.appendRow(_total_row("MoM diff", diff)) tree = QTreeView() tree.setRootIsDecorated(False) tree.setModel(model) for col in range(model.columnCount()): tree.resizeColumnToContents(col) layout.addWidget(tree) btn = QPushButton("Close") btn.clicked.connect(self.accept) layout.addWidget(btn) # ============================================================ # ASSIGN UNASSIGNED LEDGER TRANSACTION TO A CONTRACT # ============================================================ class AssignContractDialog(QDialog): def __init__(self, customer_id: str, reference_no: str, current_contract_id: str = None, can_clear: bool = False, parent=None): super().__init__(parent) self.selected_contract = None # (contract_id, contract_name) on accept self.clear_requested = False self._current_contract_id = current_contract_id self._customer_contracts = self._load_contracts(customer_id) self._all_contracts = self._load_all_contracts() is_reassign = current_contract_id is not None self.setWindowTitle("Change Contract Assignment" if is_reassign else "Assign Transaction to Contract") self.setMinimumWidth(560) self.setMinimumHeight(400) layout = QVBoxLayout(self) action_word = "Reassigning" if is_reassign else "Assigning" layout.addWidget(QLabel( f"{action_word} reference {reference_no} " f"(customer {customer_id}) to a contract:" )) self.list_widget = QListWidget() layout.addWidget(self.list_widget) self.chk_all = QCheckBox("Show contracts from all customers") self.chk_all.toggled.connect(self._populate_list) layout.addWidget(self.chk_all) btn_box = QDialogButtonBox() self._btn_ok = btn_box.addButton("Assign", QDialogButtonBox.AcceptRole) btn_cancel = btn_box.addButton("Cancel", QDialogButtonBox.RejectRole) self._btn_ok.clicked.connect(self._on_ok) btn_cancel.clicked.connect(self.reject) if can_clear: btn_clear = btn_box.addButton("Clear Assignment", QDialogButtonBox.ResetRole) btn_clear.clicked.connect(self._on_clear) self.list_widget.itemDoubleClicked.connect(lambda _: self._on_ok()) layout.addWidget(btn_box) # Populate with same-customer contracts initially self._populate_list(show_all=False) def _populate_list(self, show_all: bool = False): self.list_widget.clear() if show_all: # Each entry: (customer_name, contract_id, contract_name) for cust_name, cid, cname in self._all_contracts: self.list_widget.addItem(f"{cust_name} / {cid} — {cname}") data = [(cid, cname) for _, cid, cname in self._all_contracts] else: for cid, cname in self._customer_contracts: self.list_widget.addItem(f"{cid} — {cname}") data = self._customer_contracts self._active_data = data # Pre-select current contract if visible preselect = 0 for i, (cid, _) in enumerate(data): if cid == self._current_contract_id: preselect = i break if data: self.list_widget.setCurrentRow(preselect) self._btn_ok.setEnabled(bool(data)) def _load_contracts(self, customer_id: str): try: conn = sqlite3.connect("data_source/mrr_data.db") df = pd.read_sql( "SELECT DISTINCT contract_id, contract_name FROM contracts " "WHERE customer_id = ? ORDER BY contract_name", conn, params=(customer_id,) ) conn.close() except Exception: return [] result = [] for _, r in df.iterrows(): cid = "" if pd.isna(r["contract_id"]) else str(r["contract_id"]).strip() cname = "" if pd.isna(r["contract_name"]) else str(r["contract_name"]).strip() if cid and cid.lower() not in ("nan", "none"): result.append((cid, cname or "(unnamed)")) return result def _load_all_contracts(self): try: conn = sqlite3.connect("data_source/mrr_data.db") df = pd.read_sql( "SELECT DISTINCT customer_name, contract_id, contract_name FROM contracts " "ORDER BY customer_name, contract_name", conn ) conn.close() except Exception: return [] result = [] for _, r in df.iterrows(): cid = "" if pd.isna(r["contract_id"]) else str(r["contract_id"]).strip() cname = "" if pd.isna(r["contract_name"]) else str(r["contract_name"]).strip() cust = "" if pd.isna(r["customer_name"]) else str(r["customer_name"]).strip() if cid and cid.lower() not in ("nan", "none"): result.append((cust or "(unknown)", cid, cname or "(unnamed)")) return result def _on_ok(self): row = self.list_widget.currentRow() if 0 <= row < len(self._active_data): self.selected_contract = self._active_data[row] self.accept() def _on_clear(self): self.clear_requested = True self.accept() # ============================================================ # CATEGORY OVERRIDE DIALOG # ============================================================ class CategoryDialog(QDialog): def __init__(self, current_code: int, is_manual: bool = False, parent=None): super().__init__(parent) self.selected_code = None self.clear_requested = False self.setWindowTitle("Set Category Override") self.setMinimumWidth(380) layout = QVBoxLayout(self) status = "manual override" if is_manual else "auto" layout.addWidget(QLabel(f"Current category: {current_code} ({status})")) self.list_widget = QListWidget() for code, label in ALL_CATEGORIES: self.list_widget.addItem(f"{code:>4} — {label}") if code == current_code: self.list_widget.setCurrentRow(self.list_widget.count() - 1) layout.addWidget(self.list_widget) btn_box = QDialogButtonBox() btn_ok = btn_box.addButton("OK", QDialogButtonBox.AcceptRole) btn_cancel = btn_box.addButton("Cancel", QDialogButtonBox.RejectRole) if is_manual: btn_clear = btn_box.addButton("Clear Override", QDialogButtonBox.ResetRole) btn_clear.clicked.connect(self._on_clear) btn_ok.clicked.connect(self._on_ok) btn_cancel.clicked.connect(self.reject) layout.addWidget(btn_box) def _on_ok(self): row = self.list_widget.currentRow() if row >= 0: self.selected_code = ALL_CATEGORIES[row][0] self.accept() def _on_clear(self): self.clear_requested = True self.accept() # ============================================================ # CHECKABLE MULTI-SELECT COMBO BOX (used by the per-tab item/account filters) # ============================================================ class CheckableComboBox(QComboBox): """ QComboBox with multi-select checkable items. A pinned top row labelled "All" toggles every other row at once. The line edit shows a compact summary of the current selection. The popup stays open while the user toggles individual items; clicking outside (or pressing Esc) closes it. """ selectionChanged = Signal() def __init__(self, parent=None): super().__init__(parent) self.setEditable(True) self.lineEdit().setReadOnly(True) self.lineEdit().installEventFilter(self) self._model = QStandardItemModel(self) self.setModel(self._model) self.view().viewport().installEventFilter(self) self._all_items: list[str] = [] # Suppress hidePopup() once after an item click so toggling stays open. self._block_close = False # ---- public API ------------------------------------------------------- def set_items(self, items, default_selected=None, all_label='All'): """ Populate the dropdown with *items*. *default_selected* is an optional iterable of item values that should start checked. If None, every item starts checked. Items not present in *default_selected* (when provided) start unchecked. """ self._all_items = [str(i) for i in items] default_set = None if default_selected is None else {str(x) for x in default_selected} self._model.clear() all_item = QStandardItem(all_label) all_item.setFlags(Qt.ItemIsEnabled) self._model.appendRow(all_item) for it in self._all_items: row = QStandardItem(it) row.setFlags(Qt.ItemIsEnabled) checked = (default_set is None) or (it in default_set) row.setData(Qt.Checked if checked else Qt.Unchecked, Qt.CheckStateRole) self._model.appendRow(row) # Sync the "All" row to reflect the aggregate of the per-item states. states = [self._model.item(r).checkState() for r in range(1, self._model.rowCount())] if not states or all(s == Qt.Checked for s in states): all_item.setCheckState(Qt.Checked) elif all(s == Qt.Unchecked for s in states): all_item.setCheckState(Qt.Unchecked) else: all_item.setCheckState(Qt.PartiallyChecked) self._refresh_text() def selected_items(self): m = self._model return [m.item(r).text() for r in range(1, m.rowCount()) if m.item(r).checkState() == Qt.Checked] def all_items(self): return list(self._all_items) # ---- internals -------------------------------------------------------- def eventFilter(self, obj, event): if obj is self.view().viewport(): if event.type() == QEvent.MouseButtonRelease: idx = self.view().indexAt(event.pos()) if idx.isValid(): self._toggle_row(idx.row()) self._block_close = True return True elif obj is self.lineEdit(): if event.type() == QEvent.MouseButtonPress: self.showPopup() return True return super().eventFilter(obj, event) def hidePopup(self): if self._block_close: self._block_close = False return super().hidePopup() def _toggle_row(self, row): m = self._model item = m.item(row) if item is None: return new_state = Qt.Unchecked if item.checkState() == Qt.Checked else Qt.Checked if row == 0: # "All" toggled — apply to every row for r in range(m.rowCount()): m.item(r).setCheckState(new_state) else: item.setCheckState(new_state) # Sync the "All" row to reflect the aggregate state. states = [m.item(r).checkState() for r in range(1, m.rowCount())] if all(s == Qt.Checked for s in states): m.item(0).setCheckState(Qt.Checked) elif all(s == Qt.Unchecked for s in states): m.item(0).setCheckState(Qt.Unchecked) else: m.item(0).setCheckState(Qt.PartiallyChecked) self._refresh_text() self.selectionChanged.emit() def _refresh_text(self): sel = self.selected_items() n_sel, n_all = len(sel), len(self._all_items) if n_all == 0: txt = '(none available)' elif n_sel == n_all: txt = f'All ({n_all})' elif n_sel == 0: txt = 'None selected' elif n_sel <= 3: txt = ', '.join(sel) else: txt = f'{n_sel} of {n_all} selected' self.lineEdit().setText(txt) # ============================================================ # MAIN WINDOW # ============================================================ class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("MRR Reconciliation Tool") self.resize(1500, 800) self.p_c = None self.p_l = None self.p_l_diff = None self.all_months = [] self.category_map = {} # (customer_id, contract_id) → {month → code} self._contracts_only_category_map = {} # contracts-only base map (no ledger augmentation) self._base_category_map = {} # programmatic map (pre user overrides) self.manual_override_keys = set() # {(customer_id, contract_id, month)} self._raw_df_l = None # raw ledger DataFrame (before assignments / pivoting) self.ledger_assignments = {} # {(reference_no, customer_id): (contract_id, contract_name)} central_widget = QWidget() self.setCentralWidget(central_widget) layout = QVBoxLayout(central_widget) # ── View navigation (real tabs along the top) ──────────────────── self.main_tabs = QTabBar() self.main_tabs.setDrawBase(False) self.main_tabs.setExpanding(False) for name in ("Contracts", "Ledger", "MRR Summary", "Backlog"): self.main_tabs.addTab(name) # ── Filter / action toolbar (below tabs, above content) ────────── ctrl_layout = QHBoxLayout() self.period_selector = QComboBox() self.from_selector = QComboBox() self.to_selector = QComboBox() self.lbl_status = QLabel("") self.lbl_status.setFixedWidth(25) self.lbl_status.setStyleSheet("color: red; font-weight: bold; font-size: 16px;") self.check_diff = QCheckBox("Show Ledger MoM Differences") self.check_diff.toggled.connect(self.update_views) self.check_highlight = QCheckBox("Highlight Discrepancies") self.check_highlight.setChecked(True) self.check_highlight.toggled.connect(self.update_views) self.btn_export = QPushButton("Export to Excel") self.btn_export.setStyleSheet("background-color: #2c7d32; color: white; font-weight: bold;") self.btn_export.clicked.connect(self.export_to_excel) ctrl_layout.addWidget(QLabel("Highlight Month:")) ctrl_layout.addWidget(self.period_selector) ctrl_layout.addSpacing(20) ctrl_layout.addWidget(QLabel("Range:")) ctrl_layout.addWidget(self.from_selector) ctrl_layout.addWidget(QLabel("-")) ctrl_layout.addWidget(self.to_selector) ctrl_layout.addSpacing(30) ctrl_layout.addWidget(self.check_diff) ctrl_layout.addWidget(self.check_highlight) ctrl_layout.addWidget(self.lbl_status) ctrl_layout.addStretch() ctrl_layout.addWidget(self.btn_export) ctrl_layout.addSpacing(10) layout.addWidget(self.main_tabs) layout.addLayout(ctrl_layout) for cb in [self.period_selector, self.from_selector, self.to_selector]: cb.currentTextChanged.connect(self.update_views) self.stack = QStackedWidget() # Contracts View: one QTreeView per item category, each with its own # item-filter dropdown above the tree. self.contracts_tabs = QTabWidget() self.contract_trees = {key: QTreeView() for key in CONTRACT_TAB_KEYS} self.contract_item_combos = {key: CheckableComboBox() for key in CONTRACT_TAB_KEYS} for key in CONTRACT_TAB_KEYS: self.contracts_tabs.addTab( self._build_filtered_tab_page(self.contract_trees[key], self.contract_item_combos[key], "Items:"), key, ) # Default to MS — it carries the bulk of recurring revenue and matches # the prior single-pane contracts view most closely. self.contracts_tabs.setCurrentIndex(CONTRACT_TAB_KEYS.index('MS')) # Ledger View mirrors the contracts pattern, with per-tab GL-account filters. self.ledger_tabs = QTabWidget() self.ledger_trees = {key: QTreeView() for key in LEDGER_TAB_KEYS} self.ledger_account_combos = {key: CheckableComboBox() for key in LEDGER_TAB_KEYS} for key in LEDGER_TAB_KEYS: tree = self.ledger_trees[key] # Belt-and-braces: prevent Qt's default inline editor from opening on # double-click. doubleClicked must fire cleanly for ref reassignment. tree.setEditTriggers(QAbstractItemView.NoEditTriggers) # Right-click anywhere in the ledger view to assign an unassigned ref. tree.setContextMenuPolicy(Qt.CustomContextMenu) self.ledger_tabs.addTab( self._build_filtered_tab_page(tree, self.ledger_account_combos[key], "Accounts:"), key, ) self.ledger_tabs.setCurrentIndex(LEDGER_TAB_KEYS.index('MS')) self.tree_summary = QTreeView() self.tree_backlog = QTreeView() self.tree_summary.setRootIsDecorated(False) # Backlog page wraps the tree with its own toolbar. self.rb_backlog_general = QRadioButton("General PS") self.rb_backlog_recurring = QRadioButton("Recurring PS (I904)") self.rb_backlog_general.setChecked(True) _backlog_ps_group = QButtonGroup(self) _backlog_ps_group.addButton(self.rb_backlog_general) _backlog_ps_group.addButton(self.rb_backlog_recurring) self.rb_backlog_general.toggled.connect(self._rebuild_backlog_tree) self.chk_backlog_outstanding = QCheckBox("Show only contracts with outstanding amounts") self.chk_backlog_outstanding.toggled.connect(self._rebuild_backlog_tree) backlog_toolbar = QHBoxLayout() backlog_toolbar.addWidget(self.rb_backlog_general) backlog_toolbar.addWidget(self.rb_backlog_recurring) backlog_toolbar.addSpacing(20) backlog_toolbar.addWidget(self.chk_backlog_outstanding) backlog_toolbar.addStretch() backlog_page = QWidget() backlog_layout = QVBoxLayout(backlog_page) backlog_layout.setContentsMargins(0, 0, 0, 0) backlog_layout.addLayout(backlog_toolbar) backlog_layout.addWidget(self.tree_backlog) contracts_page = QWidget() contracts_layout = QVBoxLayout(contracts_page) contracts_layout.setContentsMargins(0, 0, 0, 0) contracts_layout.addWidget(self.contracts_tabs) ledger_page = QWidget() ledger_layout = QVBoxLayout(ledger_page) ledger_layout.setContentsMargins(0, 0, 0, 0) ledger_layout.addWidget(self.ledger_tabs) self.stack.addWidget(contracts_page) self.stack.addWidget(ledger_page) self.stack.addWidget(self.tree_summary) self.stack.addWidget(backlog_page) layout.addWidget(self.stack) self.main_tabs.currentChanged.connect(self.stack.setCurrentIndex) # Start on Contracts to match prior default self.main_tabs.setCurrentIndex(0) self.stack.setCurrentIndex(0) for tree in self.ledger_trees.values(): tree.clicked.connect(self.on_category_clicked) tree.doubleClicked.connect(self.on_ledger_ref_double_clicked) tree.customContextMenuRequested.connect(self.on_ledger_context_menu) self.tree_summary.doubleClicked.connect(self.on_summary_double_clicked) self.initial_load() # ── Per-tab item / account filter dropdowns ──────────────────────── def _populate_filter_combos(self): """ Populate every dropdown with the full universe of codes present in the data, with each tab's default selection pre-checked. The user can then check items from any other category to include them in the tab. """ # Contracts: every distinct item in the data; tab default = its category list. if 'item' in self._raw_df_c.columns: items_all = sorted(self._raw_df_c['item'].astype(str).dropna().unique()) else: items_all = [] items_set = set(items_all) contract_defaults = { 'PS': items_set & set(_CONTRACT_TAB_PS_ITEMS), 'MS': items_set - set(_CONTRACT_TAB_MS_EXCLUDED_ITEMS), 'Cloud': items_set & set(_CONTRACT_TAB_CLOUD_ITEMS), 'Other Resell': items_set & set(_CONTRACT_TAB_OTHER_RESELL_ITEMS), } for key in CONTRACT_TAB_KEYS: self.contract_item_combos[key].set_items( items_all, default_selected=contract_defaults[key] ) # Ledger: every distinct base account in the data; tab default = its account list. if 'account' in self._df_l_for_pivot.columns: bases = self._df_l_for_pivot['account'].astype(str).str.split('-').str[0] accounts_all = sorted(bases.dropna().unique()) else: accounts_all = [] accounts_set = set(accounts_all) for key in LEDGER_TAB_KEYS: defaults = accounts_set & set(_LEDGER_TAB_ACCOUNTS[key]) self.ledger_account_combos[key].set_items( accounts_all, default_selected=defaults ) def _connect_filter_combos(self): for key in CONTRACT_TAB_KEYS: self.contract_item_combos[key].selectionChanged.connect( lambda k=key: self._on_contract_filter_changed(k) ) for key in LEDGER_TAB_KEYS: self.ledger_account_combos[key].selectionChanged.connect( lambda k=key: self._on_ledger_filter_changed(k) ) def _build_contract_tab_pivot(self, key): """Pivot the raw contracts data filtered to the items the tab's dropdown allows.""" selected = set(self.contract_item_combos[key].selected_items()) if not selected: return pd.DataFrame(columns=_CONTRACT_INDEX_COLS) sub = _preprocess_contracts_df( self._raw_df_c, item_filter=lambda items: items.isin(selected), ) if sub.empty: return pd.DataFrame(columns=_CONTRACT_INDEX_COLS) return pd.pivot_table( sub, values="amount", index=_CONTRACT_INDEX_COLS, columns="month", aggfunc="sum", fill_value=0, ).reset_index() def _build_ledger_tab_pivot(self, key): """Build the ledger pivot for *key* using the tab's selected accounts.""" selected = list(self.ledger_account_combos[key].selected_items()) if not selected: return pd.DataFrame() return build_ledger_pivot(self._df_l_for_pivot, selected) def _rebuild_all_contract_tab_pivots(self): for key in CONTRACT_TAB_KEYS: self._contract_tab_pivots[key] = self._build_contract_tab_pivot(key) def _rebuild_all_ledger_tab_pivots(self): for key in LEDGER_TAB_KEYS: pivot = self._build_ledger_tab_pivot(key) self._ledger_tab_pivots[key] = pivot self._ledger_tab_diffs[key] = calculate_mom_diff(pivot) def _on_contract_filter_changed(self, key): self._contract_tab_pivots[key] = self._build_contract_tab_pivot(key) self._refresh_contract_tab(key) def _on_ledger_filter_changed(self, key): pivot = self._build_ledger_tab_pivot(key) self._ledger_tab_pivots[key] = pivot self._ledger_tab_diffs[key] = calculate_mom_diff(pivot) self._refresh_ledger_tab(key) def _refresh_contract_tab(self, key): tree = self.contract_trees[key] visible_months = self.get_visible_range() highlight_m = self.period_selector.currentText() pivot = self._contract_tab_pivots.get(key, pd.DataFrame()) filtered = self.filter_active_rows(pivot, visible_months) exp = self._save_expanded(tree) tree.setModel( build_qt_model(filtered, highlight_m, visible_months, is_ledger=False, is_diff_view=False) ) tree.resizeColumnToContents(0) self._restore_expanded(tree, exp) def _refresh_ledger_tab(self, key): tree = self.ledger_trees[key] visible_months = self.get_visible_range() highlight_m = self.period_selector.currentText() use_diff = self.check_diff.isChecked() source = (self._ledger_tab_diffs if use_diff else self._ledger_tab_pivots).get(key, pd.DataFrame()) filtered = self.filter_active_rows(source, visible_months, is_diff=use_diff) # MS is the only tab participating in discrepancy + categorisation. is_ms = (key == 'MS') discrepancy_colors = (self._compute_discrepancy_colors(highlight_m) if is_ms and self.check_highlight.isChecked() else None) exp = self._save_expanded(tree) tree.setModel( build_qt_model(filtered, highlight_m, visible_months, is_ledger=True, is_diff_view=use_diff, reference_totals=discrepancy_colors, category_map=self.category_map if is_ms else None, manual_keys=self.manual_override_keys if is_ms else None, assigned_refs=set(self.ledger_assignments.keys())) ) tree.resizeColumnToContents(0) self._restore_expanded(tree, exp) def _compute_discrepancy_colors(self, highlight_m): """Per-customer contracts-vs-ledger discrepancy palette for the MS tab.""" cid_col = self.p_c.columns[0] lcid_col = self.p_l.columns[0] c_abs_data = (self.p_c.groupby(cid_col)[highlight_m].sum().to_dict() if highlight_m in self.p_c.columns else {}) l_abs_data = (self.p_l.groupby(lcid_col)[highlight_m].sum().to_dict() if highlight_m in self.p_l.columns else {}) colors = {} for c_id, c_val in c_abs_data.items(): l_val = l_abs_data.get(c_id, 0) if abs(c_val) > 0.01: diff_pct = abs(l_val - c_val) / abs(c_val) if diff_pct > 0.10: colors[c_id] = QColor(150, 100, 100) elif diff_pct > 0.05: colors[c_id] = QColor(155, 120, 50) elif abs(l_val) > 1.00: colors[c_id] = QColor(150, 100, 100) return colors # ── Tab-page construction helper ─────────────────────────────────── def _build_filtered_tab_page(self, tree, combo, label_text): """Wrap *tree* in a QWidget that puts *combo* (with *label_text*) above it.""" page = QWidget() layout = QVBoxLayout(page) layout.setContentsMargins(0, 0, 0, 0) toolbar = QHBoxLayout() toolbar.addWidget(QLabel(label_text)) combo.setMinimumWidth(220) toolbar.addWidget(combo) toolbar.addStretch() layout.addLayout(toolbar) layout.addWidget(tree) return page # Only the columns the data pipeline actually touches. Narrowing the SELECT # dramatically reduces the SQLite → pandas read cost on a ~100MB DB. _LEDGER_SQL = ( 'SELECT account, amount, customer_id, customer_name, ' 'contract_id, contract_name, posting_date, "reference_no." ' 'FROM ledger' ) _CONTRACTS_SQL = ( 'SELECT customer_id, customer_name, contract_id, contract_name, ' 'line_number, item, item_description, scheduled_posting_date, ' 'amount, status, mrr_category ' 'FROM contracts' ) _BILLING_SQL = ( 'SELECT customer_id, customer_name, contract_id, contract_name, ' 'item, scheduled_billing_date, amount, billed, status ' 'FROM contracts_billing' ) def initial_load(self): _t_il = time.perf_counter() try: conn = init_database() df_l = pd.read_sql(self._LEDGER_SQL, conn) df_c = pd.read_sql(self._CONTRACTS_SQL, conn) try: df_b = pd.read_sql(self._BILLING_SQL, conn) except Exception as e: print(f"--- [WARN] Could not read contracts_billing table: {e} ---") df_b = pd.DataFrame() conn.close() # Keep raw ledger so _rebuild_ledger can re-apply assignments from a clean base. # Keep raw contracts so the per-tab item-filter dropdowns can re-pivot subsets. self._raw_df_l = df_l.copy() self._raw_df_c = df_c.copy() # Load any previously saved manual ledger→contract assignments and patch the ledger # BEFORE pivoting, so the row groups under its assigned contract from the start. ensure_ledger_assignments_table() self.ledger_assignments = load_ledger_assignments() self._df_l_for_pivot = apply_ledger_assignments(df_l.copy(), self.ledger_assignments) print(f"--- [CHECK] Applied {len(self.ledger_assignments)} ledger assignment(s). ---") # Canonical MS pivots (built from the full MS account list, ignoring the # MS-tab dropdown filter) — these drive category map, discrepancy, MRR # summary, and Excel export. The per-tab pivots used for display are # built separately below and follow the dropdown selections. self.p_l = build_ledger_pivot(self._df_l_for_pivot, _LEDGER_TAB_MS_ACCOUNTS) self.p_l_diff = calculate_mom_diff(self.p_l) self.p_c, detail_df = build_contract_outputs(df_c) # Populate the per-tab item/account filter dropdowns with whatever # codes actually appear in the data for each tab, then build the # initial filtered pivots (everything checked → matches full tab pivot). self._populate_filter_combos() self._connect_filter_combos() self._contract_tab_pivots = {} self._ledger_tab_pivots, self._ledger_tab_diffs = {}, {} self._rebuild_all_contract_tab_pivots() self._rebuild_all_ledger_tab_pivots() self._backlog_df_general, self._backlog_has_remaining_general = build_backlog_data(df_b, _PS_ITEMS_GENERAL) self._backlog_df_recurring, self._backlog_has_remaining_recurring = build_backlog_data(df_b, _PS_ITEMS_RECURRING) self.tree_backlog.setModel(build_backlog_model( self._backlog_df_general, self._backlog_has_remaining_general )) for col in range(4): self.tree_backlog.resizeColumnToContents(col) l_months = [c for c in self.p_l.columns if "-" in str(c)] c_months = [c for c in self.p_c.columns if "-" in str(c)] self.all_months = sorted(list(set(l_months) | set(c_months))) # Contracts-only category map (cached so _rebuild_ledger can re-augment from a clean base) self._contracts_only_category_map = build_contract_category_map(detail_df, self.all_months) self.category_map = augment_category_map_from_ledger( copy.deepcopy(self._contracts_only_category_map), self.p_l_diff, self.all_months ) self._base_category_map = copy.deepcopy(self.category_map) print(f"--- [CHECK] Category map built for {len(self.category_map)} (customer, contract) pairs. ---") # Apply saved user overrides on top of the programmatic map ensure_category_overrides_table() self.manual_override_keys = set() for key, month_codes in load_category_overrides().items(): self.category_map.setdefault(key, {}) for month, code in month_codes.items(): self.category_map[key][month] = code self.manual_override_keys.add((key[0], key[1], month)) print(f"--- [CHECK] Loaded {len(self.manual_override_keys)} manual category override(s). ---") for cb in [self.period_selector, self.from_selector, self.to_selector]: cb.blockSignals(True) cb.addItems(self.all_months) now = datetime.now() current_m = now.strftime("%Y-%m") year_start = now.strftime("%Y-01") year_end = now.strftime("%Y-12") if current_m in self.all_months: self.period_selector.setCurrentText(current_m) elif self.all_months: self.period_selector.setCurrentIndex(len(self.all_months) - 1) if year_start in self.all_months: self.from_selector.setCurrentText(year_start) else: self.from_selector.setCurrentIndex(0) if year_end in self.all_months: self.to_selector.setCurrentText(year_end) elif self.all_months: self.to_selector.setCurrentIndex(len(self.all_months) - 1) for cb in [self.period_selector, self.from_selector, self.to_selector]: cb.blockSignals(False) _t_data = time.perf_counter() print(f"[timing] initial_load (data pipeline only): {_ms(_t_il):7.1f} ms") self.update_views() print(f"[timing] update_views (Qt model building): {_ms(_t_data):7.1f} ms") print(f"[timing] initial_load total: {_ms(_t_il):7.1f} ms") except Exception as e: import traceback traceback.print_exc() print(f"Data Load Error: {e}") def get_visible_range(self): start_m = self.from_selector.currentText() end_m = self.to_selector.currentText() try: idx_s = self.all_months.index(start_m) idx_e = self.all_months.index(end_m) if idx_s > idx_e: idx_s, idx_e = idx_e, idx_s return self.all_months[idx_s : idx_e + 1] except ValueError: return self.all_months def filter_active_rows(self, df, requested_months, is_diff=False): if df is None or df.empty: return df existing_months = [m for m in requested_months if m in df.columns] if not existing_months: return df.iloc[0:0] meta_cols = [c for c in df.columns if "-" not in str(c)] if is_diff: # In diff view, keep rows that have at least one month with meaningful movement. # Using sum() would incorrectly drop contracts where e.g. +500 in Jan and -500 in Feb # cancel out — those rows have real changes and must be shown. mask = df[existing_months].abs().max(axis=1) > 0.01 else: mask = df[existing_months].abs().sum(axis=1) > 0.01 return df[mask][meta_cols + existing_months] # ── Expanded-state helpers ────────────────────────────────────────── def _save_expanded(self, tree) -> set: """Return a set of text-path tuples for every expanded node (depth ≤ 2).""" model = tree.model() if model is None: return set() expanded = set() for r0 in range(model.rowCount()): i0 = model.index(r0, 0) if not tree.isExpanded(i0): continue t0 = model.data(i0) or "" expanded.add((t0,)) for r1 in range(model.rowCount(i0)): i1 = model.index(r1, 0, i0) if tree.isExpanded(i1): expanded.add((t0, model.data(i1) or "")) return expanded def _restore_expanded(self, tree, expanded: set): """Re-expand nodes whose text-path appears in *expanded*.""" model = tree.model() if model is None or not expanded: return for r0 in range(model.rowCount()): i0 = model.index(r0, 0) t0 = model.data(i0) or "" if (t0,) not in expanded: continue tree.expand(i0) for r1 in range(model.rowCount(i0)): i1 = model.index(r1, 0, i0) if (t0, model.data(i1) or "") in expanded: tree.expand(i1) # ── View refresh ─────────────────────────────────────────────────── def update_views(self): visible_months = self.get_visible_range() # Range reversal indicator idx_s = self.all_months.index(self.from_selector.currentText()) idx_e = self.all_months.index(self.to_selector.currentText()) self.lbl_status.setText("●" if idx_s > idx_e else "") for key in CONTRACT_TAB_KEYS: self._refresh_contract_tab(key) for key in LEDGER_TAB_KEYS: self._refresh_ledger_tab(key) self.tree_summary.setModel( build_category_summary_model(self.category_map, self.p_l_diff, visible_months) ) self.tree_summary.resizeColumnToContents(0) def on_category_clicked(self, index): if not index.isValid() or not index.data(ROLE_IS_CATEGORY_CELL): return customer_id = index.data(ROLE_CUSTOMER_ID) contract_id = index.data(ROLE_CONTRACT_ID) month = index.data(ROLE_MONTH) current_code = self.category_map.get((customer_id, contract_id), {}).get(month, 0) is_manual = (customer_id, contract_id, month) in self.manual_override_keys dialog = CategoryDialog(current_code, is_manual=is_manual, parent=self) if dialog.exec() != QDialog.Accepted: return if dialog.clear_requested: delete_category_override(customer_id, contract_id, month) self.manual_override_keys.discard((customer_id, contract_id, month)) base_code = self._base_category_map.get((customer_id, contract_id), {}).get(month, 0) self.category_map.setdefault((customer_id, contract_id), {})[month] = base_code elif dialog.selected_code is not None: save_category_override(customer_id, contract_id, month, dialog.selected_code) self.category_map.setdefault((customer_id, contract_id), {})[month] = dialog.selected_code self.manual_override_keys.add((customer_id, contract_id, month)) self.update_views() # ── Manual ledger→contract assignment ────────────────────────────── def _get_assignable_index(self, index): """Return an index bearing assignment roles if the row is assignable/reassignable, else None.""" if not index.isValid(): return None def _valid(idx): if idx.isValid() and idx.data(ROLE_IS_UNASSIGNED_REF): cust = idx.data(ROLE_CUSTOMER_ID) if cust and cust not in MISSING_SENTINELS: return idx return None valid = _valid(index) if valid is not None: return valid # Fallback: column 0 of the same row. With the custom model, the roles are # returned identically across columns of a line row, but this guards against # any future divergence. return _valid(index.sibling(index.row(), 0)) def on_ledger_ref_double_clicked(self, index): idx = self._get_assignable_index(index) if idx is None: return self._open_assign_dialog( idx.data(ROLE_CUSTOMER_ID), idx.data(ROLE_REF_NO), idx.data(ROLE_ASSIGNED_CONTRACT_ID), bool(idx.data(ROLE_CAN_CLEAR)), ) def on_ledger_context_menu(self, pos): # Signal fires on whichever ledger tab tree the user right-clicked. tree = self.sender() if not isinstance(tree, QTreeView): return index = tree.indexAt(pos) idx = self._get_assignable_index(index) if idx is None: return customer_id = idx.data(ROLE_CUSTOMER_ID) ref_no = idx.data(ROLE_REF_NO) current_contract_id = idx.data(ROLE_ASSIGNED_CONTRACT_ID) can_clear = bool(idx.data(ROLE_CAN_CLEAR)) menu = QMenu(self) label = f"Change contract for \"{ref_no}\"…" if current_contract_id else f"Assign \"{ref_no}\" to a contract…" action = menu.addAction(label) if menu.exec(tree.viewport().mapToGlobal(pos)) == action: self._open_assign_dialog(customer_id, ref_no, current_contract_id, can_clear) def _open_assign_dialog(self, customer_id: str, ref_no: str, current_contract_id: str = None, can_clear: bool = False): dialog = AssignContractDialog(customer_id, ref_no, current_contract_id=current_contract_id, can_clear=can_clear, parent=self) if dialog.exec() != QDialog.Accepted: return if dialog.clear_requested: delete_ledger_assignment(ref_no, customer_id) self.ledger_assignments.pop((str(ref_no), str(customer_id)), None) self._rebuild_ledger() elif dialog.selected_contract is not None: contract_id, contract_name = dialog.selected_contract save_ledger_assignment(ref_no, customer_id, contract_id, contract_name) self.ledger_assignments[(str(ref_no), str(customer_id))] = (contract_id, contract_name) self._rebuild_ledger() def _rebuild_ledger(self): """Re-pivot the ledger after a new assignment and cascade through the category map.""" QApplication.setOverrideCursor(Qt.WaitCursor) try: self._rebuild_ledger_inner() finally: QApplication.restoreOverrideCursor() def _rebuild_ledger_inner(self): self._df_l_for_pivot = apply_ledger_assignments(self._raw_df_l.copy(), self.ledger_assignments) # Canonical MS pivot — full account list, independent of the dropdown filter. self.p_l = build_ledger_pivot(self._df_l_for_pivot, _LEDGER_TAB_MS_ACCOUNTS) self.p_l_diff = calculate_mom_diff(self.p_l) # Per-tab display pivots follow the current dropdown selections. self._rebuild_all_ledger_tab_pivots() # Re-augment category map from updated ledger; contracts-only base stays unchanged. self.category_map = augment_category_map_from_ledger( copy.deepcopy(self._contracts_only_category_map), self.p_l_diff, self.all_months ) self._base_category_map = copy.deepcopy(self.category_map) # Re-apply user category overrides (these are independent of ledger assignments). for key, month_codes in load_category_overrides().items(): self.category_map.setdefault(key, {}) for month, code in month_codes.items(): self.category_map[key][month] = code self.update_views() def _rebuild_backlog_tree(self, _=False): """Rebuild the backlog tree based on the current PS-type toggle and outstanding filter.""" if not hasattr(self, "_backlog_df_general"): return if self.rb_backlog_recurring.isChecked(): df, remaining = self._backlog_df_recurring, self._backlog_has_remaining_recurring else: df, remaining = self._backlog_df_general, self._backlog_has_remaining_general self.tree_backlog.setModel(build_backlog_model( df, remaining, outstanding_only=self.chk_backlog_outstanding.isChecked(), )) for col in range(4): self.tree_backlog.resizeColumnToContents(col) def on_summary_double_clicked(self, index): if not index.isValid(): return code = index.data(ROLE_CATEGORY_CODE) month = index.data(ROLE_MONTH) if code is None or month is None: return visible_months = self.get_visible_range() DrillDownDialog(code, month, self.category_map, self.p_l_diff, visible_months, self._df_l_for_pivot, parent=self).exec() def _write_grouped_sheet(self, writer, sheet_name, df, summary_fn=None): """ Write *df* to *sheet_name* with three-level collapsible row groups. Input columns: customer_id, customer_name, contract_id, contract_name, , value_col1, value_col2, ... Output columns: customer_name | contract_name | detail | value_cols... ID columns are dropped. Each name only appears on its own summary row: customer_name on level-0 rows, contract_name on level-1 rows, detail on level-2 leaf rows. Contracts and detail rows are hidden by default so the sheet opens fully collapsed. """ from openpyxl.worksheet.properties import Outline from openpyxl.styles import Font if df is None or df.empty: (df if df is not None else pd.DataFrame()).to_excel( writer, sheet_name=sheet_name, index=False ) return cols = list(df.columns) cust_id_col, cust_name_col, ct_id_col, ct_name_col = cols[:4] detail_col = cols[4] value_cols = cols[5:] out_cols = [cust_name_col, ct_name_col, detail_col] + value_cols df_sorted = df.sort_values([cust_name_col, ct_name_col], kind="stable") def _default_summary(group, vcols): return {v: group[v].sum() for v in vcols} agg = summary_fn or _default_summary out_rows = [] levels = [] for (cid, cname), c_grp in df_sorted.groupby( [cust_id_col, cust_name_col], sort=False ): # Customer summary row — only customer_name populated row = {c: "" for c in out_cols} row[cust_name_col] = cname row.update(agg(c_grp, value_cols)) out_rows.append(row) levels.append(0) for (ctid, ctname), ct_grp in c_grp.groupby( [ct_id_col, ct_name_col], sort=False ): # Contract summary row — only contract_name populated row = {c: "" for c in out_cols} row[ct_name_col] = ctname row.update(agg(ct_grp, value_cols)) out_rows.append(row) levels.append(1) # Detail rows — only detail_col and value_cols populated for _, r in ct_grp.iterrows(): row = {c: "" for c in out_cols} row[detail_col] = r[detail_col] for v in value_cols: row[v] = r[v] out_rows.append(row) levels.append(2) pd.DataFrame(out_rows, columns=out_cols).to_excel( writer, sheet_name=sheet_name, index=False ) ws = writer.sheets[sheet_name] # summaryBelow=False puts the parent row above its detail rows, matching # the in-app tree view where a customer/contract sits above its children. ws.sheet_properties.outlinePr = Outline(summaryBelow=False) bold_font = Font(bold=True) for i, lvl in enumerate(levels): excel_row = i + 2 if lvl > 0: rd = ws.row_dimensions[excel_row] rd.outline_level = lvl rd.hidden = True if lvl == 0: for cell in ws[excel_row]: cell.font = bold_font def export_to_excel(self): visible_months = self.get_visible_range() filename = f"MRR_Export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" try: with pd.ExcelWriter(filename, engine='openpyxl') as writer: # 1. Contracts View df_exp_c = self.filter_active_rows(self.p_c, visible_months) self._write_grouped_sheet(writer, "Contracts", df_exp_c) # 2. Ledger View (Full) df_exp_l = self.filter_active_rows(self.p_l, visible_months) self._write_grouped_sheet(writer, "Ledger_Actuals", df_exp_l) # 3. Ledger MoM Differences df_exp_diff = self.filter_active_rows(self.p_l_diff, visible_months) self._write_grouped_sheet(writer, "Ledger_MoM_Changes", df_exp_diff) # 4. MRR Summary (category × month) cid_col = self.p_l_diff.columns[0] ctid_col = self.p_l_diff.columns[2] month_cols = [m for m in visible_months if m in self.p_l_diff.columns] contract_totals = self.p_l_diff.groupby([cid_col, ctid_col])[month_cols].sum() cat_amounts = _aggregate_summary_by_category( self.category_map, contract_totals, month_cols ) summary_rows = [] totals: dict = {} for code, label in ALL_CATEGORIES: if code not in cat_amounts: continue r = {"Category": f"{code} — {label}"} for m in visible_months: r[m] = cat_amounts[code].get(m, 0.0) totals[m] = totals.get(m, 0.0) + r[m] summary_rows.append(r) if summary_rows: total_row = {"Category": "Net MRR Change"} total_row.update(totals) summary_rows.append(total_row) pd.DataFrame(summary_rows).to_excel(writer, sheet_name="MRR_Summary", index=False) # 5. Backlog reports — General PS and Recurring PS (I904). def _backlog_summary(group, _vcols): sched = float(group["scheduled_amount"].sum()) billed = float(group["billed_amount"].sum()) return { "scheduled_amount": sched, "billed_amount": billed, "remaining_balance": sched - billed, } for sheet_name, bl_df in ( ("Backlog_General_PS", getattr(self, "_backlog_df_general", None)), ("Backlog_Recurring_PS", getattr(self, "_backlog_df_recurring", None)), ): if bl_df is None or bl_df.empty: continue bl_sorted = bl_df.sort_values( ["customer_name", "contract_name", "month"], kind="stable" ).copy() bl_sorted["remaining_balance"] = ( bl_sorted.groupby(["customer_id", "contract_id"], sort=False)["amount"].transform("sum") - bl_sorted.groupby(["customer_id", "contract_id"], sort=False)["billed"].cumsum() ) bl_sorted = bl_sorted.rename(columns={ "amount": "scheduled_amount", "billed": "billed_amount", })[[ "customer_id", "customer_name", "contract_id", "contract_name", "month", "scheduled_amount", "billed_amount", "remaining_balance", ]] self._write_grouped_sheet(writer, sheet_name, bl_sorted, summary_fn=_backlog_summary) QMessageBox.information(self, "Export Successful", f"Data exported successfully to:\n{os.path.abspath(filename)}") except Exception as e: QMessageBox.critical(self, "Export Failed", f"An error occurred while exporting:\n{str(e)}") if __name__ == "__main__": _t = time.perf_counter() app = QApplication(sys.argv) print(f"[timing] QApplication(sys.argv): {_ms(_t):7.1f} ms") # Increase application-wide font size app_font = QApplication.font() app_font.setPointSize(app_font.pointSize() + _FONT_SIZE_OFFSET) QApplication.setFont(app_font) _t = time.perf_counter() window = MainWindow() print(f"[timing] MainWindow() (incl. initial_load): {_ms(_t):7.1f} ms") _t = time.perf_counter() window.show() print(f"[timing] window.show(): {_ms(_t):7.1f} ms") def _first_event_tick(): print(f"[timing] first event-loop tick (≈ first paint): {_ms(_T0):7.1f} ms TOTAL since process start") QTimer.singleShot(0, _first_event_tick) sys.exit(app.exec())