diff --git a/src/rid_reservoir.py b/src/rid_reservoir.py index a7bd584..4f85446 100644 --- a/src/rid_reservoir.py +++ b/src/rid_reservoir.py @@ -28,6 +28,22 @@ logger = logging.getLogger(__name__) RID_DAMS_URL = "https://app.rid.go.th/reservoir/api/dams" MAE_NGAT_DAM_ID = "200103" +# Per-column NUMERIC capacity; source junk beyond these becomes NULL instead +# of overflowing the insert and discarding the whole daily batch. +_MEASURE_BOUNDS = { + "storage_mcm": 1e8, + "storage_pct": 1e6, + "inflow_mcm": 1e8, + "outflow_mcm": 1e8, + "level_msl": 1e6, +} + + +def _bounded(value: Optional[float], limit: float) -> Optional[float]: + if value is not None and abs(value) >= limit: + return None + return value + def _to_float(value: Any) -> Optional[float]: """API numerics arrive as strings ('222.01'), ' - ' placeholders, or None.""" @@ -151,7 +167,7 @@ class RidReservoirStore: dam_id VARCHAR(10) NOT NULL, date DATE NOT NULL, storage_mcm NUMERIC(10,2), - storage_pct NUMERIC(6,2), + storage_pct NUMERIC(8,2), inflow_mcm NUMERIC(10,2), outflow_mcm NUMERIC(10,2), level_msl NUMERIC(8,2), @@ -168,6 +184,26 @@ class RidReservoirStore: with self.engine.begin() as conn: for statement in ddl: conn.execute(text(statement)) + # Widen storage_pct on tables created before 2026-08-13: the source + # publishes junk percents (dam 100602 reports 87798%) that overflowed + # NUMERIC(6,2) and discarded whole daily batches. + if self.db_type == "postgresql": + migrations = ( + "ALTER TABLE rid_reservoir_daily " + "ALTER COLUMN storage_pct TYPE NUMERIC(8,2)", + ) + elif self.db_type == "mysql": + migrations = ( + "ALTER TABLE rid_reservoir_daily MODIFY storage_pct NUMERIC(8,2)", + ) + else: # sqlite: NUMERIC is affinity only, nothing to widen + migrations = () + for statement in migrations: + try: + with self.engine.begin() as conn: + conn.execute(text(statement)) + except Exception as e: + logger.warning(f"rid_reservoir_daily migration skipped: {e}") def _upsert(self, table: str, key_cols: List[str], value_cols: List[str]) -> str: cols = key_cols + value_cols @@ -222,7 +258,9 @@ class RidReservoirStore: dam_row = {c: record.get(c) for c in dam_cols} dam_row.update({"dam_id": record["dam_id"], "updated_at": now}) dams[record["dam_id"]] = dam_row - measure_row = {c: record.get(c) for c in measure_cols} + measure_row = { + c: _bounded(record.get(c), _MEASURE_BOUNDS[c]) for c in measure_cols + } measure_row.update( {"dam_id": record["dam_id"], "date": record["date"]} ) diff --git a/tests/test_rid_reservoir.py b/tests/test_rid_reservoir.py index 8c03a94..fdcf69d 100644 --- a/tests/test_rid_reservoir.py +++ b/tests/test_rid_reservoir.py @@ -123,6 +123,35 @@ class TestStore: def test_save_empty(self, store): assert store.save([]) == 0 + def test_junk_source_values_are_nulled_not_fatal(self, store): + # Real junk from 2019-01-05: dam 100602 reported 87798% storage, + # which overflowed NUMERIC(6,2) and discarded the whole 33-dam batch. + payload = _dams_payload() + payload["regions"][0]["dams"].append( + { + "DAM_ID": "100602", + "DAM_Name": "junk", + "DMD_Date": "2026-08-13", + "DMD_QUse": "343292.00", + "PERCENT_DMD_QUse": "87798.00", + "DMD_Inflow": "0.59", + "DMD_Outflow": "1e12", # beyond NUMERIC(10,2) -> NULL + } + ) + assert store.save(parse_dam_records(payload)) == 3 + from sqlalchemy import text + + with store.engine.begin() as conn: + row = conn.execute( + text( + "SELECT storage_mcm, storage_pct, outflow_mcm " + "FROM rid_reservoir_daily WHERE dam_id = '100602'" + ) + ).fetchone() + assert float(row[0]) == 343292.00 # fits NUMERIC(10,2), kept raw + assert float(row[1]) == 87798.00 # fits widened NUMERIC(8,2) + assert row[2] is None # beyond capacity -> NULL, batch survives + def test_present_dates(self, store): lo, hi = datetime.date(2026, 8, 1), datetime.date(2026, 8, 31) assert store.present_dates(lo, hi) == set()