Skip to content

Use canonical true spend for YSWS Airtable cost/hour #69

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
zachlatta wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Use canonical true spend for YSWS Airtable cost/hour #69

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
29f2a47
Sync YSWS costs directly from the canonical true-spend mart
zachlatta Sep 10, 2026
0575e6e
Replace existing YSWS spend and cost fields in place
zachlatta Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 61 additions & 0 deletions docs/ysws-true-spend-airtable-cutover.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# 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

Replace the existing columns in place. **Do not create additional fields.**
The previous two-field approval request is superseded and should not be approved.

Keep `Total Spent From HCB Fund` as the writable spend column; it now receives
canonical **true spend**, not gross HCB outflow.

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(AND({Total Spend}, {Weighted–Total} > 0), {Total Spend} / {Weighted–Total} / 10)
```

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. 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 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 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.
231 changes: 8 additions & 223 deletions orpheus_engine/defs/unified_ysws_db/definitions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand All @@ -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()]

Expand Down Expand Up @@ -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
}
Expand All @@ -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",
)
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Loading