generate_homeval.py

#!/usr/bin/env python3
"""
Generate HomeVal report markdown (and optionally HTML) files for a given comps
model run ID. A user may ask for either **one or more explicit PINs** *or* for **all
PINs that belong to a triad** (city, north, south). Exactly one of the two must
be supplied. If the user passes an empty string for either of the --pin or --triad
arguments, the script will ignore that argument.

Examples
--------
Generate every PIN:
    $ python3 generate_homeval.py \
          --run-id 2025-06-14-flamboyant-rob \

Generate two specific PINs:
    $ python3 generate_homeval.py \
          --run-id 2025-06-14-flamboyant-rob \
          --pin 01011000040000 10112040080000

Generate every PIN in towns 10 and 11:
    $ python3 generate_homeval.py \
          --run-id 2025-06-14-flamboyant-rob \
          --township 10 11
"""

from __future__ import annotations

import argparse
import os
import subprocess as sp
import gc
import re
import shutil
import time
from pathlib import Path

import ccao

import numpy as np
import pandas as pd
import orjson
from pyathena import connect
from pyathena.pandas.cursor import PandasCursor

import constants


# Argparse interface
# ─────────────────────────
# ───────────────────────────────────────────────────
def parse_args() -> argparse.Namespace:
    """Parse command‑line arguments and perform basic validation."""

    parser = argparse.ArgumentParser(
        description="Generate HomeVal report markdown (and optionally HTML) files",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )

    parser.add_argument(
        "--run-id",
        required=True,
        help="Comps run ID used by the pinval.vw_comps view (e.g. 2025-06-14-flamboyant-rob)",
    )

    parser.add_argument(
        "--pin",
        nargs="*",
        metavar="PIN",
        help=(
            "One or more Cook County PINs to generate reports for. When empty, "
            "generates reports for all PINs in the assessment year"
        ),
    )

    parser.add_argument(
        "--skip-html",
        action="store_true",
        help="Generate only frontmatter files; skip running the Hugo build step",
    )

    parser.add_argument(
        "--township",
        help=(
            "Restrict all-PIN mode to a single Cook County township code "
            "(two-digit string, e.g. 01, 23). Ignored if --pin is used."
        ),
    )

    parser.add_argument(
        "--environment",
        choices=["dev", "prod"],
        default="dev",
        help="Deployment target",
    )

    parser.add_argument(
        "--eligible-only",
        action="store_true",
        help=(
            "When present, restricts the assessment query to PINs where "
            "is_report_eligible = true to avoid generating static pages for "
            "ineligible parcels."
        ),
    )

    parser.add_argument(
        "--build-by-neighborhood",
        action="store_true",
        help=(
            "When present, writes Markdown content files in neighborhood "
            "subdirectories and instructs Hugo to build reports by "
            "neighborhood. This makes the build slower, but allows Hugo to use "
            "less memory, which can be useful for larger towns or in "
            "memory-constrained build environments."
        ),
    )

    args = parser.parse_args()

    if args.pin == [""]:
        # Remove empty string
        args.pin = []

    return args


# Declare functions
# ────────────────────────────────────────────────────────────────────────────
def pin_pretty(raw_pin: str) -> str:
    """Format 14‑digit Cook County PIN for display.

    If the PIN ends in '0000' we truncate it and return a 10-digit PIN."""

    truncated_pin = f"{raw_pin[:2]}-{raw_pin[2:4]}-{raw_pin[4:7]}-{raw_pin[7:10]}"
    if raw_pin[10:] == "0000":
        return truncated_pin
    else:
        return f"{truncated_pin}-{raw_pin[10:]}"


def _clean_predictors(raw: np.ndarray | list | str) -> list[str]:
    """
    Return a *clean* list of raw predictor column names.
    """

    # Parse numpy arrays and Arrow lists into plain Python lists
    if isinstance(raw, np.ndarray):
        raw = raw.tolist()

    # Clean up existing lists
    if isinstance(raw, list):
        preds_cleaned = [str(x).strip() for x in raw if str(x).strip()]
    else:
        # Fix list parsing
        txt = str(raw).strip()
        if txt.startswith("[") and txt.endswith("]"):
            txt = txt[1:-1]
        preds_cleaned = [p.strip() for p in txt.split(",") if p.strip()]

    # Add top chars to the front of the list
    top_chars = [
        "meta_nbhd_code",
        "char_class",
        "char_yrblt",
        "char_bldg_sf",
        "char_land_sf",
        "char_beds",
        "char_fbath",
        "char_hbath",
    ]
    preds_cleaned = [c for c in top_chars if c in preds_cleaned] + [
        p for p in preds_cleaned if p not in top_chars
    ]

    return preds_cleaned


def build_front_matter(
    df_target_pin: pd.DataFrame,
    df_comps: pd.DataFrame,
    vars_dict: dict[str, dict],
    environment: str,
) -> dict:
    """
    Assemble the front-matter dict for **one PIN**.

    Parameters
    ----------
    df_target_pin : DataFrame
        All assessment rows for this PIN (one per card).
    df_comps : DataFrame
        All comp rows for this PIN (across cards).
    vars_dict : dict
        Mapping of variable code for labels and tooltips
    """
    special_multi = bool(df_target_pin["is_parcel_small_multicard"].iloc[0])

    # Header
    tp = df_target_pin.iloc[0]  # all cards share the same PIN-level chars
    preds_cleaned: list[str] = _clean_predictors(tp["model_predictor_all_name"])

    # swap out the original sqft column for the combined version
    if special_multi:
        preds_cleaned = [
            "combined_bldg_sf" if p == "char_bldg_sf" else p for p in preds_cleaned
        ]

    front: dict = {
        "layout": "report",
        "title": "Report: How Did the Assessor's Model Estimate My Home Value?",
        "assessment_year": tp["assessment_year"],
        "final_model_run_date": pd.to_datetime(tp["final_model_run_date"]).strftime(
            "%B %d, %Y"
        )
        if tp["final_model_run_date"]
        else None,
        "pin": tp["meta_pin"],
        "pin_pretty": pin_pretty(tp["meta_pin"]),
        "pred_pin_final_fmv_round": tp["pred_pin_final_fmv_round"],
        "pred_pin_final_fmv_round_per_sqft": tp["pred_pin_final_fmv_round_per_sqft"],
        "meta_pin_num_cards": tp["meta_pin_num_cards"],
        "cards": [],
        "var_info": vars_dict,
        "special_case_multi_card": special_multi,
        "environment": environment,
        "is_prorated": tp["is_prorated"],
    }

    # Exit early if this PIN is ineligible for a report, in which case we
    # will pass the doc some info to help explain why the parcel is ineligible
    if not tp["is_report_eligible"]:
        front["layout"] = "ineligible"
        for attr in [
            "reason_report_ineligible",
            "assessment_triad_name",
            "char_class",
            "char_class_desc",
            "meta_triad_name",
        ]:
            front[attr] = tp[attr]
        return front

    # Per card
    for card_num, card_df in df_target_pin.groupby("meta_card_num"):
        card_df = card_df.iloc[0]

        comps_df = (
            df_comps[df_comps["card"] == card_num]
            .sort_values("comp_num")
            .reset_index(drop=True)
        )

        # Add all of the feature columns to the card
        subject_chars = {
            pred: card_df[pred] for pred in preds_cleaned if pred in card_df
        }
        # Keep the original building-SF for the subject only, while the
        # comps and all downstream outputs use the combined building sqft
        if special_multi and "char_bldg_sf" in card_df:
            subject_chars["char_bldg_sf"] = card_df["char_bldg_sf"]

        # Extract weights and scores for all features
        subject_char_scores = {
            pred: card_df[f"score_{pred}"]
            for pred in preds_cleaned
            if f"score_{pred}" in card_df
        }
        subject_char_weights = {
            pred: card_df[f"wt_{pred}"]
            for pred in preds_cleaned
            if f"wt_{pred}" in card_df
        }
        if special_multi and "score_char_bldg_sf" in card_df:
            # Combined value doesn't exist in the view, so construct it artificially
            subject_char_scores["char_bldg_sf"] = subject_char_scores[
                "combined_bldg_sf"
            ] = card_df["score_char_bldg_sf"]
            subject_char_weights["char_bldg_sf"] = subject_char_weights[
                "combined_bldg_sf"
            ] = card_df["wt_char_bldg_sf"]

        # Comps
        comps_list = []
        for _, comp in comps_df.iterrows():
            comp_dict = {
                "comp_num": comp["comp_num"],
                "pin": comp["comp_pin"],
                "pin_pretty": pin_pretty(comp["comp_pin"]),
                "is_subject_pin_sale": comp["is_subject_pin_sale"],
                "sale_price": comp["meta_sale_price"],
                "sale_price_short": comp["sale_price_short"],
                "sale_price_per_sq_ft": comp["sale_price_per_sq_ft"],
                "sale_date": comp["sale_month_year"],
                "document_num": comp["comp_document_num"],
                "property_address": comp["property_address"],
                "meta_nbhd_code": comp["meta_nbhd_code"],
            }

            # Make preds human-readable
            for pred in preds_cleaned:
                if pred not in comp_dict and pred in comp:
                    comp_dict[pred] = comp[pred]

            comps_list.append(comp_dict)

        # Comp summary
        comp_summary = {
            "sale_year_range_prefix": (
                "between" if " and " in comps_df["sale_year_range"].iloc[0] else "in"
            ),
            "sale_year_range": comps_df["sale_year_range"].iloc[0]
            if not comps_df.empty
            else "",
            "min_sale_price": comps_df["comps_min_sale_price"].iloc[0],
            "max_sale_price": comps_df["comps_max_sale_price"].iloc[0],
            "min_price_per_sqft": comps_df["comps_min_price_per_sqft"].iloc[0],
            "max_price_per_sqft": comps_df["comps_max_price_per_sqft"].iloc[0],
        }
        # Complete the card
        front["cards"].append(
            {
                "pin_pretty": pin_pretty(tp["meta_pin"]),
                "card_num": int(card_num),
                "is_frankencard": card_df["is_frankencard"],
                "char_class_detailed": card_df["char_class_detailed"],
                "location": {
                    k: v
                    for k, v in {
                        "property_address": card_df["property_address"],
                        "municipality": card_df["loc_property_city"],
                        "township": tp["meta_township_name"],
                        "meta_nbhd_code": card_df["meta_nbhd_code"],
                        "loc_school_elementary_district_name": card_df[
                            "school_elementary_district_name"
                        ],
                        "loc_school_secondary_district_name": card_df[
                            "school_secondary_district_name"
                        ],
                        "loc_latitude": card_df["loc_latitude"],
                        "loc_longitude": card_df["loc_longitude"],
                    }.items()
                },
                "chars": subject_chars,
                "char_scores": subject_char_scores,
                "char_weights": subject_char_weights,
                "has_subject_pin_sale": bool(comps_df["is_subject_pin_sale"].any()),
                "pred_card_initial_fmv": card_df["pred_card_initial_fmv"],
                "pred_card_initial_fmv_per_sqft": card_df[
                    "pred_card_initial_fmv_per_sqft"
                ],
                "comps": comps_list,
                "comp_summary": comp_summary,
                "predictors": preds_cleaned,
            }
        )

    return front


def convert_to_builtin_types(obj) -> object:
    """
    Recursively convert numpy types to native Python types in a nested structure.
    This is so the frontmatter doesn't through data type errors when being passed
    to the hugo template.
    """

    # Standardize NA string representations
    if isinstance(obj, str) and obj in {"nan"}:
        return ""

    if isinstance(obj, dict):
        return {k: convert_to_builtin_types(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_to_builtin_types(v) for v in obj]
    elif isinstance(obj, np.ndarray):
        if obj.size == 1:
            # Collapse 1-element arrays to their scalar value,
            # this is currently here because of municipality
            # behaviour in vw_pin10_location
            return convert_to_builtin_types(obj.item())
        else:
            # keep multi-element arrays as lists
            return [convert_to_builtin_types(v) for v in obj.tolist()]
    elif obj is pd.NA:  # pandas NA scalar
        return ""
    # Wrap NaN in quotes, otherwise the .nan breaks html map rendering
    elif isinstance(obj, (float, np.floating)) and np.isnan(obj):
        return ""
    elif isinstance(obj, np.generic):
        return obj.item()
    return obj


def convert_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cast expected columns in df to appropriate dtypes before building dicts.
    This avoids manual casting during dict construction later.
    """

    # Define conversions
    dtype_conversions = {
        "comp_num": "Int64",
        "char_yrblt": "Int64",
        "char_beds": "Int64",
        "char_fbath": "Int64",
        "char_hbath": "Int64",
        "loc_latitude": "float",
        "loc_longitude": "float",
        "is_subject_pin_sale": "boolean",
    }

    for col, dtype in dtype_conversions.items():
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="raise").astype(dtype)

    return df


def write_json(front_dict: dict, outfile: str | Path) -> None:
    """Write the frontmatter dict to a JSON file."""
    front_dict = convert_to_builtin_types(front_dict)

    json_bytes: bytes = orjson.dumps(
        front_dict,
        option=orjson.OPT_INDENT_2,
    )
    Path(outfile).write_text(json_bytes.decode("utf-8") + "\n", encoding="utf-8")


def label_percent(s: pd.Series) -> pd.Series:
    """0.123 → '12%' (handles NaNs)."""
    return s.mul(100).round(0).astype("Int64").astype(str).str.replace("<NA>", "") + "%"


def label_dollar(s: pd.Series) -> pd.Series:
    """45000 → '$45,000' (handles NaNs)."""
    return s.apply(lambda x: f"${x:,.0f}" if pd.notna(x) else "")


def format_df(df: pd.DataFrame, chars_recode=False) -> pd.DataFrame:
    """
    Format the DataFrame for frontmatter output.
    """

    def _get_display_dtype(col: str) -> str | None:
        """Helper function that takes a column name and returns a string
        representing the data type for that column, which indicates how we
        should format the column for display.

        Returns None if no data type is specified for this column."""
        INT_COLS = {
            # Columns like year shouldn't be formatted with commas,
            # so we preserve them as integers
            "time_sale_year",
            "time_sale_day",
            "acs5_median_household_total_occupied_year_built",
            "char_yrblt",
            "meta_pin_num_cards",
        }

        # Columns that should be formatted as dollar amounts
        DOLLAR_COLS = {
            "comps_min_sale_price",
            "comps_max_sale_price",
            "comps_min_price_per_sqft",
            "comps_max_price_per_sqft",
            "meta_sale_price",
            "sale_price_per_sq_ft",
            "acs5_median_household_renter_occupied_gross_rent",
            "pred_pin_final_fmv_round",
            "pred_pin_final_fmv_round_per_sqft",
            "pred_card_initial_fmv",
            "pred_card_initial_fmv_per_sqft",
        }

        # Columns that should be rounded to 5 significant digits
        ROUND_TO_5_COLS = {"loc_latitude", "loc_longitude"}

        if col in INT_COLS or col.startswith("score_"):
            return "int"

        if col in DOLLAR_COLS or col.startswith("acs5_median_income"):
            return "dollar"

        if col in ROUND_TO_5_COLS:
            return "round_to_5"

        if col.startswith("wt_"):
            return "16digit_float"

        if col.startswith("acs5_percent"):
            return "percent"

        # None indicates no specified dtype, in which case we may still
        # format the column based on the column data type (e.g.
        # we convert all unspecified numerics to comma-separated strings)
        return None

    def _fmt_generic_numeric(x):
        """Helper function for generic formatting of numeric columns as
        comma-separated strings."""
        if isinstance(x, (int, np.integer)):
            return f"{x:,}"
        if isinstance(x, (float, np.floating)):
            txt = f"{x:,.2f}".rstrip("0").rstrip(".")
            return txt
        return x

    # Recode character columns to human-readable values
    if chars_recode:
        chars_to_recode = [
            col for col in df.columns if col.startswith("char_") and col != "char_apts"
        ]

        df = ccao.vars_recode(
            data=df.copy(),
            cols=chars_to_recode,
            code_type="long",
            as_factor=False,
            dictionary=ccao.vars_dict,
        )

    # Generate comps summary stats needed for frontmatter, these conditionals
    # are our way of checking whether we're working with the comps table
    if "meta_sale_price" in df.columns:
        df["comps_min_sale_price"] = df.groupby(["pin", "card"])[
            "meta_sale_price"
        ].transform("min")
        df["comps_max_sale_price"] = df.groupby(["pin", "card"])[
            "meta_sale_price"
        ].transform("max")
    if "sale_price_per_sq_ft" in df.columns:
        df["comps_min_price_per_sqft"] = df.groupby(["pin", "card"])[
            "sale_price_per_sq_ft"
        ].transform("min")
        df["comps_max_price_per_sqft"] = df.groupby(["pin", "card"])[
            "sale_price_per_sq_ft"
        ].transform("max")

    formatted_df = (
        # Convert columns based on their configured data type
        df.pipe(
            lambda d: d.assign(
                **{
                    col: pd.to_numeric(d[col], errors="raise").astype("Int64")
                    for col in d.columns
                    if _get_display_dtype(col) == "int"
                }
            )
        )
        .pipe(
            lambda d: d.assign(
                **{
                    col: d[col].apply(round, ndigits=5)
                    for col in d.columns
                    if _get_display_dtype(col) == "round_to_5"
                }
            )
        )
        .pipe(
            lambda d: d.assign(
                **{
                    col: d[col].apply(lambda x: f"{x:.16f}")
                    for col in d.columns
                    if _get_display_dtype(col) == "16digit_float"
                }
            )
        )
        .pipe(
            lambda d: d.assign(
                **{
                    col: label_percent(d[col])
                    for col in d.columns
                    if _get_display_dtype(col) == "percent"
                }
            )
        )
        .pipe(
            lambda d: d.assign(
                **{
                    col: label_dollar(d[col])
                    for col in d.columns
                    if _get_display_dtype(col) == "dollar"
                }
            )
        )
        # Convert generic numerics → comma-separated strings (except cols with
        # a specified data type)
        .pipe(
            lambda d: d.assign(
                **{
                    col: d[col].apply(_fmt_generic_numeric)
                    for col in d.select_dtypes(include="number").columns
                    if _get_display_dtype(col) is None
                }
            )
        )
    )

    return formatted_df


def compute_shap_weights(df: pd.DataFrame) -> pd.DataFrame:
    """Given a dataframe containing SHAP values for all predictors, use those
    SHAP values to compute importance weights for each predictor.

    We define the "importance weight" of a predictor as the absolute magnitude
    of its SHAP value in proportion to the summed absolute magnitude of all
    SHAPs for all predictors (minus the SHAP baseline). This gives us a sense
    of how much that predictor affects the prediction for an observation
    compared to all other predictors that the model uses.

    This function adds two new columns for each predictor. The columns follow
    this naming convention:

    - wt_<predictor>: The weight of the predictor, on the range [0, 1]
    - score_<predictor> The score of the weight, on the range [1, 4]

    We use integer bins to compute a score for each predictor in the report
    that is easier for readers to interpret than the raw weight.
    """

    shap_cols = [col for col in df.columns if col.startswith("shap_")]
    # Sum the absolute values of every SHAP in each row, setting sums of zero
    # to null to be safe (avoids division by zero)
    sum_df = df[shap_cols].abs().sum(axis=1).replace(0, np.nan)
    weights_df = df[shap_cols].abs().div(sum_df, axis=0)

    # Cut the weights into predefined bins so that we can display them using a
    # simplified "score" on the range [1, 4]
    bins = [-np.inf, 0.0005, 0.005, 0.025, np.inf]
    scores_df = weights_df.apply(
        lambda row: pd.cut(row, bins=bins, labels=list(range(1, 5))),
        axis=1,
    )

    # Drop SHAP columns since we don't need them in the report and they just
    # take up space on disk
    df = df.drop(shap_cols, axis=1)

    # Add weights and scores to the output dataframe
    df = pd.concat(
        [
            df,
            weights_df.rename(columns=lambda col: re.sub("^shap_", "wt_", col)),
            scores_df.rename(columns=lambda col: re.sub("^shap_", "score_", col)),
        ],
        axis=1,
    )

    return df


def run_athena_query(cursor, sql: str, params: dict = None) -> pd.DataFrame:
    cursor.execute(sql, parameters=params)
    return cursor.as_pandas()


def load_vars_dict(cursor) -> dict[str, dict]:
    """
    Pull the crosswalk for labels and tooltips into a dict:
    """
    sql = f"""SELECT code, display_name, description FROM {constants.HOMEVAL_DATA_DICT_TABLE}"""
    df = run_athena_query(cursor, sql)
    # Return a dict where each key is a column code and the value is a dict with label and description info
    return {
        row.code: {
            "label": (
                row.display_name
                if getattr(row, "display_name", None)
                else str(row.code)
            ),
            "description": (
                row.description
                if getattr(row, "description", None) not in (None, "", "nan")
                else ""
            ),
        }
        for row in df.itertuples(index=False)
    }


def main() -> None:
    args = parse_args()

    project_root = Path(sp.getoutput("git rev-parse --show-toplevel"))
    os.chdir(project_root)

    # Athena connection (one per run)
    cursor = connect(
        # We add '+ "/"' to the end of the line below because enabling unload
        # requires that the staging directory end with a slash. Add rstrip because
        # the missing '/' seems to depend on different environments
        s3_staging_dir=os.getenv("AWS_ATHENA_S3_STAGING_DIR").rstrip("/") + "/",
        region_name=os.getenv("AWS_REGION"),
        cursor_class=PandasCursor,
    ).cursor(unload=True)

    # Extract assessment year for model run, which we'll use to determine
    # the output path
    assessment_year_df = run_athena_query(
        cursor,
        "SELECT assessment_year FROM model.metadata WHERE run_id = %(run_id)s",
        {"run_id": args.run_id},
    )
    if assessment_year_df.empty:
        raise ValueError(f"No model metadata found for model run {args.run_id}")

    assessment_year = assessment_year_df.iloc[0]["assessment_year"]

    assessment_clauses = ["assessment_year = %(assessment_year)s"]
    params_assessment = {"assessment_year": assessment_year}

    # Shard by township **only** in the assessment query
    if args.township:
        assessment_clauses.append("meta_township_code = %(township)s")
        params_assessment["township"] = args.township

    if args.pin:
        pins: list[str] = list(set(args.pin))  # de-dupe
        pin_params = {f"pin{i}": p for i, p in enumerate(pins)}
        placeholders = ",".join(f"%({k})s" for k in pin_params)

        assessment_clauses.append(f"meta_pin IN ({placeholders})")
        params_assessment = {**params_assessment, **pin_params}

    if args.eligible_only:
        assessment_clauses.append("is_report_eligible = TRUE")

    where_assessment = " AND ".join(assessment_clauses)

    assessment_sql = f"""
        SELECT *
        FROM {constants.HOMEVAL_ASSESSMENT_CARD_TABLE}
        WHERE {where_assessment}
    """

    print("Querying data from Athena ...")
    df_assessment_all = run_athena_query(cursor, assessment_sql, params_assessment)
    print("Shape of df_assessment_all:", df_assessment_all.shape)

    if df_assessment_all.empty:
        raise ValueError(
            f"No assessment rows returned for the following params: {params_assessment}"
        )

    print("Computing SHAP weights for feature ranking")
    df_assessment_all = compute_shap_weights(df_assessment_all)

    print("Formatting assessment dataframe for display")
    df_assessment_all = format_df(df_assessment_all)

    comps_sql = f"""
        SELECT comp.*
        FROM {constants.HOMEVAL_COMP_TABLE} AS comp
        INNER JOIN (
            SELECT DISTINCT meta_pin
            FROM {constants.HOMEVAL_ASSESSMENT_CARD_TABLE}
            WHERE {where_assessment}
        ) AS card
          ON comp.pin = card.meta_pin
        WHERE comp.run_id = %(run_id)s
    """

    params_comps = {
        "run_id": args.run_id,
        **params_assessment,
    }

    print("Executing comps query …")
    start_q = time.time()

    df_comps_all = run_athena_query(cursor, comps_sql, params_comps)
    print(f"Comps query finished in {time.time() - start_q:.2f}s")
    print("Shape of df_comps_all:", df_comps_all.shape)
    # Transform values in character columns to human-readable values.
    # This is already done in the assessment query, so we only need to do it here.
    # We don't need to do it if the comps are empty, in which case we're
    # probably working on a township that was not reassessed
    if not df_comps_all.empty:
        df_comps_all = format_df(convert_dtypes(df_comps_all), chars_recode=True)

    # Variables dictionary for labels + tooltips
    vars_dict = load_vars_dict(cursor)

    # Close the database cursor since we're done with queries
    cursor.close()

    # Declare outputs paths
    hugo_root = project_root / "hugo"
    content_root = hugo_root / "content"
    md_outdir = content_root / "homeval-reports"
    md_outdir.mkdir(parents=True, exist_ok=True)

    start_time_dict_groupby = time.time()

    # Group dfs by PIN in dict for theoretically faster access
    df_assessments_by_pin = df_assessment_all.groupby("meta_pin")
    df_comps_by_pin = (
        {} if df_comps_all.empty else dict(tuple(df_comps_all.groupby("pin")))
    )
    end_time_dict_groupby = time.time()

    del df_assessment_all
    del df_comps_all
    gc.collect()
    print(
        f"Grouping by PIN took {end_time_dict_groupby - start_time_dict_groupby:.2f} seconds"
    )

    # Iterate over each unique PIN and output frontmatter
    print("Iterating pins to generate frontmatter")
    start_time = time.time()
    pin_groups = df_assessments_by_pin.groups
    for i, pin in enumerate(pin_groups):
        if i % 5000 == 0:
            print(f"Processing PIN {i + 1} of {len(pin_groups)}")

        # Use get_group to lower memory use when iterating grouped DF
        df_target = df_assessments_by_pin.get_group(pin)

        # Determine where to write content files. If the user has specified
        # that we should write content files in neighborhood subdirectories,
        # adjust the output path based on the PIN's neighborhood
        md_dirpath = md_outdir
        if args.build_by_neighborhood:
            nbhd_code = df_target.iloc[0]["meta_nbhd_code"]
            md_dirpath = md_outdir / nbhd_code
            md_dirpath.mkdir(exist_ok=True)

        md_path = md_dirpath / f"{pin}.md"

        df_comps = df_comps_by_pin.get(pin)

        front = build_front_matter(
            df_target,
            df_comps,
            vars_dict,
            environment=args.environment,
        )
        front["url"] = f"/{assessment_year}/{pin}.html"

        write_json(front, md_path)

    elapsed_time = time.time() - start_time
    print(
        f"✓ Completed generating frontmatter for {len(df_assessments_by_pin)} PINs in {elapsed_time:.4f} seconds."
    )

    # ------------------------------------------------------------------
    # Optional Hugo build
    # ------------------------------------------------------------------

    if not args.skip_html:
        # Clean up memory before running Hugo, this prevents github runners
        # from running out of memory
        del df_assessments_by_pin
        del df_comps_by_pin
        del df_target
        del df_comps
        del pin_groups

        gc.collect()

        # Generate the HTML files using Hugo
        if args.build_by_neighborhood:
            print("Building Hugo reports by neighborhood …")

            # Move reports directory out of the Hugo content root so that we
            # can move reports back in neighborhood-by-neighborhood
            temp_reports_dir = hugo_root / "temp-homeval-reports"
            if temp_reports_dir.exists():
                shutil.rmtree(temp_reports_dir)
            shutil.move(str(md_outdir), str(temp_reports_dir))

            # Move neighborhoods into the Hugo content root one by one and
            # generate reports
            for nbhd_dir in temp_reports_dir.iterdir():
                if nbhd_dir.is_dir():
                    target_dir = content_root / nbhd_dir.name
                    print(f"Moving {nbhd_dir} to {target_dir} and building reports")
                    shutil.move(str(nbhd_dir), str(target_dir))
                    proc = sp.run(["hugo", "--minify"], cwd=hugo_root, text=True)
                    if proc.returncode != 0:
                        raise RuntimeError("Hugo build failed.")
                    print(f"Deleting {target_dir}")
                    shutil.rmtree(str(target_dir))

            # Remove temporary reports dir that should now be empty
            temp_reports_dir.rmdir()
        else:
            print("Building Hugo reports …")
            proc = sp.run(["hugo", "--minify"], cwd=hugo_root, text=True)
            if proc.returncode != 0:
                raise RuntimeError("Hugo build failed.")

            # Remove markdown files now that HTML is baked.
            shutil.rmtree(md_outdir)

        print("✓ Hugo build complete")


if __name__ == "__main__":
    main()