"""pubsalesacc/pipeline.py — Core business logic for the Publishing Sales Accounting Run.

Each public function corresponds to one step in the quarterly workflow and maps
to a numbered menu item in pub.py.
"""

import logging
import os
import time
from pathlib import Path
from typing import Optional

import numpy as np
import pandas as pd

import config
import pubsalesacc.connectors.snowflake as sf
import pubsalesacc.connectors.neo4j as n4
import pubsalesacc.connectors.jira as ja
import pubsalesacc.utils as util

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# SQL statements — loaded once at module level (no side-effects, just file I/O)
# ---------------------------------------------------------------------------

_SQL_FILE = Path(__file__).parent.parent / "sql" / "accountingrun.sql"
_CYPHER_FILE = Path(__file__).parent.parent / "sql" / "SWAlookupV3.cypher"
_DBPR_TEMPLATE_FILE = Path(__file__).parent.parent / "sql" / "n4j-dbpr-template.txt"

sql: list[str] = util.read_sql_file(_SQL_FILE)

# ---------------------------------------------------------------------------
# Sales file columns — fixed 31-column Sony Publishing schema
# ---------------------------------------------------------------------------

_SALES_COLUMNS = [
    "Song No.",
    "Song",
    "Writer",
    "Src Nm",
    "Src Ctry",
    "Src2 Nm",
    "Src2Ctry",
    "Src3 Nm",
    "Src3Ctry",
    "Src4 Nm",
    "Src4Ctry",
    "Inc Typ",
    "SH ID",
    "Rpt'g Pd",
    "Sales Pd",
    "ProdNo.",
    "Art Prd#",
    "Units",
    "SngShr%",
    "Cntrl %",
    "Amount",
    "Roy %",
    "RoyAmt",
    "D/F",
    "Src Prod",
    "ISWC Cd",
    "ExtSong",
    "Artist",
    "SrcSong",
    "ISRC",
    "SalesTyp",
]

_FLOAT_COLS = ["Units", "SngShr%", "Cntrl %", "Amount", "Roy %", "RoyAmt"]

# ---------------------------------------------------------------------------
# Step 0 — Read sales file
# ---------------------------------------------------------------------------


def read_file_to_df(
    file: str,
    drop_head_rows: bool = False,
    drop_header: bool = True,
) -> Optional[pd.DataFrame]:
    """Load the Sony Publishing sales file (.xlsx or .csv) into a DataFrame.

    If an XLSX file fails to load due to the known Sony font-family XML issue,
    the function automatically applies the Excel fix and retries once.

    Args:
        file:            Path to the sales file.
        drop_head_rows:  If True, drop the first 9 rows (pre-header content).
        drop_header:     If True, drop the first row after any heading rows (column header row).

    Returns:
        DataFrame with 31 columns and numeric types applied, or None on failure.
    """

    def _load(filepath: str) -> pd.DataFrame:
        if filepath.endswith("xlsx"):
            dffile = pd.read_excel(
                filepath,
                sheet_name=None,
                header=None,
                names=_SALES_COLUMNS,
                dtype=str,
            )
            df = pd.concat(dffile.values(), ignore_index=True)
        elif filepath.endswith("csv"):
            df = pd.read_csv(
                filepath,
                header=None,
                names=_SALES_COLUMNS,
                dtype=str,
            )
        else:
            raise ValueError(
                f"Unsupported file type: {filepath}. Expected .xlsx or .csv"
            )
        return df

    try:
        df = _load(file)
    except Exception as exc:
        if file.endswith("xlsx"):
            logger.warning(
                "XLSX load failed (%s). Attempting automatic Excel fix and retry...",
                exc,
            )
            try:
                util.fix_sony_excel(file)
                df = _load(file)
                logger.info("XLSX loaded successfully after fix.")
            except Exception as retry_exc:
                logger.error("Excel fix retry failed: %s", retry_exc)
                logger.error(
                    "Manual fix may be needed: set all cells to a standard font "
                    "(e.g. Calibri) in Excel and save, then retry."
                )
                return None
        else:
            logger.error("Failed to load file: %s", exc)
            return None

    if drop_head_rows:
        df.drop(df.index[0:9], inplace=True)
    if drop_header:
        df.drop(df.index[0], inplace=True)

    df.fillna("", inplace=True)
    df.replace("nan", "", inplace=True)
    df[_FLOAT_COLS] = df[_FLOAT_COLS].apply(pd.to_numeric, errors="coerce")

    # Load log
    log = util.Textobj()
    log.add(f"{file} load log")
    log.add(f"Shape: {df.shape}")
    log.add(f"Columns: {list(df.columns)}")
    log.add(f"Royalty % distribution:\n{df['Roy %'].value_counts()}")
    log.add(
        f"RoyAmt sum for rows missing Units: {df[df['Units'].isna()]['RoyAmt'].sum()}"
    )
    bad = df.isna().sum()
    bad = bad[bad > 0]
    if not bad.empty:
        log.add(f"Columns with missing values:\n{bad}")
    log_name = "".join(Path(file).stem) + "_Load_Log"
    log.save(log_name)

    logger.info("Loaded %d rows from %s", len(df), file)
    return df


# ---------------------------------------------------------------------------
# Step 1 — Royalty output
# ---------------------------------------------------------------------------


def _royalty_sum_count(
    df: pd.DataFrame,
    roy_pct: list[float] | float = -90.0,
    roy_amt_sign: int = 0,
) -> tuple[pd.DataFrame, int, float]:
    """Filter DataFrame by royalty % and amount sign, returning filtered df, count, sum.

    Args:
        roy_pct:       List of royalty percentages to include. Negative values
                       indicate NOT IN (e.g. -90 means exclude 90%).
                       Use -101 (or other impossible value) to match all.
        roy_amt_sign:  1 = positive amounts only, -1 = negative only, 0 = all.
    """
    amt_col = "RoyAmt"
    pct_col = "Roy %"

    if not isinstance(roy_pct, list):
        roy_pct = [roy_pct]

    negate = any(x < 0 for x in roy_pct)
    pct_abs = [abs(x) for x in roy_pct]

    # Amount filter
    mask = df[amt_col].notna()
    if roy_amt_sign == 1:
        mask &= df[amt_col] >= 0
    elif roy_amt_sign == -1:
        mask &= df[amt_col] < 0

    # Royalty % filter
    pct_mask = df[pct_col].isin(pct_abs)
    mask &= ~pct_mask if negate else pct_mask

    filtered = df[mask]
    return filtered, int(filtered[amt_col].count()), float(filtered[amt_col].sum())


def royalty_output(
    df: pd.DataFrame,
    fileseries: str,
    period: int,
    reporting_month: dict,
    standard_roy_pct: list[float] = None,
) -> None:
    """Step 1 — Summarise royalties and produce output for Finance and Publishing teams.

    Generates:
    - {fileseries}-NonStandard-NegativeRoyalties.xlsx  (one sheet per condition)
    - {fileseries}-NonStandard-NegativeRoyalties.txt   (Slack/email message draft)

    Args:
        df:               Loaded sales DataFrame.
        fileseries:       Output filename prefix.
        period:           Snowflake period ID.
        reporting_month:  Dict from get_reporting_month_object().
        standard_roy_pct: List of standard royalty percentages. Default [90.0].
    """
    if standard_roy_pct is None:
        standard_roy_pct = [90.0]

    txt = util.Textobj()
    txt.add(
        f"@here\n"
        f"Welcome to another Quarterly Publishing Sales Run.\n"
        f"Period {period}.\n"
        f"Total line count in the sales file: {len(df):,}"
    )

    total_royalty = util.currency(df["RoyAmt"].sum())
    total_amount = util.currency(df["Amount"].sum())
    txt.add(f"1. Total Royalty Amount (column W):\n {total_royalty}")
    txt.add(f"2. Total Amount (column U):\n {total_amount}")

    # Build condition set: one entry per non-standard %, plus negatives
    conditions: dict = {}
    non_std_pcts = [
        float(p)
        for p in df["Roy %"].dropna().unique()
        if float(p) not in standard_roy_pct
    ]
    for pct in non_std_pcts:
        conditions[f"nonstandard_{pct}"] = {
            "desc": f"Non-standard ({pct}%) row count and total amount",
            "pct": pct,
            "sign": 0,
        }

    std_label = ", ".join(f"{x}%" for x in standard_roy_pct)
    conditions["negative"] = {
        "desc": "Negative row count and total amount",
        "pct": -101,
        "sign": -1,
    }
    conditions["nonstandard_and_negative"] = {
        "desc": f"Non-standard (not {std_label}) and negative row count and total amount",
        "pct": [-p for p in standard_roy_pct],
        "sign": -1,
    }

    out_xlsx = f"{fileseries}-NonStandard-NegativeRoyalties.xlsx"
    MAX_WS_ROWS = 1_048_575

    with pd.ExcelWriter(out_xlsx) as writer:
        for item_num, (ws_name, cond) in enumerate(conditions.items(), start=3):
            wsdf, count, amount = _royalty_sum_count(df, cond["pct"], cond["sign"])
            txt.add(f"{item_num}. {cond['desc']}:\n {count}, {util.currency(amount)}")

            if len(wsdf) > MAX_WS_ROWS:
                for i in range(0, len(wsdf), MAX_WS_ROWS):
                    wsdf[i : i + MAX_WS_ROWS].to_excel(
                        writer, sheet_name=f"{ws_name}_{i}", index=False
                    )
            else:
                wsdf.to_excel(writer, sheet_name=ws_name[:31], index=False)

    txt.print()

    # Negative summary for Finance email draft
    _, neg_count, neg_amount = _royalty_sum_count(df, -101, -1)
    neg_amount_fmt = util.currency(neg_amount)
    month = reporting_month["Month"]
    year = reporting_month["YYYY"]

    txt.add(
        "\n@Matthew Galizia — please advise if the values balance with you? "
        "If so, I'll report them to the finance folks."
    )
    txt.add(
        f"**** SUBJECT: {month} {year} Publishing Sales Run ({period})\n\n"
        f"We are prepping the publishing royalty for run {period}. "
        f"The amount to be processed is:\n\n"
        f"{total_royalty}\n\n"
        f"@Mey Tseng — can you confirm payment?\n\n"
        f"There are {neg_count} negative amount lines totalling {neg_amount_fmt}.\n\n"
        f"All amounts above are sums of column W in the sales report.\n\n"
        f"Recipients:\n" + "\n".join(config.FINANCE_EMAIL_RECIPIENTS)
    )

    txt.save(f"{fileseries}-NonStandard-NegativeRoyalties")
    logger.info("Royalty output complete.")


# ---------------------------------------------------------------------------
# Step 2 — Song ID reconciliation
# ---------------------------------------------------------------------------


def reconcile_song_ids(
    df: pd.DataFrame,
    lookup_master: str,
    fileseries: str,
    period: int,
    reporting_month: dict,
    excel_output: bool = False,
    csv_output: bool = True,
) -> pd.DataFrame:
    """Step 2 — Reconcile ExtSong IDs against Snowflake and the master lookup.

    - Applies known bad-ID → correct-ID mappings from the master lookup.
    - Applies title+artist → correct-ID mappings for blank ExtSong rows.
    - Drops rows flagged for exclusion (blank 'Correct' in master lookup).
    - Outputs a missing songs file if any IDs remain unreconciled.
    - Saves a reconciled pickle + optional XLSX/CSV if fully reconciled.
    - Always saves a run validation log.

    Returns the reconciled DataFrame.
    """
    start_len = len(df)
    start_amt = df["Amount"].sum()
    start_roy = df["RoyAmt"].sum()

    # Load master lookup
    lookup = pd.read_excel(lookup_master, dtype=str)
    lookup.fillna("", inplace=True)
    lookup["Correct"] = lookup["Correct"].str.strip()

    # Load valid pub song IDs from Snowflake
    sf_songs = sf.sf_df(sql[0])
    valid_ids = sf_songs["PUB_SONG_ID"].astype(str).str.strip().tolist()

    # --- Map 1: bad SMP ExtSong → correct ID ---
    bad_map = lookup[lookup["From Sony Music Publishing"] != ""][
        ["From Sony Music Publishing", "Correct"]
    ].drop_duplicates()
    df["ExtSong"] = df["ExtSong"].replace(
        bad_map["From Sony Music Publishing"].tolist(),
        bad_map["Correct"].tolist(),
    )

    # --- Map 2: blank ExtSong matched by Title + Artist ---
    title_map = lookup[lookup["From Sony Music Publishing"] == ""][
        ["Title", "Artist", "Correct"]
    ].drop_duplicates()

    total = len(title_map)
    for idx, (_, row) in enumerate(title_map.iterrows(), start=1):
        matched = (
            (df["Artist"] == row["Artist"])
            & (df["Song"] == row["Title"])
            & (df["ExtSong"] == "")
        )
        n = matched.sum()
        logger.info(
            "[%d/%d] Matched %d rows for artist '%s' / song '%s'",
            idx,
            total,
            n,
            row["Artist"],
            row["Title"],
        )
        df.loc[matched, "ExtSong"] = row["Correct"]

    # --- Exclusions: rows whose Song title has no Correct value in lookup ---
    drop_titles = lookup[lookup["Correct"] == ""]["Title"].tolist()
    drop_df = df[df["Song"].isin(drop_titles)]

    if not drop_df.empty:
        logger.warning(
            "%d rows dropped from processing based on '%s'", len(drop_df), lookup_master
        )
        drop_df[["Song No.", "Song"]].value_counts()
        drop_df.to_excel(f"{fileseries}-LinesDroppedFromProcessing.xlsx", index=False)
        df = df[~df["Song"].isin(drop_titles)]

    # --- Check what's still unreconciled ---
    unreconciled = df[~df["ExtSong"].isin(valid_ids)].copy()
    unreconciled.fillna("", inplace=True)
    unreconciled.replace("nan", "", inplace=True)

    missing_songs = (
        unreconciled[["ExtSong", "Song", "Artist", "Writer"]]
        .drop_duplicates()
        .fillna("")
        .replace("nan", "")
    )

    # Bug fix: was > 1 (off-by-one); correct threshold is > 0
    if len(missing_songs) > 0:
        logger.warning(
            "%d unique songs still unreconciled — generating missing songs files.",
            len(missing_songs),
        )
        missing_songs.insert(0, "Orchard PUB_SONG_ID", "")
        missing_songs.insert(0, "Updated in Sales Sheet?", "")

        util.df_excel_output(
            missing_songs, f"PubSongIDsNotInListMaster-{period}", timestamp=False
        )
        month = reporting_month["Month"]
        year = reporting_month["YYYY"]
        util.df_excel_output(
            missing_songs.drop_duplicates(subset=["ExtSong", "Song"], keep="first"),
            f"{month} {year} - Missing Songs & Agreements.xlsx",
            timestamp=False,
        )
    else:
        logger.info("All Pub Song IDs reconciled.")
        df.to_pickle(f"P{period}_Reconciled_Dataframe.pkl")

        if excel_output:
            util.df_excel_output(
                df, f"{fileseries}-PubSongIDReconciled.xlsx", timestamp=False
            )
        if csv_output:
            util.df_csv_output(df, f"{fileseries}-PubSongIDReconciled.csv")

    # --- Validation log (always written) ---
    log = util.Textobj()
    log.add(
        f"P{period} — {fileseries} reconciliation validation\n"
        f"{'-' * 79}\n"
        f"Starting row count : {start_len}\n"
        f"Starting Amount    : $ {start_amt}\n"
        f"Starting RoyAmt    : $ {start_roy}\n\n"
        f"Ending row count   : {len(df)}\n"
        f"Ending Amount      : $ {df['Amount'].sum()}\n"
        f"Ending RoyAmt      : $ {df['RoyAmt'].sum()}"
    )
    if not drop_df.empty:
        summary = drop_df[["Song", "Amount", "RoyAmt"]].groupby("Song").sum()
        with pd.option_context("display.max_rows", None, "display.max_columns", None):
            log.add(
                f"\nRows dropped : {len(drop_df)}\n"
                f"Amount dropped : $ {drop_df['Amount'].sum()}\n"
                f"RoyAmt dropped : $ {drop_df['RoyAmt'].sum()}\n\n"
                f"{summary}"
            )
    log.save(f"{fileseries}-PubSongIDReconciled_Log-{util.get_timestamp()}")

    return df


# ---------------------------------------------------------------------------
# Step 3 — Update master lookup
# ---------------------------------------------------------------------------


def update_master_lookup(
    master_file: str,
    full_file: str,
    lite_file: str,
) -> None:
    """Step 3 — Merge stakeholder-corrected lite file into full, then into master.

    Workflow:
      lite  → Aidan fills in Orchard PUB_SONG_IDs for missing/incorrect songs
      full  → the full unreconciled set from this run
      master → accumulates all corrections across quarters

    Args:
        master_file: Path to the master lookup XLSX.
        full_file:   Path to this run's full missing-songs XLSX.
        lite_file:   Path to the lite (stakeholder-corrected) XLSX.
    """
    lite = pd.read_excel(lite_file, dtype=str)
    full = pd.read_excel(full_file, dtype=str)

    # Restrict full to only the songs present in lite (partial update safety)
    full = full.merge(lite[["ExtSong", "Song"]], on=["ExtSong", "Song"], how="inner")

    for _, row in lite.iterrows():
        # Bug fix: pd.isna() on dtype=str gives 'nan' string, not NaN — check string
        if row["ExtSong"].strip() in ("", "nan"):
            # Match by Song title when ExtSong is blank
            mask = full["ExtSong"].str.strip().isin(("", "nan")) & (
                full["Song"] == row["Song"]
            )
        else:
            mask = (full["ExtSong"] == row["ExtSong"]) & (full["Song"] == row["Song"])
        full.loc[mask, "Orchard PUB_SONG_ID"] = row["Orchard PUB_SONG_ID"]

    missing_cnt = full["Orchard PUB_SONG_ID"].isna().sum() + (
        full["Orchard PUB_SONG_ID"].str.strip().isin(("", "nan")).sum()
    )
    if missing_cnt > 0:
        logger.warning(
            "%d rows still missing a PUB_SONG_ID after merge — manual review needed.",
            missing_cnt,
        )
        print(
            full[full["Orchard PUB_SONG_ID"].str.strip().isin(("", "nan"))][
                ["ExtSong", "Song"]
            ]
        )
    else:
        logger.info("Addendum lookup fully reconciled.")

    # Merge into master
    master = pd.read_excel(master_file, dtype=str)
    add_rows = pd.DataFrame(columns=master.columns)
    add_rows[master.columns] = full[
        ["Orchard PUB_SONG_ID", "ExtSong", "Song", "Writer", "Artist"]
    ]

    combined = (
        pd.concat([master, add_rows])
        .astype(str)
        .fillna("")
        .replace("nan", "")
        .drop_duplicates()
    )

    backup = master_file.replace(".xlsx", f"-Backup-{util.get_timestamp()}.xlsx")
    logger.info("Backing up master lookup to %s", backup)
    os.rename(master_file, backup)
    util.df_excel_output(combined, master_file, timestamp=False)
    logger.info("Master lookup updated: %s", master_file)


# ---------------------------------------------------------------------------
# Step 4 — Missing songwriter agreement check
# ---------------------------------------------------------------------------


def check_missing_songwriter_agreements(
    df: pd.DataFrame,
    period: int,
) -> None:
    """Step 4 — Find songs missing songwriter agreements in Snowflake and Neo4j.

    Queries Snowflake for songs with valid agreements, then falls back to
    Neo4j for any still missing. If Neo4j has agreements that Snowflake
    doesn't, a DBPR XML changeset is generated to force a sync.

    Outputs:
    - PubSongIDsWithoutSWA-FULL-{timestamp}.xlsx
    - PubSongIDsWithoutSWA-LITE-{timestamp}.xlsx
    - PUB-{key}_touch_composition_node.xml  (if DBPR is generated)
    """
    swa1 = sf.sf_df(sql[1])
    swa2 = sf.sf_df(sql[2])

    for d in [swa1, swa2]:
        for col in ["PUB_SONG_ID", "LABEL_ID"]:
            d[col] = d[col].astype(str).str.strip()

    known_ids = pd.concat([swa1["PUB_SONG_ID"], swa2["PUB_SONG_ID"]]).unique().tolist()
    missing_df = df[~df["ExtSong"].isin(known_ids)].copy()

    unique_missing = missing_df[missing_df["ExtSong"] != ""]["ExtSong"].nunique()
    logger.info("%d unique PUB_SONG_IDs have no songwriter agreement.", unique_missing)

    if missing_df.empty:
        logger.info("No songs missing songwriter agreements.")
        return

    pub_ids = [x for x in missing_df["ExtSong"].unique() if x]
    neo4j_found = _check_swa_neo4j(pub_ids, period)

    if neo4j_found:
        for pub_id in neo4j_found:
            missing_df = missing_df[missing_df["ExtSong"] != pub_id]

    if not missing_df.empty:
        util.df_excel_output(missing_df, "PubSongIDsWithoutSWA-FULL")
        util.df_excel_output(
            missing_df[["ExtSong", "Song", "Artist", "Writer"]]
            .drop_duplicates()
            .fillna("")
            .replace("nan", ""),
            "PubSongIDsWithoutSWA-LITE",
        )
    else:
        logger.info("All missing SWAs were found in Neo4j.")


def _check_swa_neo4j(pub_ids: list[str], period: int) -> list[dict] | None:
    """Query Neo4j for songwriter agreements. Returns list of found records or None."""
    cypher_template = _CYPHER_FILE.read_text()
    n4_db = "graph.db"
    found: list[dict] = []

    for pub_id in pub_ids:
        gql = cypher_template.replace("{pubsongid}", str(pub_id))
        with n4.get_driver("prod") as driver:
            result = driver.execute_query(gql, database=n4_db)
        if result.records:
            pc_id = result.records[0].data()["SongHasWriterAgreementInNeo4j"]
            entry = {"pubsongid": pub_id, "pcid": pc_id}
            if entry not in found:
                found.append(entry)
            logger.info("%s: SWA found in Neo4j.", pub_id)
        else:
            logger.info("%s: no SWA in Neo4j.", pub_id)

    if not found:
        logger.warning("No SWAs found in Neo4j. Review agreements in OA.")
        return None

    logger.info(
        "%d/%d songs have agreements in Neo4j — a DBPR may be required.",
        len(found),
        len(pub_ids),
    )
    print(
        f"\n{len(found)} songs found in Neo4j but not Snowflake.\n"
        "Reference PRs: https://github.com/theorchard/database/pull/18083/files\n"
    )

    create_ticket = input("Create a Jira ticket for the DBPR? (Y/N): ").strip().lower()

    if create_ticket == "y":
        ids_summary = ", ".join(str(x["pubsongid"]) for x in found)
        description = (
            f"Several songs have failed to sync from Neo4j to Snowflake "
            f"for period {period}. Affected pub_song_ids: {ids_summary}.\n\n"
            "We need agreements in place in Snowflake before the publishing royalty run."
        )
        key = ja.create_jira(
            project_id="PUB",
            summary=f"Sync songs from Neo4j to Snowflake P{period}",
            description=description,
            issue_type="Task",
            assignee_id=config.JIRA_ASSIGNEE_ID,
        )
    else:
        key = input(f"Enter the PUB Jira ticket number for P{period}: ").strip()

    if len(key) >= 3:
        jira_url = f"https://theorchard.atlassian.net/browse/PUB-{key}"
        print(f"Jira: {jira_url}")
        print(
            f"Create a `database` repo DBPR with name:\n  PUB-{key} Touch composition node to force sync."
        )
        generate_swa_dbpr(found, key)
    else:
        logger.warning("Invalid Jira key length — DBPR not generated.")

    return [x["pubsongid"] for x in found]


def generate_swa_dbpr(pc_id_list: list[dict | str], jira_key: str) -> None:
    """Generate a Liquibase/Cypher XML changeset file to force a Neo4j → Snowflake sync.

    Args:
        pc_id_list: List of dicts with 'pcid' key (output of _check_swa_neo4j),
                    or a plain list of PC ID strings.
        jira_key:   Jira issue number (without 'PUB-' prefix), e.g. '1234'.
    """
    # Normalise input — accept either list of dicts or list of strings
    if pc_id_list and isinstance(pc_id_list[0], dict):
        ids = [x["pcid"] for x in pc_id_list]
    else:
        ids = list(pc_id_list)

    template_parts = util.read_sql_file(str(_DBPR_TEMPLATE_FILE))
    output = util.Textobj()

    output.add(
        template_parts[0]
        .replace("{changesetcount}", "1")
        .replace("{author}", config.DBPR_AUTHOR)
        .replace("{date}", time.strftime("%Y-%m-%d"))
        .replace("{jira}", jira_key)
        .replace("{song_id_count}", str(len(ids)))
    )

    for pc_id in ids:
        output.add(
            template_parts[1]
            .replace("{uuid}", config.DBPR_AUTHOR_UUID)
            .replace("{pcid}", str(pc_id))
        )
        output.add(";\n\n")

    out_file = f"PUB-{jira_key}_touch_composition_node.xml"
    output.save(out_file)
    logger.info("DBPR changeset written to %s", out_file)


# ---------------------------------------------------------------------------
# Step 5 — Contract rate calculation
# ---------------------------------------------------------------------------


def contract_rate_cal(
    james_file: str,
    label_map_file: str,
    fileseries: str,
    period: int,
    reporting_month: dict,
) -> None:
    """Step 5 — Compile contract rates from James's file and the deal tracker.

    Outputs:
    - Publishing Contract Rates - {Month} {Year} - Period {period}.xlsx
    - Publishing Contract Rates Run Log - {timestamp}.txt

    Args:
        james_file:      TSV file received back from James Kass.
        label_map_file:  Local copy of the Publishing Admin Deal Tracker.
                         Download fresh from DEAL_TRACKER_GDRIVE_URL before running.
        fileseries:      Output filename prefix.
        period:          Snowflake period ID.
        reporting_month: Dict from get_reporting_month_object().
    """
    james_df = pd.read_csv(james_file, delimiter="\t")

    rates = pd.DataFrame({"Label ID": james_df["vendor_id"].unique()})

    label_map_raw = pd.read_excel(label_map_file, sheet_name=None)
    label_map = _parse_label_map(label_map_raw)

    contract_rates = rates.merge(label_map, how="left", on="Label ID").copy()
    contract_rates.sort_values(
        by=["Base Fee", "Label Name"],
        ascending=[False, True],
        ignore_index=True,
        inplace=True,
    )
    contract_rates.loc[contract_rates["Base Fee"] == 1, "Orchard Owned"] = "Yes"
    contract_rates.loc[contract_rates["SheetName"] == "Terminated", "Notes"] = (
        "Terminated"
    )
    contract_rates = contract_rates[
        ["Label ID", "Label Name", "Base Fee", "Orchard Owned", "Currency", "Notes"]
    ]

    month = reporting_month["Month"]
    year = reporting_month["YYYY"]
    util.df_excel_output(
        contract_rates,
        f"Publishing Contract Rates - {month} {year} - Period {period}",
        timestamp=False,
    )

    # Statistics
    deduped = james_df.drop_duplicates(subset=["roy_amt", "transaction_id"])
    roy_total = deduped["roy_amt"].sum()
    txn_cnt = len(deduped)

    txt = util.Textobj()
    txt.add(
        f"===== Contract Rate Calculation Statistics — Period {period} =====\n"
        f"{james_file}\n"
        f"  Unique Transaction IDs : {txn_cnt}\n"
        f"  Total Roy Amt          : $ {roy_total}\n"
        f"  Flat Row Count         : {len(james_df)}\n\n"
        f"{label_map_file}\n"
        f"  Label Count            : {len(label_map)}"
    )

    # Labels with missing rates
    missing_rates = contract_rates[
        (contract_rates["Notes"] != "Terminated")
        & contract_rates["Label Name"].notna()
        & (contract_rates["Base Fee"].isna() | contract_rates["Currency"].isna())
    ]
    if not missing_rates.empty:
        txt.add(
            f"***** Labels with missing rates or currency *****\n"
            f"{missing_rates.to_string(index=False)}"
        )
        cnt = (
            contract_rates[contract_rates["Label ID"].isin(missing_rates["Label ID"])][
                ["Label ID", "Label Name"]
            ]
            .merge(
                james_df[james_df["vendor_id"].isin(missing_rates["Label ID"])][
                    "vendor_id"
                ].value_counts(),
                how="inner",
                left_on="Label ID",
                right_on="vendor_id",
            )
            .sort_values("count", ascending=False)
        )
        txt.add(str(cnt))

    # Labels not in deal tracker at all
    missing_labels = contract_rates[contract_rates["Label Name"].isna()]["Label ID"]
    if not missing_labels.empty:
        summary = (
            james_df[james_df["vendor_id"].isin(missing_labels)]
            .groupby("vendor_id")
            .agg(row_count=("vendor_id", "size"), roy_amt_sum=("roy_amt", "sum"))
            .reset_index()
        )
        txt.add(
            f"***** Labels not found in deal tracker *****\n"
            f"{summary.to_markdown(index=False)}"
        )

    txt.print()
    txt.save(f"Publishing Contract Rates Run Log - {util.get_timestamp()}")


def _parse_label_map(raw: dict) -> pd.DataFrame:
    """Parse the Publishing Admin Deal Tracker workbook into a flat DataFrame.

    Expected sheets: 'Orchard Publishing Deals (Exter', 'Terminated', 'Orchard Owned'.
    Handles both 'Label ID'/'Label Name' and 'Client ID'/'Client Name' column naming.
    """
    result = pd.DataFrame(
        columns=["Label ID", "Label Name", "SheetName", "Base Fee", "Currency"],
        dtype=str,
    )
    target_sheets = ["Orchard Publishing Deals (Exter", "Terminated", "Orchard Owned"]

    for sheet_name, sheet_df in raw.items():
        if sheet_name not in target_sheets:
            continue

        has_fee = True
        if "Client ID" in sheet_df.columns:
            ids = sheet_df[["Client ID", "Client Name"]].rename(
                columns={"Client ID": "Label ID", "Client Name": "Label Name"}
            )
        elif "Label ID" in sheet_df.columns:
            ids = sheet_df[["Label ID", "Label Name"]]
        else:
            logger.warning(
                "Sheet '%s' has no recognised ID column — skipped.", sheet_name
            )
            continue

        ids = ids.copy()
        ids["SheetName"] = sheet_name

        if "Base Fee" in sheet_df.columns and "Currency" in sheet_df.columns:
            fees = sheet_df[["Base Fee", "Currency"]]
        else:
            fees = pd.DataFrame(
                {
                    "Base Fee": np.ones(len(sheet_df)),
                    "Currency": np.full(len(sheet_df), "USD"),
                }
            )

        merged = ids.join(fees)
        result = pd.concat([result, merged], ignore_index=True)

    result.dropna(
        subset=["Label ID", "Label Name", "Base Fee", "Currency"],
        how="all",
        inplace=True,
    )
    result.reset_index(drop=True, inplace=True)

    # Validate base fees
    txt = util.Textobj()
    result["Base Fee"] = pd.to_numeric(result["Base Fee"], errors="coerce")
    bad_fee_cnt = result["Base Fee"].isna().sum()
    if bad_fee_cnt > 0:
        txt.add(
            f"Non-numeric Base Fee values:\n{result[result['Base Fee'].isna()].to_markdown()}"
        )
        util.error_prompt(f"{bad_fee_cnt} invalid Base Fee rows will be dropped", 5)
    result.dropna(subset=["Base Fee"], inplace=True)

    invalid = result[
        (result["Base Fee"] > 1) | (result["Base Fee"] <= 0) | result["Currency"].isna()
    ]
    if not invalid.empty:
        txt.add(
            f"Invalid base fees or currency:\n"
            f"{invalid[['Label ID', 'Label Name', 'Base Fee', 'Currency']].to_markdown()}"
        )

    dupes = result[result["Label ID"].duplicated(keep=False)]
    if not dupes.empty:
        txt.add(
            f"Duplicated Label IDs:\n"
            f"{dupes[['Label ID', 'Label Name', 'Base Fee', 'Currency']].to_markdown()}"
        )

    if bad_fee_cnt > 0 or not invalid.empty or not dupes.empty:
        txt.save(f"Publishing Contract Rates Run Log - {util.get_timestamp()}")

    return result


# ---------------------------------------------------------------------------
# Step 6 — Payment summary calculation
# ---------------------------------------------------------------------------


def payment_summary_cal(
    contract_rates_file: str,
    currency_rates_file: str,
    fileseries: str,
    period: int,
) -> pd.DataFrame:
    """Step 6 — Assemble the Publishing Run Results report.

    Pulls summed sales data from Snowflake, joins with approved contract rates,
    applies currency codes fetched live from Abacus, and applies exchange rates
    from a locally downloaded Abacus payout rates file.

    Outputs:
    - {fileseries}-PublishingResults.xlsx

    Args:
        contract_rates_file: Approved contract rates XLSX (from step 5, post-review).
        currency_rates_file: Abacus payout rates XLS.
                             Download from Abacus > Reports > Payout Rates.
        fileseries:          Output filename prefix.
        period:              Snowflake period ID.

    Returns:
        The assembled results DataFrame.
    """
    sf_schema = os.getenv("SF_ROYALTY_SCHEMA", "qa")

    # Load approved contract rates
    approved = pd.read_excel(contract_rates_file)
    approved["Vendor ID"] = approved["Vendor ID"].astype(str)

    # Pull payout currency per vendor from Abacus (live Snowflake query)
    vendor_ids = ", ".join(approved["Vendor ID"].tolist())
    currency_query = config.ABACUS_CURRENCY_QUERY.format(account_ids=vendor_ids)
    currency_codes = sf.sf_df(currency_query)
    currency_codes["ACCOUNT_ID"] = currency_codes["ACCOUNT_ID"].astype(str)

    # Load local exchange rates file
    df_fx = _read_currency_file(currency_rates_file)

    # Build base results frame from approved contract rates
    results = approved[["Vendor ID", "Vendor Name", "Base Fee"]].copy()
    results.rename(columns={"Base Fee": "Base Fee from Contracts"}, inplace=True)

    # Merge in payout currency from Abacus
    results = results.merge(
        currency_codes[["ACCOUNT_ID", "CURRENCY_CODE"]].rename(
            columns={"ACCOUNT_ID": "Vendor ID", "CURRENCY_CODE": "Payout Currency"}
        ),
        on="Vendor ID",
        how="left",
    )

    # Apply exchange rate per vendor
    results["Exchange Rate"] = results["Payout Currency"].apply(
        lambda c: _currency_lookup(df_fx, c, "USD") if pd.notna(c) else None
    )

    # Pull summed sales data from Snowflake
    sales_query = sql[3].format(SF_ROYALTY_SCHEMA=sf_schema)
    sales_data = sf.sf_df(sales_query)
    sales_data["VENDOR ID"] = sales_data["VENDOR ID"].astype(str)
    sales_data.rename(columns=lambda c: c.title().replace("_", " "), inplace=True)

    # Confirm latest period available
    period_query = sql[4].format(SF_ROYALTY_SCHEMA=sf_schema)
    max_period = sf.sf_df(period_query).iloc[0, 0]
    logger.info("Latest Snowflake publishing period: %s", max_period)
    if str(max_period) != str(period):
        logger.warning(
            "config.PERIOD (%s) does not match the latest Snowflake period (%s). "
            "Confirm you are querying the correct period.",
            period,
            max_period,
        )

    # Join sales data
    results = results.merge(sales_data, how="left", on="Vendor ID")

    # QA calculations
    results["Calculated Base Fee"] = (
        results["Orchard Fees"] / results["Adjusted Gross"]
    ).round(4)
    results["Calculated Fee Diff"] = (
        results["Base Fee from Contracts"] - results["Calculated Base Fee"]
    ).round(4)
    results["% Gross"] = (
        results["Adjusted Gross"] / results["Adjusted Gross"].sum()
    ).round(4)

    col_order = [
        "Vendor ID",
        "Vendor Name",
        "Gross",
        "Adjusted Gross",
        "Base Fee from Contracts",
        "Orchard Fees",
        "Net Revenue To Customers",
        "Calculated Base Fee",
        "Calculated Fee Diff",
        "Payout Currency",
        "Exchange Rate",
        "Net Revenue To Customer In Payout Currency",
        "% Gross",
    ]
    # Only include columns that actually exist in results
    col_order = [c for c in col_order if c in results.columns]
    results = results[col_order].sort_values("% Gross", ascending=False)

    # Flag QA issues
    bad_diff = results[results["Calculated Fee Diff"].abs() > 0.001]
    if not bad_diff.empty:
        logger.warning(
            "%d vendor(s) have a non-zero Calculated Fee Diff:\n%s",
            len(bad_diff),
            bad_diff[["Vendor ID", "Vendor Name", "Calculated Fee Diff"]].to_string(
                index=False
            ),
        )

    util.df_excel_output(results, f"{fileseries}-PublishingResults", timestamp=False)
    logger.info("Payment summary complete.")
    return results


def _read_currency_file(filepath: str) -> pd.DataFrame:
    """Read an Abacus payout rates XLS into a normalised exchange-rate DataFrame."""
    df = pd.read_excel(filepath, skiprows=4)
    if "period_id" in df.columns:
        df.drop(columns="period_id", inplace=True)
    df.rename(columns={df.columns[0]: "From"}, inplace=True)
    return df


def _currency_lookup(df_fx: pd.DataFrame, from_ccy: str, to_ccy: str) -> float | None:
    """Look up an exchange rate from df_fx. Returns None if not found."""
    try:
        return df_fx.loc[df_fx["From"] == from_ccy.upper(), to_ccy.upper()].item()
    except (KeyError, ValueError):
        logger.warning("Exchange rate not found: %s → %s", from_ccy, to_ccy)
        return None


# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------


def reload_reconciled_file(period: int) -> pd.DataFrame:
    """Utility — Load a previously saved reconciled DataFrame from pickle."""
    pkl = f"P{period}_Reconciled_Dataframe.pkl"
    if not Path(pkl).exists():
        raise FileNotFoundError(
            f"{pkl} not found. Run step 2 (reconciliation) to completion first."
        )
    logger.info("Loading reconciled DataFrame from %s", pkl)
    return pd.read_pickle(pkl)


def cleanup_files(
    days: int,
    period: int,
    extensions: list[str],
    exceptions: list[str] = None,
    path: str = None,
) -> None:
    """Utility — Archive files older than `days` days to archive/{period}/{ext}/.

    Args:
        days:       Minimum file age in days. Use 0 to archive all matching files.
        period:     Period number used to name the archive subfolder.
        extensions: List of file extensions to archive (without leading dot).
        exceptions: List of filenames to leave in place.
        path:       Working directory. Defaults to current directory.
    """
    if exceptions is None:
        exceptions = []
    work_dir = Path(path or os.getcwd())
    cutoff = time.time() - (days * 86400)
    arch_base = work_dir / "archive" / str(period)

    count = 0
    log = util.Textobj()

    for f in work_dir.iterdir():
        if (
            f.is_file()
            and f.suffix.lstrip(".") in extensions
            and f.name not in exceptions
            and not f.name.startswith("~")
        ):
            if f.stat().st_mtime < cutoff:
                dest_dir = arch_base / f.suffix.lstrip(".")
                util.make_dir(dest_dir, quiet=True)
                dest = dest_dir / f.name
                f.rename(dest)
                count += 1
                # Bug fix: was missing separator between ext dir and filename
                log.add(f"{util.get_timestamp()} — Archived {f.name} → {dest}")

    logger.info("%d file(s) archived to %s", count, arch_base)
    log.save(str(arch_base / f"Cleanup-Log-{util.get_timestamp()}"))