From 29f2a47207f6423d98e65d6a7cc06126f378be2d Mon Sep 17 00:00:00 2001 From: Zach Latta Date: Thu, 10 Sep 2026 10:08:46 -0400 Subject: [PATCH 1/2] Sync YSWS costs directly from the canonical true-spend mart --- docs/ysws-true-spend-airtable-cutover.md | 59 +++++ .../defs/unified_ysws_db/definitions.py | 231 +----------------- .../defs/unified_ysws_db/true_spend_sync.py | 143 +++++++++++ orpheus_engine/schedules.py | 1 + .../test_ysws_true_spend_sync.py | 118 +++++++++ 5 files changed, 329 insertions(+), 223 deletions(-) create mode 100644 docs/ysws-true-spend-airtable-cutover.md create mode 100644 orpheus_engine/defs/unified_ysws_db/true_spend_sync.py create mode 100644 orpheus_engine_tests/test_ysws_true_spend_sync.py diff --git a/docs/ysws-true-spend-airtable-cutover.md b/docs/ysws-true-spend-airtable-cutover.md new file mode 100644 index 0000000..88e4799 --- /dev/null +++ b/docs/ysws-true-spend-airtable-cutover.md @@ -0,0 +1,59 @@ +# YSWS true-spend Airtable cutover + +The website and YSWS cost sync both read +`public_hcb_ysws_true_spend_analytics.ysws_spend_by_program`. +Do not calculate cost from HCB raised minus balance, add estimated postage, or +redivide by an individual Airtable row's hours. Several program versions may +share one HCB root; the website reports their combined spend/hours/rate. +The sync uses Airtable record IDs in `member_ids`, not program names, and checks +that the current HCB link still matches the snapshot's root. + +## One-time Airtable schema change + +In Unified YSWS Projects DB → YSWS Programs, add: + +- `True Spend — Weighted Hours`: number (4 decimal display places). +- `True Spend — Cost Per Weighted Hour`: currency (2 decimal places). + +Keep `Total Spent From HCB Fund` as the writable spend column for compatibility; +it now receives canonical **true spend**, not gross HCB outflow. + +Change the existing **Cost Per Hour** formula (keep its field ID): + +```text +IF({True Spend — Weighted Hours} > 0, {True Spend — Cost Per Weighted Hour}, BLANK()) +``` + +Change the existing **Total Spend** formula (keep its field ID): + +```text +IF(({Total Spent From HCB Fund} & "") != "", {Total Spent From HCB Fund}, BLANK()) +``` + +The string check preserves a real zero without turning missing mappings into +zero spend. Keep the old postage inputs for reference; the formula must no +longer add the estimate on top of the reconciled ledger. + +## Deployment / verification + +1. Create the two fields through the reviewed schema-change path. +2. Deploy the code: the legacy gross-spend writer is removed. The new writer + refuses writes until the existing formula dependencies have been updated. +3. Update the two formulas above in Airtable's UI (the API cannot edit formulas). +4. Materialize `ysws_programs_hcb_stats` and + `ysws_programs_true_spend_update_status`. No unrelated signup or project + processing needs to run for this cutover. +5. Reconcile every matched record's spend, hours and displayed cost/hour against + its canonical mart row. Check zero spend, zero/missing hours, removed HCB + links, and shared roots. Unknown mappings clear old derived numbers to blank; + they never fall back to gross spend. + +The new sync is selected in the existing 15-minute Unified YSWS job. It reads +the latest successful true-spend snapshot, requiring a materialization within +36 hours. It does not trigger a heavy HCB mirror/dbt rebuild every 15 minutes. +It is deliberately outside `ysws_programs_prepared_for_update` and +`unified_ysws_db_processing_done`: those lead to the source warehouse refresh +that dbt reads, so making them depend on the true-spend output creates a cycle. + +Do not claim live reconciliation is complete until schema, formulas, production +materialization, and read-back are all verified. diff --git a/orpheus_engine/defs/unified_ysws_db/definitions.py b/orpheus_engine/defs/unified_ysws_db/definitions.py index 073a372..f38ddde 100644 --- a/orpheus_engine/defs/unified_ysws_db/definitions.py +++ b/orpheus_engine/defs/unified_ysws_db/definitions.py @@ -26,6 +26,7 @@ from ..airtable.definitions import airtable_config from ..geocoder.resources import GeocoderResource, GeocodingError from ..airtable.generated_ids import AirtableIDs +from .true_spend_sync import ysws_programs_hcb_stats, ysws_programs_true_spend_update_status from ..shared.address_utils import build_address_string_from_airtable_row from ..shared.daily_backups import ( ParquetBackupFile, @@ -751,219 +752,6 @@ def ysws_programs_sign_up_stats_candidates( ) -@asset( - group_name="unified_ysws_db_processing", - description="Prepares YSWS programs with HCB data by extracting HCB IDs from URLs", - compute_kind="data_preparation", - deps=[AssetKey(["airtable", "unified_ysws_projects_db", "ysws_programs"])], - required_resource_keys={"airtable"}, -) -def ysws_programs_hcb_candidates( - context: AssetExecutionContext, -) -> Output[pl.DataFrame]: - """ - Loads YSWS programs and extracts HCB IDs from HCB URLs. - - Returns: - DataFrame with id, hcb (URL), and hcb_id (extracted ID) - """ - log = context.log - airtable: AirtableResource = context.resources.airtable - - # Get the ysws_programs data from airtable - programs_df = airtable.get_all_records_as_polars( - context=context, - base_key="unified_ysws_projects_db", - table_key="ysws_programs", - ) - - log.info(f"Processing {programs_df.height} YSWS programs for HCB data") - - # Get the HCB field ID - hcb_field_id = UnifiedYSWS.ysws_programs.hcb - - if hcb_field_id not in programs_df.columns: - log.warning(f"HCB field {hcb_field_id} not found in programs data") - # Return empty DataFrame with correct schema - return Output( - pl.DataFrame(schema={ - "id": pl.Utf8, - "hcb": pl.Utf8, - "hcb_id": pl.Utf8 - }), - metadata={"num_programs": 0, "num_with_hcb": 0} - ) - - # Filter for programs that have HCB field set and extract HCB IDs - processed_df = programs_df.filter( - (pl.col(hcb_field_id).is_not_null()) & - (pl.col(hcb_field_id) != "") - ).with_columns([ - pl.col(hcb_field_id).map_elements( - lambda x: _extract_hcb_id_from_url(x) if x else "", - return_dtype=pl.Utf8 - ).alias("hcb_id") - ]).select([ - pl.col("id"), - pl.col(hcb_field_id).alias("hcb"), - pl.col("hcb_id") - ]).filter( - # Only include programs with valid HCB IDs (skip malformed URLs) - pl.col("hcb_id") != "" - ) - - log.info(f"Found {processed_df.height} programs with valid HCB URLs out of {programs_df.height} total programs") - if processed_df.height > 0: - log.info(f"Sample HCB data: {processed_df.head(3).to_dicts()}") - - # Generate preview metadata - if processed_df.height > 0: - try: - preview_metadata = MetadataValue.md(processed_df.head(10).to_pandas().to_markdown(index=False)) - except Exception: - preview_metadata = MetadataValue.text(str(processed_df.head(10))) - else: - preview_metadata = MetadataValue.text("No programs with valid HCB URLs found") - - return Output( - processed_df, - metadata={ - "num_programs": programs_df.height, - "num_with_hcb": processed_df.height, - "hcb_ids": processed_df["hcb_id"].to_list(), - "preview": preview_metadata - } - ) - - -@asset( - group_name="unified_ysws_db_processing", - description="Fetches HCB financial data and calculates total spent from HCB fund", - compute_kind="api_request", -) -def ysws_programs_hcb_stats( - context: AssetExecutionContext, - ysws_programs_hcb_candidates: pl.DataFrame, -) -> Output[pl.DataFrame]: - """ - Fetches HCB organization data and calculates total spent from HCB fund. - - Returns: - DataFrame with id and total_spent_from_hcb_fund field for Airtable updates - """ - log = context.log - input_df = ysws_programs_hcb_candidates - - total_spent_field_id = UnifiedYSWS.ysws_programs.total_spent_from_hcb_fund - - if input_df.height == 0: - log.info("No HCB candidates to process.") - return Output( - pl.DataFrame(schema={ - "id": pl.Utf8, - total_spent_field_id: pl.Float64, - }), - metadata={"num_processed": 0, "num_successful": 0, "num_failed": 0} - ) - - log.info(f"Processing HCB data for {input_df.height} programs") - - successful_records = [] - failures = [] - - for row in input_df.iter_rows(named=True): - program_id = row.get("id") - hcb_id = row.get("hcb_id") - hcb_url = row.get("hcb") - - if not hcb_id: - log.warning(f"No HCB ID for program {program_id} (url={hcb_url!r})") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": "No HCB ID extracted"}) - continue - - try: - api_url = f"https://hcb.hackclub.com/api/v3/organizations/{hcb_id}" - log.debug(f"Fetching HCB data for {hcb_id}: {api_url}") - - response = requests.get( - api_url, - headers={"Accept": "application/json"}, - timeout=30 - ) - - if response.status_code != 200: - try: - error_data = response.json() - error_message = error_data.get("message", f"HTTP {response.status_code}") - except Exception: - error_message = f"HTTP {response.status_code}" - - log.warning(f"HCB API returned {response.status_code} for {hcb_id}: {response.text}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": error_message}) - continue - - data = response.json() - balances = data.get("balances", {}) - total_raised = balances.get("total_raised", 0) - balance_cents = balances.get("balance_cents", 0) - total_spent_dollars = (total_raised - balance_cents) / 100.0 - - successful_records.append({ - "id": program_id, - total_spent_field_id: total_spent_dollars, - }) - - log.info(f"Successfully processed {hcb_id}: total_raised={total_raised}, balance_cents={balance_cents}, total_spent=${total_spent_dollars:.2f}") - - except requests.exceptions.RequestException as e: - log.error(f"Request failed for HCB ID {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Request failed: {e}"}) - except (KeyError, ValueError, TypeError) as e: - log.error(f"Error parsing HCB data for {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Data parsing error: {e}"}) - except Exception as e: - log.error(f"Unexpected error processing HCB ID {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Unexpected error: {e}"}) - - if successful_records: - output_df = pl.DataFrame(successful_records) - else: - output_df = pl.DataFrame(schema={ - "id": pl.Utf8, - total_spent_field_id: pl.Float64, - }) - - successful_count = len(successful_records) - failed_count = len(failures) - log.info(f"HCB processing completed. Successful: {successful_count}, Failed: {failed_count}") - - # Generate preview metadata - if output_df.height > 0: - try: - preview_metadata = MetadataValue.md(output_df.head(10).to_pandas().to_markdown(index=False)) - except Exception: - preview_metadata = MetadataValue.text(str(output_df.head(10))) - else: - preview_metadata = MetadataValue.text("No HCB data processed successfully") - - metadata = { - "num_processed": input_df.height, - "num_successful": successful_count, - "num_failed": failed_count, - "success_rate": round((successful_count / max(input_df.height, 1)) * 100, 2), - "preview": preview_metadata, - } - if failures: - try: - metadata["failures"] = MetadataValue.md( - pl.DataFrame(failures).to_pandas().to_markdown(index=False) - ) - except Exception: - metadata["failures"] = MetadataValue.text(str(failures)) - - return Output(output_df, metadata=metadata) - - def _get_signup_analysis_data(search_terms: List[str]) -> pl.DataFrame: """ Get sign-up analysis data from hack_clubbers for the given search terms. @@ -1839,16 +1627,15 @@ def ysws_programs_sign_up_stats( @asset( group_name="unified_ysws_db_processing", - description="Prepares YSWS programs data for Airtable batch update by merging sign-up stats and HCB data.", + description="Prepares YSWS programs data for Airtable batch update from sign-up stats. True-spend updates run separately after dbt.", compute_kind="data_preparation", ) def ysws_programs_prepared_for_update( context: AssetExecutionContext, ysws_programs_sign_up_stats: pl.DataFrame, - ysws_programs_hcb_stats: pl.DataFrame, ) -> Output[pl.DataFrame]: """ - Merges sign-up stats and HCB data for YSWS programs and prepares for Airtable batch update. + Prepares sign-up stats for YSWS programs and prepares for Airtable batch update. Checks for conflicts: if multiple inputs provide different non-null values for the same field for a given program ID, an error is raised. @@ -1857,13 +1644,12 @@ def ysws_programs_prepared_for_update( log = context.log # --- 1. Collect and Filter Input DataFrames --- - dfs = [ysws_programs_sign_up_stats, ysws_programs_hcb_stats] + dfs = [ysws_programs_sign_up_stats] input_heights = { "sign_up_stats": ysws_programs_sign_up_stats.height, - "hcb_stats": ysws_programs_hcb_stats.height, } - log.info(f"Input counts - Sign-up stats: {input_heights['sign_up_stats']}, HCB stats: {input_heights['hcb_stats']}") + log.info(f"Input counts - Sign-up stats: {input_heights['sign_up_stats']}") dfs = [df for df in dfs if not df.is_empty()] @@ -1992,7 +1778,6 @@ def ysws_programs_prepared_for_update( metadata={ "num_records_prepared": final_df.height, "num_sign_up_stats_input": input_heights["sign_up_stats"], - "num_hcb_stats_input": input_heights["hcb_stats"], "update_columns": list(final_df.columns), "preview": preview_metadata } @@ -2001,7 +1786,7 @@ def ysws_programs_prepared_for_update( @asset( group_name="unified_ysws_db_processing", - description="Updates YSWS programs in Airtable with merged sign-up stats and HCB data.", + description="Updates YSWS programs in Airtable with sign-up stats (true-spend costs have a separate post-dbt writer).", required_resource_keys={"airtable"}, compute_kind="airtable_update", ) @@ -2010,7 +1795,7 @@ def ysws_programs_update_status( ysws_programs_prepared_for_update: pl.DataFrame, ) -> Output[None]: """ - Performs batch update of YSWS programs in Airtable with merged sign-up stats and HCB data. + Performs batch update of YSWS programs in Airtable with sign-up stats (true-spend costs have a separate post-dbt writer). """ log = context.log airtable: AirtableResource = context.resources.airtable @@ -2443,8 +2228,8 @@ def airtable_unified_ysws_db_daily_parquet_backup( approved_projects_mention_search_batch, ysws_programs_sign_up_stats_candidates, ysws_programs_sign_up_stats, - ysws_programs_hcb_candidates, ysws_programs_hcb_stats, + ysws_programs_true_spend_update_status, ysws_programs_prepared_for_update, ysws_programs_update_status, unified_ysws_db_processing_done, diff --git a/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py new file mode 100644 index 0000000..44ec50c --- /dev/null +++ b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py @@ -0,0 +1,143 @@ +"""Sync the website's canonical HCB-root costs, without redoing its accounting. + +Several Airtable program versions can share a root. Each gets the same root +spend, hours and rate shown on the website, NOT an invented per-version split. +Keep this writer downstream of dbt and outside the pre-warehouse processing +marker: feeding it back into that marker would create a Dagster asset cycle. +""" +from contextlib import closing +from datetime import datetime, timezone +import os +from urllib.parse import urlparse + +import polars as pl +import psycopg2 +from dagster import AssetKey, AssetExecutionContext, Output, asset + +from ..airtable.generated_ids import AirtableIDs + +PROGRAMS = AirtableIDs.unified_ysws_projects_db.ysws_programs +SPEND_FIELD = PROGRAMS.total_spent_from_hcb_fund +# Airtable accepts field names in updates. Resolve these approved schema fields +# at execution time so no fabricated field IDs enter the generated IDs file. +HOURS_FIELD = "True Spend — Weighted Hours" +RATE_FIELD = "True Spend — Cost Per Weighted Hour" +SYNC_FIELDS = (SPEND_FIELD, HOURS_FIELD, RATE_FIELD) +SOURCE_ASSET = AssetKey(["hcb_ysws_true_spend_analytics", "ysws_spend_by_program"]) +SOURCE_SQL = """ +SELECT root_slug, member_ids, true_spend_dollars, weighted_hours, + cost_per_weighted_hour +FROM public_hcb_ysws_true_spend_analytics.ysws_spend_by_program +WHERE is_ysws_program +""" + + +def _root_slug(url): + if not isinstance(url, str): + return None + parsed = urlparse(url) + if parsed.scheme != "https" or parsed.netloc != "hcb.hackclub.com": + return None + return parsed.path.strip("/").split("/")[0] or None + + +def build_true_spend_updates(programs: pl.DataFrame, costs: pl.DataFrame) -> pl.DataFrame: + """Exact record-ID + current HCB-link match; unknowns clear stale values. + + Zero spend is a real zero. Missing/zero hours keep the mart's NULL rate. + No gross-outflow fallback, postage estimate, or local rate calculation. + """ + if costs.is_empty(): + raise ValueError("True-spend source is empty; refusing to clear all Airtable costs") + if programs["id"].n_unique() != programs.height: + raise ValueError("Duplicate Airtable program IDs") + by_id = {} + for cost in costs.iter_rows(named=True): + for member in cost["member_ids"] or []: + if member in by_id: + raise ValueError("Program belongs to multiple canonical HCB roots") + by_id[member] = cost + rows = [] + for program in programs.iter_rows(named=True): + cost = by_id.get(program["id"]) + if cost is not None and _root_slug(program.get(PROGRAMS.hcb)) != cost["root_slug"]: + cost = None # Link changed since the warehouse snapshot: never use the old root. + rows.append({ + "id": program["id"], + SPEND_FIELD: float(cost["true_spend_dollars"]) if cost else None, + HOURS_FIELD: float(cost["weighted_hours"]) if cost and cost["weighted_hours"] is not None else None, + RATE_FIELD: float(cost["cost_per_weighted_hour"]) if cost and cost["cost_per_weighted_hour"] is not None else None, + }) + return pl.DataFrame(rows, schema={"id": pl.String, **{f: pl.Float64 for f in SYNC_FIELDS}}) + + +@asset( + group_name="ysws_true_spend_sync", + description="Read canonical true spend and weighted-hour cost from the same mart as the website.", + compute_kind="warehouse_query", + deps=[SOURCE_ASSET], + required_resource_keys={"airtable"}, +) +def ysws_programs_hcb_stats(context: AssetExecutionContext) -> Output[pl.DataFrame]: + event = context.instance.get_latest_materialization_event(SOURCE_ASSET) + if event is None or datetime.now(timezone.utc).timestamp() - event.timestamp > 36 * 3600: + raise ValueError("True-spend mart has no successful materialization within 36 hours") + programs = context.resources.airtable.get_all_records_as_polars( + context=context, base_key="unified_ysws_projects_db", table_key="ysws_programs", + ) + with closing(psycopg2.connect( + os.environ["WAREHOUSE_COOLIFY_URL"], connect_timeout=15, + options="-c statement_timeout=120000 -c lock_timeout=10000", + )) as conn: + costs = pl.read_database(SOURCE_SQL, conn) + updates = build_true_spend_updates(programs, costs) + return Output(updates, metadata={ + "programs": updates.height, + "matched": updates[SPEND_FIELD].is_not_null().sum(), + "missing_mapping": updates[SPEND_FIELD].is_null().sum(), + "source_materialized_at": datetime.fromtimestamp(event.timestamp, timezone.utc).isoformat(), + "source": "public_hcb_ysws_true_spend_analytics.ysws_spend_by_program", + }) + + +def write_true_spend_updates(table, updates: pl.DataFrame) -> int: + schema = table.schema(force=True) + fields = {f.name: f.id for f in schema.fields} + for name in (HOURS_FIELD, RATE_FIELD): + if name not in fields: + raise ValueError(f"Create the approved Airtable field first: {name}") + by_id = {f.id: f for f in schema.fields} + expected_references = { + PROGRAMS.cost_per_hour: {fields[HOURS_FIELD], fields[RATE_FIELD]}, + PROGRAMS.total_spend: {SPEND_FIELD}, + } + for field_id, expected in expected_references.items(): + field = by_id.get(field_id) + actual = set(getattr(getattr(field, "options", None), "referenced_field_ids", None) or []) + if actual != expected: + raise ValueError("Update Airtable Cost Per Hour and Total Spend formulas to canonical true spend before syncing") + # Explicit nulls are intentional clears. The generic resource helper drops + # None, which would leave old rates after hours or HCB mappings disappear. + records = [{"id": row["id"], "fields": { + fields.get(k, k): row[k] for k in SYNC_FIELDS + }} for row in updates.iter_rows(named=True)] + if not records: + return 0 + result = table.batch_update(records) + if len(result) != len(records): + raise RuntimeError("Incomplete true-spend Airtable update") + return len(result) + + +@asset( + group_name="ysws_true_spend_sync", + description="Update Airtable costs after true-spend dbt, separate from pre-dbt YSWS processing.", + compute_kind="airtable_update", + required_resource_keys={"airtable"}, +) +def ysws_programs_true_spend_update_status( + context: AssetExecutionContext, ysws_programs_hcb_stats: pl.DataFrame, +) -> Output[None]: + table = context.resources.airtable.get_table("unified_ysws_projects_db", "ysws_programs") + count = write_true_spend_updates(table, ysws_programs_hcb_stats) + return Output(None, metadata={"updates_successful": count}) diff --git a/orpheus_engine/schedules.py b/orpheus_engine/schedules.py index 9dd214f..4cc6a2b 100644 --- a/orpheus_engine/schedules.py +++ b/orpheus_engine/schedules.py @@ -85,6 +85,7 @@ def _evaluation_fn(context: dg.ScheduleEvaluationContext): # Unified YSWS refresh, daily Parquet backup, and warehouse assets UNIFIED_YSWS_SELECTION = ( + dg.AssetSelection.groups("ysws_true_spend_sync") | dg.AssetSelection.groups("airtable_unified_ysws_projects_db_refresh") | dg.AssetSelection.groups("dlt_airtable_unified_ysws") | dg.AssetSelection.assets("unified_ysws_ysws_programs_weighted_referral_count") diff --git a/orpheus_engine_tests/test_ysws_true_spend_sync.py b/orpheus_engine_tests/test_ysws_true_spend_sync.py new file mode 100644 index 0000000..159391f --- /dev/null +++ b/orpheus_engine_tests/test_ysws_true_spend_sync.py @@ -0,0 +1,118 @@ +"""Synthetic fixtures only: cost sync must preserve the website's accounting.""" +from types import SimpleNamespace + +import polars as pl +import pytest + +from orpheus_engine.defs.unified_ysws_db.true_spend_sync import ( + PROGRAMS, SPEND_FIELD, HOURS_FIELD, RATE_FIELD, SOURCE_ASSET, + build_true_spend_updates, write_true_spend_updates, ysws_programs_hcb_stats, +) + + +def costs(**overrides): + row = dict(root_slug="synthetic-root", member_ids=["record-a", "record-b"], + true_spend_dollars=120.25, weighted_hours=30.0, cost_per_weighted_hour=4.01) + row.update(overrides) + return pl.DataFrame([row]) + + +def programs(*ids, url="https://hcb.hackclub.com/synthetic-root"): + return pl.DataFrame({"id": ids, PROGRAMS.hcb: [url] * len(ids)}) + + +def test_shared_root_copies_canonical_rate_without_reallocating_or_recalculating(): + rows = build_true_spend_updates(programs("record-a", "record-b"), costs()).to_dicts() + assert len(rows) == 2 + for r in rows: + assert r[SPEND_FIELD] == 120.25 + assert r[HOURS_FIELD] == 30.0 + assert r[RATE_FIELD] == 4.01 # Direct rounded mart value, not 120.25 / 30. + + +def test_unknown_program_is_null_not_zero_or_old_gross_spend(): + r = build_true_spend_updates(programs("unmatched"), costs()).row(0, named=True) + assert all(r[f] is None for f in (SPEND_FIELD, HOURS_FIELD, RATE_FIELD)) + + +@pytest.mark.parametrize('url', [None, '', 'https://example.invalid/synthetic-root', + 'https://hcb.hackclub.com/changed-root']) +def test_removed_or_remapped_hcb_link_clears_old_values(url): + r = build_true_spend_updates(programs("record-a", url=url), costs()).row(0, named=True) + assert r[SPEND_FIELD] is None + assert r[RATE_FIELD] is None + + +def test_transaction_url_matches_root_and_record_id_not_program_name(): + r = build_true_spend_updates(programs("record-a", url="https://hcb.hackclub.com/synthetic-root/transactions?x=1"), costs()) + assert r[SPEND_FIELD][0] == 120.25 + + +def test_zero_spend_and_null_rate_are_preserved(): + r = build_true_spend_updates(programs("record-a"), costs(true_spend_dollars=0, weighted_hours=0, cost_per_weighted_hour=None)) + assert r[SPEND_FIELD][0] == 0 + assert r[RATE_FIELD][0] is None + + +def test_empty_source_fails_without_clearing_everything(): + with pytest.raises(ValueError, match="source is empty"): + build_true_spend_updates(programs("record-a"), costs().head(0)) + + +def test_duplicate_source_mapping_fails_instead_of_double_counting(): + with pytest.raises(ValueError, match="multiple canonical"): + build_true_spend_updates(programs("record-a"), pl.concat([costs(), costs()])) + + +def test_duplicate_airtable_ids_fail(): + with pytest.raises(ValueError, match="Duplicate Airtable"): + build_true_spend_updates(programs("record-a", "record-a"), costs()) + + +def test_dependency_is_canonical_mart(): + assert SOURCE_ASSET in ysws_programs_hcb_stats.asset_deps[ysws_programs_hcb_stats.key] + + +class FakeTable: + def __init__(self, missing=False): + self.records = None + self.missing = missing + def schema(self, force=False): + assert force + return SimpleNamespace(fields=[] if self.missing else [ + SimpleNamespace(name=HOURS_FIELD, id="synthetic-hours-field"), + SimpleNamespace(name=RATE_FIELD, id="synthetic-rate-field"), + SimpleNamespace(name="Cost Per Hour", id=PROGRAMS.cost_per_hour, + options=SimpleNamespace(referenced_field_ids=["synthetic-hours-field", "synthetic-rate-field"])), + SimpleNamespace(name="Total Spend", id=PROGRAMS.total_spend, + options=SimpleNamespace(referenced_field_ids=[SPEND_FIELD])), + ]) + def batch_update(self, records): + self.records = records + return records + + +def test_writer_preserves_explicit_nulls_and_only_writes_owned_fields(): + table = FakeTable() + updates = build_true_spend_updates(programs("unmatched"), costs()) + assert write_true_spend_updates(table, updates) == 1 + assert table.records == [{"id": "unmatched", "fields": { + SPEND_FIELD: None, "synthetic-hours-field": None, "synthetic-rate-field": None, + }}] + + +def test_missing_schema_blocks_writes(): + table = FakeTable(missing=True) + with pytest.raises(ValueError, match="approved Airtable field"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None + + +def test_legacy_formulas_block_cutover_before_any_writes(): + table = FakeTable() + schema = table.schema(force=True) + schema.fields[-1].options.referenced_field_ids.append("synthetic-postage-field") + table.schema = lambda **kwargs: schema + with pytest.raises(ValueError, match="formulas"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None From 0575e6e993efa9f9c9a54e1140736ecff45eedd5 Mon Sep 17 00:00:00 2001 From: Zach Latta Date: Thu, 10 Sep 2026 14:43:57 -0400 Subject: [PATCH 2/2] Replace existing YSWS spend and cost fields in place --- docs/ysws-true-spend-airtable-cutover.md | 26 +++++++++------- .../defs/unified_ysws_db/true_spend_sync.py | 31 +++++++------------ .../test_ysws_true_spend_sync.py | 27 ++++++++++------ 3 files changed, 42 insertions(+), 42 deletions(-) diff --git a/docs/ysws-true-spend-airtable-cutover.md b/docs/ysws-true-spend-airtable-cutover.md index 88e4799..61d0c54 100644 --- a/docs/ysws-true-spend-airtable-cutover.md +++ b/docs/ysws-true-spend-airtable-cutover.md @@ -10,18 +10,20 @@ that the current HCB link still matches the snapshot's root. ## One-time Airtable schema change -In Unified YSWS Projects DB → YSWS Programs, add: +Replace the existing columns in place. **Do not create additional fields.** +The previous two-field approval request is superseded and should not be approved. -- `True Spend — Weighted Hours`: number (4 decimal display places). -- `True Spend — Cost Per Weighted Hour`: currency (2 decimal places). +Keep `Total Spent From HCB Fund` as the writable spend column; it now receives +canonical **true spend**, not gross HCB outflow. -Keep `Total Spent From HCB Fund` as the writable spend column for compatibility; -it now receives canonical **true spend**, not gross HCB outflow. - -Change the existing **Cost Per Hour** formula (keep its field ID): +Convert the existing **Cost Per Hour** field from formula to currency (2 decimal +places), preserving its name and field ID. The pipeline writes the mart's +`cost_per_weighted_hour` directly into this field. This keeps shared-root rates +identical to the website without changing existing project-hour rollups. +Before conversion, save the old formula for rollback: ```text -IF({True Spend — Weighted Hours} > 0, {True Spend — Cost Per Weighted Hour}, BLANK()) +IF(AND({Total Spend}, {Weighted–Total} > 0), {Total Spend} / {Weighted–Total} / 10) ``` Change the existing **Total Spend** formula (keep its field ID): @@ -36,14 +38,14 @@ longer add the estimate on top of the reconciled ledger. ## Deployment / verification -1. Create the two fields through the reviewed schema-change path. +1. Confirm browser access to edit the existing Airtable fields. 2. Deploy the code: the legacy gross-spend writer is removed. The new writer - refuses writes until the existing formula dependencies have been updated. -3. Update the two formulas above in Airtable's UI (the API cannot edit formulas). + refuses writes until Cost Per Hour is writable and Total Spend no longer adds postage. +3. Convert Cost Per Hour and update Total Spend in Airtable's UI (the API cannot change field types or edit formulas). 4. Materialize `ysws_programs_hcb_stats` and `ysws_programs_true_spend_update_status`. No unrelated signup or project processing needs to run for this cutover. -5. Reconcile every matched record's spend, hours and displayed cost/hour against +5. Reconcile every matched record's spend and displayed cost/hour against its canonical mart row. Check zero spend, zero/missing hours, removed HCB links, and shared roots. Unknown mappings clear old derived numbers to blank; they never fall back to gross spend. diff --git a/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py index 44ec50c..4c5d2ed 100644 --- a/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py +++ b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py @@ -18,11 +18,9 @@ PROGRAMS = AirtableIDs.unified_ysws_projects_db.ysws_programs SPEND_FIELD = PROGRAMS.total_spent_from_hcb_fund -# Airtable accepts field names in updates. Resolve these approved schema fields -# at execution time so no fabricated field IDs enter the generated IDs file. -HOURS_FIELD = "True Spend — Weighted Hours" -RATE_FIELD = "True Spend — Cost Per Weighted Hour" -SYNC_FIELDS = (SPEND_FIELD, HOURS_FIELD, RATE_FIELD) +# Replace the existing columns in place; preserve their IDs and consumers. +RATE_FIELD = PROGRAMS.cost_per_hour +SYNC_FIELDS = (SPEND_FIELD, RATE_FIELD) SOURCE_ASSET = AssetKey(["hcb_ysws_true_spend_analytics", "ysws_spend_by_program"]) SOURCE_SQL = """ SELECT root_slug, member_ids, true_spend_dollars, weighted_hours, @@ -65,7 +63,6 @@ def build_true_spend_updates(programs: pl.DataFrame, costs: pl.DataFrame) -> pl. rows.append({ "id": program["id"], SPEND_FIELD: float(cost["true_spend_dollars"]) if cost else None, - HOURS_FIELD: float(cost["weighted_hours"]) if cost and cost["weighted_hours"] is not None else None, RATE_FIELD: float(cost["cost_per_weighted_hour"]) if cost and cost["cost_per_weighted_hour"] is not None else None, }) return pl.DataFrame(rows, schema={"id": pl.String, **{f: pl.Float64 for f in SYNC_FIELDS}}) @@ -102,24 +99,18 @@ def ysws_programs_hcb_stats(context: AssetExecutionContext) -> Output[pl.DataFra def write_true_spend_updates(table, updates: pl.DataFrame) -> int: schema = table.schema(force=True) - fields = {f.name: f.id for f in schema.fields} - for name in (HOURS_FIELD, RATE_FIELD): - if name not in fields: - raise ValueError(f"Create the approved Airtable field first: {name}") by_id = {f.id: f for f in schema.fields} - expected_references = { - PROGRAMS.cost_per_hour: {fields[HOURS_FIELD], fields[RATE_FIELD]}, - PROGRAMS.total_spend: {SPEND_FIELD}, - } - for field_id, expected in expected_references.items(): - field = by_id.get(field_id) - actual = set(getattr(getattr(field, "options", None), "referenced_field_ids", None) or []) - if actual != expected: - raise ValueError("Update Airtable Cost Per Hour and Total Spend formulas to canonical true spend before syncing") + for field_id in SYNC_FIELDS: + if getattr(by_id.get(field_id), "type", None) not in ("currency", "number"): + raise ValueError("Existing spend and Cost Per Hour columns must be writable currency fields before syncing") + field = by_id.get(PROGRAMS.total_spend) + actual = set(getattr(getattr(field, "options", None), "referenced_field_ids", None) or []) + if actual != {SPEND_FIELD}: + raise ValueError("Update existing Total Spend formula to use only true spend before syncing") # Explicit nulls are intentional clears. The generic resource helper drops # None, which would leave old rates after hours or HCB mappings disappear. records = [{"id": row["id"], "fields": { - fields.get(k, k): row[k] for k in SYNC_FIELDS + k: row[k] for k in SYNC_FIELDS }} for row in updates.iter_rows(named=True)] if not records: return 0 diff --git a/orpheus_engine_tests/test_ysws_true_spend_sync.py b/orpheus_engine_tests/test_ysws_true_spend_sync.py index 159391f..f9b8d15 100644 --- a/orpheus_engine_tests/test_ysws_true_spend_sync.py +++ b/orpheus_engine_tests/test_ysws_true_spend_sync.py @@ -5,7 +5,7 @@ import pytest from orpheus_engine.defs.unified_ysws_db.true_spend_sync import ( - PROGRAMS, SPEND_FIELD, HOURS_FIELD, RATE_FIELD, SOURCE_ASSET, + PROGRAMS, SPEND_FIELD, RATE_FIELD, SOURCE_ASSET, build_true_spend_updates, write_true_spend_updates, ysws_programs_hcb_stats, ) @@ -26,13 +26,12 @@ def test_shared_root_copies_canonical_rate_without_reallocating_or_recalculating assert len(rows) == 2 for r in rows: assert r[SPEND_FIELD] == 120.25 - assert r[HOURS_FIELD] == 30.0 assert r[RATE_FIELD] == 4.01 # Direct rounded mart value, not 120.25 / 30. def test_unknown_program_is_null_not_zero_or_old_gross_spend(): r = build_true_spend_updates(programs("unmatched"), costs()).row(0, named=True) - assert all(r[f] is None for f in (SPEND_FIELD, HOURS_FIELD, RATE_FIELD)) + assert all(r[f] is None for f in (SPEND_FIELD, RATE_FIELD)) @pytest.mark.parametrize('url', [None, '', 'https://example.invalid/synthetic-root', @@ -80,10 +79,8 @@ def __init__(self, missing=False): def schema(self, force=False): assert force return SimpleNamespace(fields=[] if self.missing else [ - SimpleNamespace(name=HOURS_FIELD, id="synthetic-hours-field"), - SimpleNamespace(name=RATE_FIELD, id="synthetic-rate-field"), - SimpleNamespace(name="Cost Per Hour", id=PROGRAMS.cost_per_hour, - options=SimpleNamespace(referenced_field_ids=["synthetic-hours-field", "synthetic-rate-field"])), + SimpleNamespace(name="Total Spent From HCB Fund", id=SPEND_FIELD, type="currency"), + SimpleNamespace(name="Cost Per Hour", id=RATE_FIELD, type="currency"), SimpleNamespace(name="Total Spend", id=PROGRAMS.total_spend, options=SimpleNamespace(referenced_field_ids=[SPEND_FIELD])), ]) @@ -97,13 +94,13 @@ def test_writer_preserves_explicit_nulls_and_only_writes_owned_fields(): updates = build_true_spend_updates(programs("unmatched"), costs()) assert write_true_spend_updates(table, updates) == 1 assert table.records == [{"id": "unmatched", "fields": { - SPEND_FIELD: None, "synthetic-hours-field": None, "synthetic-rate-field": None, + SPEND_FIELD: None, RATE_FIELD: None, }}] def test_missing_schema_blocks_writes(): table = FakeTable(missing=True) - with pytest.raises(ValueError, match="approved Airtable field"): + with pytest.raises(ValueError, match="writable currency"): write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) assert table.records is None @@ -113,6 +110,16 @@ def test_legacy_formulas_block_cutover_before_any_writes(): schema = table.schema(force=True) schema.fields[-1].options.referenced_field_ids.append("synthetic-postage-field") table.schema = lambda **kwargs: schema - with pytest.raises(ValueError, match="formulas"): + with pytest.raises(ValueError, match="formula"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None + + +def test_old_cost_formula_blocks_writes_until_in_place_conversion(): + table = FakeTable() + schema = table.schema(force=True) + schema.fields[1].type = "formula" + table.schema = lambda **kwargs: schema + with pytest.raises(ValueError, match="writable currency"): write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) assert table.records is None