[EPA_USAirPollutantEmissionTrends_NationalAndState] Updated config.json for latest updated data

scripts/us_epa/air_emissions_inventory/national_state/README.md

@@ -52,31 +52,29 @@ or
 
 ### Import Procedure
 
-The below script will download the data.
+The below script will download the data:
 
 `python3 download.py --config_path=unresolved_mcf/us_epa/config.json`
 
-Note on Configuration Management for Data Source URLs
+#### Configuration Management & Automatic Fallback for Data Source URLs
 
-To ensure consistent access to the most current data, a dedicated configuration file, config.json, will be maintained within our Google Cloud Storage (GCS) environment. This approach addresses the challenge posed by source URLs that contain dynamic and unpredictable year and date components, precluding their programmatic generation.
+To access the latest data, a configuration file (`config.json`) is stored in Google Cloud Storage (GCS) at `gs://unresolved_mcf/us_epa/config.json`. Because the EPA periodically publishes revised National and State `.xlsx` workbooks with new release-date suffixes (retiring the older URLs with HTTP 404), `download.py` uses a two-tier resolution strategy:
 
-Content Structure:
-The config.json file will adhere to a JSON array format, with each object in the array defining a download_path attribute. This attribute will contain the fully qualified URL for the latest available National and State-level data files.
+1. **Primary (GCS Config)**: Reads `download_path` entries from `--config_path` (passed as `bucket_name/path/to/config.json` without the `gs://` prefix) and verifies that all configured URLs return HTTP 200.
+2. **Automatic Fallback (Live EPA Trends Page Discovery)**: If the GCS config cannot be read or any configured URL returns a non-200 status (such as HTTP 404 after an EPA release update), `download.py` automatically scrapes `https://www.epa.gov/air-emissions-inventories/air-pollutant-emissions-trends-data` to discover and validate the current `national_tier1_caps*.xlsx` and `state_tier1*ktons.xlsx` URLs.
 
-The config_path parameter designates a Google Cloud Storage (GCS) location. It should be provided as the bucket and object path (e.g., my-bucket/path/to/config.json), without the gs:// prefix. The system will automatically prepend the gs:// scheme internally when accessing the resource."
-
-Example config.json Content:
-
-JSON
+Current `config.json` Content (`gs://unresolved_mcf/us_epa/config.json`):
 
+```json
 [
-  {"download_path": "https://www.epa.gov/system/files/other-files/2025-04/national_tier1_caps_21feb2025.xlsx"},
-  {"download_path": "https://www.epa.gov/system/files/other-files/2025-04/state_tier1_21feb2005_ktons.xlsx"}
+  {"download_path": "https://www.epa.gov/system/files/other-files/2026-09/national_tier1_caps_04sep2026.xlsx"},
+  {"download_path": "https://www.epa.gov/system/files/other-files/2026-09/state_tier1_1990_2025_04sep2026_ktons.xlsx"}
 ]
+```
+*(When the EPA publishes a newer release on the [Air Pollutant Emissions Trends Data](https://www.epa.gov/air-emissions-inventories/air-pollutant-emissions-trends-data) page, `download.py` automatically discovers the updated URLs if `config.json` has not yet been updated.)*
 
-Maintenance:
-This configuration mechanism requires manual updates to the config.json file whenever the source URLs for the National or State data undergo revision, thereby ensuring the application always references the latest datasets.
+#### Processing
 
-The below script will clean the data, Also generate final csv, mcf and tmcf files.
+The below script cleans the downloaded workbooks and generates the final CSV, MCF, and TMCF files:
 
-`python3 process.py`
+`python3 process.py`

scripts/us_epa/air_emissions_inventory/national_state/download.py

@@ -18,7 +18,11 @@
 
 import os
 import json
-import urllib.request
+import re
+import urllib.parse
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 from google.cloud import storage  # For GCS integration
 from absl import flags, app, logging  # Use absl for flag handling
 
@@ -27,31 +31,128 @@
 flags.DEFINE_string('config_path', '',
                     'Path to the configuration file in the GCS bucket.')
 
-# Hardcoded GCS bucket name
-GCS_BUCKET_NAME = "unresolved_mcf"
+# EPA Air Pollutant Emissions Trends Data provenance page
+EPA_TRENDS_PAGE_URL = ("https://www.epa.gov/air-emissions-inventories/"
+                       "air-pollutant-emissions-trends-data")
 
 # Download path for files
-_DOWNLOAD_PATH = os.path.join(os.path.dirname(__file__), 'input_files')
-
-
-def check_url_status(url: str) -> bool:
+_DOWNLOAD_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                              'input_files')
+
+_HTTP_TIMEOUT_SECONDS = 60
+_DEFAULT_HEADERS = {
+    "User-Agent": ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+                   "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36")
+}
+
+
+def _create_http_session() -> requests.Session:
+    """Creates an HTTP session with retry and exponential backoff."""
+    session = requests.Session()
+    session.headers.update(_DEFAULT_HEADERS)
+    retry_strategy = Retry(
+        total=5,
+        backoff_factor=2,
+        status_forcelist=[429, 500, 502, 503, 504],
+        allowed_methods=["HEAD", "GET", "OPTIONS"],
+    )
+    adapter = HTTPAdapter(max_retries=retry_strategy)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
+
+def check_url_status(url: str, session: requests.Session = None) -> bool:
     """
     Check if the given URL is reachable and log URL info.
 
     Args:
         url (str): The URL to check.
+        session (requests.Session, optional): HTTP session to reuse.
 
     Returns:
         bool: True if the URL is reachable (status code 200), False otherwise.
     """
+    own_session = session is None
+    if own_session:
+        session = _create_http_session()
     try:
-        with urllib.request.urlopen(url) as response:
-            status_code = response.status
+        logging.info(f"Checking URL status: {url}")
+        response = session.head(url,
+                                allow_redirects=True,
+                                timeout=_HTTP_TIMEOUT_SECONDS)
+        status_code = response.status_code
+        response.close()
+        if status_code in (403, 405):
+            with session.get(url, stream=True,
+                             timeout=_HTTP_TIMEOUT_SECONDS) as get_response:
+                status_code = get_response.status_code
+        if status_code == 200:
             logging.info(f"URL {url} is reachable. Status code: {status_code}")
-            return status_code == 200
+            return True
+        logging.warning(
+            f"URL {url} returned non-200 status code: {status_code}")
+        return False
     except Exception as e:
-        logging.fatal(f"Failed to reach URL {url}. Details: {e}")
+        logging.warning(f"Failed to reach URL {url}. Details: {e}")
         return False
+    finally:
+        if own_session:
+            session.close()
+
+
+def scrape_epa_trends_urls(session: requests.Session = None) -> list:
+    """
+    Scrapes the EPA Air Pollutant Emissions Trends Data page for the current
+    National Tier 1 CAPS and State Tier 1 CAPS .xlsx download URLs.
+
+    Args:
+        session (requests.Session, optional): HTTP session to reuse.
+
+    Returns:
+        list: Discovered [national_xlsx_url, state_xlsx_url], or empty list.
+    """
+    own_session = session is None
+    if own_session:
+        session = _create_http_session()
+    try:
+        logging.info(
+            f"Discovering latest .xlsx URLs via GET: {EPA_TRENDS_PAGE_URL}")
+        response = session.get(EPA_TRENDS_PAGE_URL,
+                               timeout=_HTTP_TIMEOUT_SECONDS)
+        response.raise_for_status()
+        html = response.text
+
+        hrefs = re.findall(r'href=["\']([^"\']+\.xlsx(?:\?[^"\']*)?)["\']',
+                           html, re.IGNORECASE)
+        national_url = None
+        state_url = None
+        for href in hrefs:
+            full_url = urllib.parse.urljoin(EPA_TRENDS_PAGE_URL, href)
+            filename = os.path.basename(
+                urllib.parse.urlsplit(full_url).path).lower()
+            if not national_url and re.match(r"^national_tier1_caps.*\.xlsx$",
+                                             filename):
+                national_url = full_url
+            elif not state_url and re.match(r"^state_tier1.*ktons\.xlsx$",
+                                            filename):
+                state_url = full_url
+
+        discovered = [u for u in (national_url, state_url) if u]
+        if len(discovered) == 2:
+            logging.info(f"Discovered current EPA trend URLs: {discovered}")
+            return discovered
+        logging.warning(f"Incomplete URL discovery from {EPA_TRENDS_PAGE_URL}; "
+                        f"expected 2 URLs, found: {discovered}")
+        return []
+    except Exception as e:
+        logging.warning(
+            f"Failed to scrape EPA trends page {EPA_TRENDS_PAGE_URL}. "
+            f"Details: {e}")
+        return []
+    finally:
+        if own_session:
+            session.close()
 
 
 def parse_gcs_path(full_path: str) -> tuple:
@@ -95,13 +196,50 @@ def read_config_from_gcs(bucket_name: str, config_path: str) -> list:
         ]
         return urls
     except Exception as e:
-        logging.fatal(
+        logging.warning(
             f"Failed to read configuration file from GCS. Details: {e}")
         return []
-    except Exception as e:
-        logging.fatal(
-            f"Failed to read configuration file from GCS. Details: {e}")
+
+
+def resolve_download_urls(config_urls: list,
+                          session: requests.Session = None) -> list:
+    """
+    Verifies configured URLs and falls back to scraping the EPA trends page
+    if any configured URL is missing or returns non-200 (e.g., HTTP 404 after
+    an EPA release update).
+    """
+    own_session = session is None
+    if own_session:
+        session = _create_http_session()
+    try:
+        if (config_urls and len(config_urls) >= 2 and
+                all(check_url_status(u, session=session) for u in config_urls)):
+            return config_urls
+
+        logging.warning(
+            "Configured GCS URLs are missing or unreachable; falling back to "
+            "live EPA trends page discovery.")
+        scraped_urls = scrape_epa_trends_urls(session=session)
+        if (len(scraped_urls) == 2 and all(
+                check_url_status(u, session=session) for u in scraped_urls)):
+            return scraped_urls
+
         return []
+    finally:
+        if own_session:
+            session.close()
+
+
+def _clean_download_directory(download_dir: str) -> None:
+    """Removes stale Excel or temporary files from previous runs."""
+    if not os.path.exists(download_dir):
+        return
+    for existing_file in sorted(os.listdir(download_dir)):
+        if existing_file.lower().endswith(('.xlsx', '.xls', '.tmp')):
+            stale_path = os.path.join(download_dir, existing_file)
+            if os.path.isfile(stale_path):
+                os.remove(stale_path)
+                logging.info(f"Removed stale file: {stale_path}")
 
 
 def download_files(urls: list) -> None:
@@ -114,49 +252,65 @@ def download_files(urls: list) -> None:
     Returns:
         None
     """
-    if not os.path.exists(_DOWNLOAD_PATH):
-        os.mkdir(_DOWNLOAD_PATH)
-    os.chdir(_DOWNLOAD_PATH)
+    os.makedirs(_DOWNLOAD_PATH, exist_ok=True)
 
-    for file_url in urls:
-        # Check URL status before proceeding
-        if not check_url_status(file_url):
-            logging.fatal(f"URL not reachable: {file_url}. Skipping download.")
+    session = _create_http_session()
+    try:
+        resolved_urls = resolve_download_urls(urls, session=session)
+        if not resolved_urls:
+            raise ValueError("No reachable URLs found to download.")
 
-        file_name = file_url.split("/")[-1]
+        _clean_download_directory(_DOWNLOAD_PATH)
 
-        try:
+        for file_url in resolved_urls:
+            file_name = os.path.basename(urllib.parse.urlsplit(file_url).path)
             file_path = os.path.join(_DOWNLOAD_PATH, file_name)
-            urllib.request.urlretrieve(file_url, file_path)
-            logging.info(f"Successfully downloaded: {file_name} to {file_path}")
-        except Exception as e:
-            logging.fatal(
-                f"Failed to download {file_name} from {file_url}. Details: {e}")
+            tmp_path = f"{file_path}.tmp"
+
+            try:
+                logging.info(f"Downloading GET {file_url} to {file_path}")
+                with session.get(file_url,
+                                 stream=True,
+                                 timeout=_HTTP_TIMEOUT_SECONDS) as response:
+                    response.raise_for_status()
+                    with open(tmp_path, "wb") as f_out:
+                        for chunk in response.iter_content(chunk_size=65536):
+                            if chunk:
+                                f_out.write(chunk)
+                file_size = os.path.getsize(tmp_path)
+                if file_size == 0:
+                    raise IOError(f"Downloaded file is empty: {tmp_path}")
+                os.replace(tmp_path, file_path)
+                logging.info(
+                    f"Successfully downloaded: {file_name} ({file_size} bytes) "
+                    f"to {file_path}")
+            except Exception as e:
+                if os.path.exists(tmp_path):
+                    os.remove(tmp_path)
+                logging.fatal(
+                    f"Failed to download {file_name} from {file_url}. "
+                    f"Details: {e}")
+    finally:
+        session.close()
 
 
 def main(argv):
+    del argv
     # Get the config path from the flags
     config_path_flag = _FLAGS.config_path
 
     if not config_path_flag:
-        logging.fatal(
+        raise ValueError(
             "Please provide the --config_path flag with a valid GCS path.")
-        return
 
     try:
         bucket_name, config_path = parse_gcs_path(config_path_flag)
     except ValueError as e:
         logging.fatal(f"Invalid GCS path: {e}")
-        return
 
-    # Read URLs from the GCS-hosted config file
+    # Read URLs from the GCS-hosted config file (with live EPA page fallback)
     urls = read_config_from_gcs(bucket_name, config_path)
-
-    # Proceed to download files
-    if urls:
-        download_files(urls)
-    else:
-        logging.fatal("No URLs to download. Exiting program.")
+    download_files(urls)
 
 
 if __name__ == '__main__':

scripts/us_epa/air_emissions_inventory/national_state/manifest.json

@@ -18,9 +18,11 @@
                 }
             ],
             "cron_schedule": "0 10 1,15 * *",
+            "validation_config_file": "validation_config.json",
             "source_files": [
-                "input_files/*"
+                "input_files/*",
+                "validation_config.json"
             ]
         }
     ]
-}
+}