diff --git a/get_data/get_EEA_data_portal.py b/get_data/get_EEA_data_portal.py index df645f0..e6e8b8c 100644 --- a/get_data/get_EEA_data_portal.py +++ b/get_data/get_EEA_data_portal.py @@ -8,10 +8,36 @@ uploaded to GCS; an annualized summary is kept locally. HTTP requests use a session with automatic retries (5 attempts, exponential backoff) to -tolerate transient 5xx / 429 errors from the portal. +tolerate transient 5xx / 429 errors from the portal. Note: 500 is excluded from the +retry list because the API returns 500 to signal "no records match filter" rather than +as a genuine server error — retrying would just waste time. CSO data is handled separately in get_eea_dp_cso.py because it uses a different API. +Incremental fetching +-------------------- +Three tables support date-range filtering via query parameters discovered in April 2026: + - drinkingWater: FromCollectedDate=YYYY-MM-DD (date col: CollectedDate) + - inspection: FromInspectionDate=YYYY-MM-DD (date col: InspectionDate) + - enforcement: FromEnforcementDate=YYYY-MM-DD (date col: EnforcementDate) + +For these tables we load the existing local CSV (or GCS copy for drinkingWater), find +the max date, and fetch only records from that date onward (inclusive, to catch records +submitted after the previous pull). Rows on the boundary date are dropped from the +cache before merging to avoid duplicates. A full fetch is used when no cache exists. + +The API returns HTTP 500 instead of an empty list when a date filter matches zero records. +We treat a non-OK response on the initial TotalCount request as "no new records" and +fall back to the existing cache unchanged. + +permit and facility have no discovered date filter and are always fetched in full. + +drinkingWater GCS flow +---------------------- +Because the full drinkingWater CSV lives only in GCS (too large to commit), the incremental +flow is: gsutil cp (download) → append new rows → gsutil cp (re-upload). If the download +fails (first run or GCS unavailable), we fall back to a full API fetch. + Outputs (per table, e.g. 'permit'): ../docs/data/EEADP_permit.csv — full table ../docs/data/EEADP_permit_sample.csv — 10-row sample @@ -35,113 +61,180 @@ API_ROOT = 'http://eeaonline.eea.state.ma.us/EEA/DataLake/V1.0/DataLakeAPI/' API_TABLES = ['permit', 'facility', 'inspection', 'enforcement', 'drinkingWater'] +# Tables with date-filter support: {table: (filter_param, date_col_in_csv)} +INCREMENTAL_TABLES = { + 'inspection': ('FromInspectionDate', 'InspectionDate'), + 'enforcement': ('FromEnforcementDate', 'EnforcementDate'), + 'drinkingWater': ('FromCollectedDate', 'CollectedDate'), +} ########################## ## Function definitions ########################## -# Generic user agent -REQ_HEADER = {'User-Agent': - 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36'} +REQ_HEADER = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36', + 'Referer': 'https://eeaonline.eea.state.ma.us/Portal/', +} def _make_session() -> requests.Session: - """Return a requests Session with automatic retries on transient errors.""" + """Return a requests Session with automatic retries on transient errors. + + 500 is intentionally excluded from status_forcelist: this API returns 500 + to mean "no records match filter", so retrying wastes time. + """ session = requests.Session() retry = Retry( total=5, backoff_factor=2, - status_forcelist=[429, 500, 502, 503, 504], + status_forcelist=[429, 502, 503, 504], ) session.mount('http://', HTTPAdapter(max_retries=retry)) session.mount('https://', HTTPAdapter(max_retries=retry)) return session -def query_iterate(table_name: str, req_size: int=100000, verbose: bool=True): - """ - Query the EEA data portal to retrieve the entirety of a data table. - Returns +def _get_max_date(csv_path: str, date_col: str) -> str | None: + """Return the max date in date_col of csv_path as YYYY-MM-DD, or None.""" + try: + df = pd.read_csv(csv_path, usecols=[date_col]) + max_val = pd.to_datetime(df[date_col], errors='coerce').max() + if pd.isna(max_val): + return None + return max_val.strftime('%Y-%m-%d') + except (FileNotFoundError, ValueError, KeyError): + return None + + +def query_iterate(table_name: str, req_size: int = 100000, verbose: bool = True, + filter_param: str | None = None, filter_val: str | None = None) -> pd.DataFrame: + """Query the EEA DataLake API, returning the full (or date-filtered) table. Args: - table_name (str): EEA data portal table to query - req_size (int): Request chunksize - verbose (bool): Print chunk position while iterating + table_name: DataLake table name (e.g. 'drinkingWater') + req_size: Rows per paginated request + verbose: Print progress + filter_param: Optional date-filter query param name (e.g. 'FromCollectedDate') + filter_val: Optional date-filter value in YYYY-MM-DD format Returns: - df: Pandas DataFrame with table contents + DataFrame of matching rows, or empty DataFrame when filter matches nothing. """ session = _make_session() - # Get total table size - try: - r = session.get(API_ROOT + table_name + '?_end=1&_start=0', headers=REQ_HEADER) - table_size = r.json()['TotalCount'] - except ValueError: - raise ValueError("EEA Data Portal request returned error " + str(r.status_code) + '; perhaps table name is not valid\n\nFull response message:\n' + r.text) + filter_qs = f'&{filter_param}={filter_val}' if filter_param and filter_val else '' + mode = f'filtered {filter_param}={filter_val}' if filter_qs else 'full' + print(f'{table_name}: {mode} fetch') - # Iterate through requests - if (table_size < req_size): + # Get total row count (with filter applied). + size_url = API_ROOT + table_name + '?_end=1&_start=0' + filter_qs + r = session.get(size_url, headers=REQ_HEADER, timeout=120) + if not r.ok or 'TotalCount' not in r.json(): + # API returns 500 with no 'TotalCount' when filter matches zero records. + print(f' {table_name}: no records match filter (HTTP {r.status_code}); returning empty') + return pd.DataFrame() + + table_size = r.json()['TotalCount'] + if table_size == 0: + return pd.DataFrame() + print(f' {table_name}: {table_size:,} rows to fetch') + + if table_size < req_size: req_bins = [0, table_size] else: - max_bin = table_size + req_size - req_bins = np.arange(0, max_bin, req_size) + req_bins = list(np.arange(0, table_size + req_size, req_size)) + dfs = [] for i in range(len(req_bins) - 1): - # Log output - if verbose: print(table_name + ': request ' + str(i + 1) + ' of ' + str(len(req_bins) - 1)) - # Make request - url = API_ROOT + table_name + '?_end=' + str(req_bins[i+1]) + '&_start='+str(req_bins[i]) - r = session.get(url, headers=REQ_HEADER) - # Add chunk contents to dataframe list - dfs += [pd.DataFrame(r.json()['Items'])] + if verbose: + print(f'{table_name}: request {i + 1} of {len(req_bins) - 1}') + url = (API_ROOT + table_name + + f'?_end={req_bins[i+1]}&_start={req_bins[i]}' + filter_qs) + r = session.get(url, headers=REQ_HEADER, timeout=180) + r.raise_for_status() + dfs.append(pd.DataFrame(r.json()['Items'])) - # Concatenate chunks - df = pd.concat(dfs) + return pd.concat(dfs, ignore_index=True) - return df -def main(): - """Query for data, persist it, and report the update +def fetch_incremental(table_name: str, csv_path: str, filter_param: str, + date_col: str) -> tuple[pd.DataFrame, bool]: + """Load cached CSV and fetch only records newer than the max cached date. + + Returns (dataframe, is_incremental). is_incremental=False means we fell + back to a full fetch because no cache was available. """ - # Get data - ## Query data for each table + max_date = _get_max_date(csv_path, date_col) + if max_date is None: + print(f' {table_name}: no cache found; running full fetch') + return query_iterate(table_name), False + + print(f' {table_name}: cache through {max_date}; fetching from {max_date} (inclusive)') + new_data = query_iterate(table_name, filter_param=filter_param, filter_val=max_date) + + existing = pd.read_csv(csv_path) + # Drop boundary-date rows from cache — they're included in the fresh fetch. + existing[date_col] = pd.to_datetime(existing[date_col], errors='coerce') + cutoff = pd.to_datetime(max_date).date() + existing = existing[existing[date_col].dt.date < cutoff] + + if new_data.empty: + print(f' {table_name}: no new records; using cache as-is') + return pd.read_csv(csv_path), True + + combined = pd.concat([existing, new_data], ignore_index=True) + print(f' {table_name}: appended {len(new_data):,} new rows (total {len(combined):,})') + return combined, True + + +def main(): + """Query for data, persist it, and report the update.""" + table_data = {} - for tab in API_TABLES: + + # --- permit and facility: always full fetch (no date filter available) --- + for tab in ['permit', 'facility']: table_data[tab] = query_iterate(tab) - ## Write out, but treat large tables separately - ## Only one table (drinkingWater) is >10MB as of 08/2017, so we handle this as a special case. - ## Could also use `size_MB = os.path.getsize('../docs/data/EEADP_' + tab + '.csv')/1024/1024` to get file size + # --- inspection and enforcement: incremental via date filter --- + for tab in ['inspection', 'enforcement']: + filter_param, date_col = INCREMENTAL_TABLES[tab] + csv_path = f'../docs/data/EEADP_{tab}.csv' + table_data[tab], _ = fetch_incremental(tab, csv_path, filter_param, date_col) + + # --- drinkingWater: incremental via GCS cache + date filter --- + filter_param, date_col = INCREMENTAL_TABLES['drinkingWater'] + dw_local = 'EEADP_drinkingWater.csv' + gcs_path = f'gs://openamend-data/{dw_local}' + + print('drinkingWater: downloading existing data from GCS...') + gcs_rc = os.system(f'gsutil cp {gcs_path} {dw_local}') + if gcs_rc == 0 and os.path.exists(dw_local): + table_data['drinkingWater'], _ = fetch_incremental( + 'drinkingWater', dw_local, filter_param, date_col) + else: + print(' drinkingWater: GCS download failed; running full fetch') + table_data['drinkingWater'] = query_iterate('drinkingWater') + + # --- Write outputs --- for tab in API_TABLES: - ## Print a sample of the file as an example - table_data[tab].sample(n=10).to_csv('../docs/data/EEADP_' + tab + '_sample.csv', index=0) - - if tab != 'drinkingWater': - table_data[tab].to_csv('../docs/data/EEADP_' + tab + '.csv', index=0) + df = table_data[tab] + df.sample(n=min(10, len(df))).to_csv(f'../docs/data/EEADP_{tab}_sample.csv', index=False) + + if tab != 'drinkingWater': + df.to_csv(f'../docs/data/EEADP_{tab}.csv', index=False) else: - ## Send to Google object store - table_data[tab].to_csv('EEADP_' + tab + '.csv', encoding='utf-8', index=0) - os.system('gsutil cp EEADP_' + tab + '.csv gs://openamend-data/EEADP_' + tab + '.csv') - - ## Include some special summary statistics tables - ## --- - ## Most recent report for each chemical for each site - ## This still ends up being ~20% of the original size, so larger than desired - #table_data[tab].sort_values('CollectedDate', inplace=True) - #df_dw_last = table_data[tab].groupby(['ChemicalName','PWSName','LocationName']).last() - - # Tests per year per PWS per contaminant group per raw/finished - ## This still ends up being ~40% of the original size, so larger than desired - table_data[tab]['CollectedDate'] = pd.to_datetime(table_data[tab]['CollectedDate'], errors='coerce') - table_data[tab]['Year'] = table_data[tab]['CollectedDate'].apply(lambda x: x.year) - df_dw_annual_group = table_data[tab].groupby(['Year','PWSName', 'ContaminantGroup','RaworFinished']).agg({'Result': pd.Series.count}) - df_dw_annual_group.to_csv('../docs/data/EEADP_' + tab + '_annual.csv', index=1) - ## Print a sample of the file as an example - df_dw_annual_group.sample(n=10).to_csv('../docs/data/EEADP_' + tab + '_annual_sample.csv', index=1) - - ## Tests per year per PWS per chemical per raw/finished - ### This still ends up being ~40% of the original size, so larger than desired - #df_dw_annual = table_data[tab].groupby(['Year','PWSName', 'ChemicalName','RaworFinished']).agg({'ContaminantGroup': lambda x: x.iloc[0], 'Result': pd.Series.count}) + df.to_csv(dw_local, encoding='utf-8', index=False) + os.system(f'gsutil cp {dw_local} {gcs_path}') + + # Annualized summary + df['CollectedDate'] = pd.to_datetime(df['CollectedDate'], errors='coerce') + df['Year'] = df['CollectedDate'].dt.year + df_annual = df.groupby(['Year', 'PWSName', 'ContaminantGroup', 'RaworFinished']).agg( + {'Result': pd.Series.count}) + df_annual.to_csv('../docs/data/EEADP_drinkingWater_annual.csv', index=True) + df_annual.sample(n=min(10, len(df_annual))).to_csv( + '../docs/data/EEADP_drinkingWater_annual_sample.csv', index=True) # Archive PDF help files os.system('wget http://eeaonline.eea.state.ma.us/Portal/documents/General%20Query%20Search%20FAQs.pdf') @@ -149,9 +242,9 @@ def main(): os.system('wget http://eeaonline.eea.state.ma.us/Portal/documents/Terms%20and%20Definitions%20for%20EEA.pdf') os.system('mv "Terms and Definitions for EEA.pdf" ../docs/assets/PDFs/EEADP_Definitions.pdf') - # Report last update with open('../docs/data/ts_update_EEADP.yml', 'w') as f: - f.write('updated: '+str(datetime.datetime.now()).split('.')[0]+'\n') + f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') + if __name__ == '__main__': main() diff --git a/get_data/get_EPARegion1_NPDES_permits.py b/get_data/get_EPARegion1_NPDES_permits.py index c8c6e5d..1530637 100644 --- a/get_data/get_EPARegion1_NPDES_permits.py +++ b/get_data/get_EPARegion1_NPDES_permits.py @@ -245,7 +245,7 @@ permit_url = html.unescape(permit) os.system('wget ' + shlex.quote(permit_url) + ' --no-clobber --timeout=30 --tries=3 -O ' + shlex.quote(local_file)) if os.path.exists(local_file): - os.system('gsutil cp ' + local_file + ' gs://openamend-data/' + local_file) + os.system('gsutil cp ' + shlex.quote(local_file) + ' ' + shlex.quote('gs://openamend-data/' + local_file)) new_pdf_count += 1 else: out_files += [['']] diff --git a/get_data/get_MA_precipitation.py b/get_data/get_MA_precipitation.py index b93254a..ced4ece 100644 --- a/get_data/get_MA_precipitation.py +++ b/get_data/get_MA_precipitation.py @@ -76,10 +76,26 @@ def fetch_daily_precip_year(year: int) -> pd.DataFrame: def main(): current_year = datetime.datetime.now().year - all_years = list(range(START_YEAR, current_year + 1)) + out_path = '../docs/data/MA_precipitation_daily.csv' + + # Load cached data to determine the earliest year that needs re-fetching. + # We always re-fetch the most recently cached year because it may be incomplete + # (partial calendar year at the time of the previous run). + try: + existing = pd.read_csv(out_path) + existing['date'] = pd.to_datetime(existing['date']) + max_cached_year = existing['date'].dt.year.max() + fetch_from_year = max_cached_year # re-fetch last year in case it was partial + print(f' Found cached data through {max_cached_year}; fetching {fetch_from_year}–{current_year}') + except FileNotFoundError: + existing = None + fetch_from_year = START_YEAR + print(f' No cache found; fetching {START_YEAR}–{current_year}') + + fetch_years = list(range(fetch_from_year, current_year + 1)) rows = [] - for year in all_years: + for year in fetch_years: print(f' Fetching {year}...', end=' ', flush=True) try: df_year = fetch_daily_precip_year(year) @@ -88,11 +104,18 @@ def main(): except Exception as e: print(f'FAILED: {e}') - df = pd.concat(rows, ignore_index=True) + new_data = pd.concat(rows, ignore_index=True) + + if existing is not None: + # Drop cached rows for years being re-fetched, then append fresh data. + retained = existing[existing['date'].dt.year < fetch_from_year].copy() + retained['date'] = retained['date'].dt.strftime('%Y-%m-%d') + df = pd.concat([retained, new_data], ignore_index=True) + else: + df = new_data - out_path = '../docs/data/MA_precipitation_daily.csv' df.to_csv(out_path, index=False) - print(f'\nWrote {len(df)} rows to {out_path}') + print(f'\nWrote {len(df)} rows to {out_path} ({len(new_data)} newly fetched)') with open('../docs/data/ts_update_MA_precipitation.yml', 'w') as f: f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') diff --git a/get_data/get_budget_CTHRU.py b/get_data/get_budget_CTHRU.py index 572225c..eeeb8d6 100644 --- a/get_data/get_budget_CTHRU.py +++ b/get_data/get_budget_CTHRU.py @@ -54,6 +54,22 @@ def fetch_agency_budget(accounts: list) -> pd.DataFrame: if __name__ == '__main__': + # Skip if the cache already covers the current fiscal year. + # MA fiscal year runs July–June: FY2026 = Jul 2025–Jun 2026. + _now = datetime.datetime.now() + _current_fy = _now.year if _now.month < 7 else _now.year + 1 + try: + _cached = pd.read_csv('../docs/data/MassBudget_environmental_summary.csv') + _max_cached_fy = int(_cached['Year'].max()) + if _max_cached_fy >= _current_fy: + print(f'Budget data current through FY{_max_cached_fy}; skipping CTHRU fetch.') + with open('../docs/data/ts_update_MassBudget_environmental.yml', 'w') as _f: + _f.write('updated: ' + str(_now).split('.')[0] + '\n') + import sys; sys.exit(0) + print(f'Cache covers through FY{_max_cached_fy}; fetching FY{_max_cached_fy + 1}–FY{_current_fy}') + except FileNotFoundError: + print('No cache found; running full CTHRU fetch') + # Load SSA AWI from CSV for inflation adjustment (2024 base year) # Read from CSV instead of DB since this runs before assemble_db.py creates tables awi = pd.read_csv('../docs/data/SSAWages.csv').set_index('Year') diff --git a/get_data/get_eea_dp_cso.py b/get_data/get_eea_dp_cso.py index 5b176c8..329cf91 100644 --- a/get_data/get_eea_dp_cso.py +++ b/get_data/get_eea_dp_cso.py @@ -7,17 +7,26 @@ pagination and auth requirements. Key implementation notes: - - The CSOAPI requires a Referer header pointing to the portal page; bare requests + - The CSOAPI requires a Referer header matching the portal page; bare requests return HTTP 500. The REQ_HEADER below must be kept in sync with the portal URL. - The API is 1-indexed (pageNumber starts at 1, not 0). - Timestamps are ISO 8601 but may or may not include milliseconds; use format='ISO8601'. - The API returns a lowercase 'year' column; we drop it to avoid a case-insensitive name collision with our added 'Year' column when writing to SQLite. + - Date-filtered queries (IncidentFromDate) work correctly when records exist, but + the API returns HTTP 500 instead of an empty list when zero records match. We + treat a 500 on the first page of a filtered query as "no new records" and fall + back to the existing cache unchanged. + +Incremental fetching: + When a cached CSV exists, we load it, find the max incidentDate, and fetch only + records from that date onward (inclusive, to catch records that arrived after the + last pull). Cached rows on the boundary date are dropped before merging so there + are no duplicates. A full fetch is used when no cache exists. Example API URL: https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/ - ?ReporterClass=Verified%20Data%20Report&IncidentFromDate=01/01/2022 - &IncidentToDate=08/02/2023&RainfallDataFrom=1&pageNumber=2&pageSize=50 + ?IncidentFromDate=01/04/2026&pageNumber=1&pageSize=50 Outputs: ../docs/data/EEADP_CSO.csv — full CSO incident table @@ -26,61 +35,76 @@ """ import requests -from typing import Optional - import datetime import pandas as pd -# The CSOAPI requires a Referer header matching the portal page; plain User-Agent requests return 500. +PORTAL_URL = 'https://eeaonline.eea.state.ma.us/portal/dep/cso-data-portal/' +API_BASE_URL = 'https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/?pageSize=50&' + REQ_HEADER = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', - 'Referer': 'https://eeaonline.eea.state.ma.us/portal/dep/cso-data-portal/', + 'Referer': PORTAL_URL, 'Origin': 'https://eeaonline.eea.state.ma.us', 'Accept': 'application/json, text/plain, */*', } -API_BASE_URL = 'https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/?pageSize=50&' + +def _make_session() -> requests.Session: + return requests.Session() + def update_query_time(): - """Update the yml file that indicates the time of last query. - """ + """Update the yml file that indicates the time of last query.""" with open('../docs/data/ts_update_EEADP_CSO.yml', 'w') as f: - f.write('updated: '+str(datetime.datetime.now()).split('.')[0]+'\n') + f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') + -def _query_page(page: int, query_params: Optional[dict[str, str]]=None) -> Optional[pd.DataFrame]: - """Query for and return a single page of API results. +def _query_page(session: requests.Session, page: int, query_params: dict[str, str] | None = None) -> pd.DataFrame | None: + """Query for and return a single page of API results, or None if empty. - If the resulting query is empty, return Non + Returns None both for a normal empty page (end of results) and for an HTTP 500, + which the CSOAPI returns instead of an empty list when a filter matches no records. """ print(f'Querying for page {page}') if query_params is None: query_params = {} query_params['pageNumber'] = page query_string = '&'.join(f'{key}={val}' for key, val in query_params.items()) - r = requests.get(API_BASE_URL + query_string, headers=REQ_HEADER) + r = session.get(API_BASE_URL + query_string, headers=REQ_HEADER) + if not r.ok or 'results' not in r.json(): + return None if len(r.json()['results']) > 0: return pd.concat([pd.Series(c) for c in r.json()['results']], axis=1).T else: return None -def run_query() -> pd.DataFrame: - """Run a full query, paging through results and returning a combined DataFrame. + +def run_query(session: requests.Session, from_date: str | None = None) -> pd.DataFrame: + """Page through API results and return a combined DataFrame. + + from_date: optional MM/DD/YYYY string passed as IncidentFromDate. """ - print('Running full query') + if from_date: + print(f'Running incremental query from {from_date}') + else: + print('Running full query') + query_params: dict[str, str] = {} + if from_date: + query_params['IncidentFromDate'] = from_date page = 1 # CSOAPI is 1-indexed result_dfs = [] while True: - df = _query_page(page) + df = _query_page(session, page, query_params) if df is None: break result_dfs.append(df) page += 1 + if not result_dfs: + return pd.DataFrame() return pd.concat(result_dfs) -def get_data() -> pd.DataFrame: - """Query data from the data portal API and do any necessary post processing. - """ - df = run_query() + +def _parse_dates(df: pd.DataFrame) -> pd.DataFrame: df['incidentDate'] = pd.to_datetime(df['incidentDate'], format='ISO8601') df['submittedDate'] = pd.to_datetime(df['submittedDate'], format='ISO8601') # API already returns a lowercase 'year' column; drop it before adding 'Year' @@ -89,20 +113,59 @@ def get_data() -> pd.DataFrame: df['Year'] = df['incidentDate'].apply(lambda x: x.year) return df + +def get_data() -> pd.DataFrame: + """Fetch CSO data incrementally when a cached CSV exists, otherwise full fetch.""" + csv_path = '../docs/data/EEADP_CSO.csv' + from_date: str | None = None + existing: pd.DataFrame | None = None + + try: + existing = pd.read_csv(csv_path, index_col=0) + existing['incidentDate'] = pd.to_datetime(existing['incidentDate'], format='ISO8601') + max_date = existing['incidentDate'].max() + from_date = max_date.strftime('%m/%d/%Y') + # Drop cached rows on the boundary date — the API refetch will include them. + existing = existing[existing['incidentDate'].dt.date < max_date.date()].copy() + print(f' Cached data through {max_date.date()}; fetching from {from_date} (inclusive)') + except FileNotFoundError: + print(' No cache found; running full query') + + session = _make_session() + raw = run_query(session, from_date=from_date) + + if raw.empty: + # API returned 500 (no-records) or genuinely empty; use cache as-is. + print(' No new records returned; using existing cache unchanged.') + if existing is not None: + # Restore the boundary rows we dropped before returning + full_existing = pd.read_csv('../docs/data/EEADP_CSO.csv', index_col=0) + return full_existing + # No cache and no data — nothing to write. + raise RuntimeError('CSO API returned no data and no cache exists.') + + new_df = _parse_dates(raw) + + if existing is not None: + df = pd.concat([existing, new_df], ignore_index=True) + else: + df = new_df + + return df + + def write_data(df: pd.DataFrame): - """Write data to a local table for integration with AMEND. - """ - print('Writing out queries data') + """Write data to a local table for integration with AMEND.""" + print('Writing out queried data') df.to_csv('../docs/data/EEADP_CSO.csv', index=True) - ## Print a sample of the file as an example - df.sample(n=10).to_csv('../docs/data/EEADP_CSO_sample.csv', index=0) + df.sample(n=10).to_csv('../docs/data/EEADP_CSO_sample.csv', index=False) + def main(): - """Query and write all data. - """ all_data = get_data() write_data(all_data) update_query_time() + if __name__ == '__main__': main()