IPEDS_StudentToFacultyRatio: Adding provisional data mapping, updated download/preprocess scripts, and golden validation

statvar_imports/ipeds/student_to_faculty_ratio/download.py

@@ -17,145 +17,169 @@
 import re
 import zipfile
 import time
-from datetime import date 
+import requests
+from datetime import date
+from urllib.parse import urlparse
 from absl import logging
 
 # --- Configuration ---
 START_YEAR = 2009
 # Set END_YEAR dynamically to the current calendar year
 END_YEAR = date.today().year
-BASE_URL = "https://nces.ed.gov/ipeds/datacenter/data/EF{}D.zip"
+
+BASE_URL_RV = "https://nces.ed.gov/ipeds/data-generator?year={year}&tableName=EF{year}D&HasRV=1&type=csv"
+BASE_URL_LEGACY = "https://nces.ed.gov/ipeds/datacenter/data/EF{year}D.zip"
+BASE_URL_COMPLETE = "https://nces.ed.gov/ipeds/complete-data-files/EF{year}D.zip"
+BASE_URL_PROV = "https://nces.ed.gov/ipeds/data-generator?year={year}&tableName=EF{year}D&HasRV=0&type=csv"
+
+BASE_URL_TEMPLATES = [
+    BASE_URL_RV,
+    BASE_URL_LEGACY,
+    BASE_URL_COMPLETE,
+    BASE_URL_PROV,
+]
+
 DOWNLOAD_DIR = "input_files"
 # Pattern to match files ending in '_rv' followed by a file extension
-# The pattern should match '_rv.txt', '_rv.csv', etc.
 RV_PATTERN = re.compile(r'_rv\.[a-z0-9]+$', re.IGNORECASE)
+# Pattern to match provisional files (e.g. ef2023d.csv, ef2024d.csv)
+PROVISIONAL_PATTERN = re.compile(r'^ef\d{4}d\.[a-z0-9]+$', re.IGNORECASE)
 # ---------------------
 
 # --- Path Adjustment for Utility Import ---
 _SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__))
-# Correct path to the directory containing download_util_script.py:
 sys.path.append(os.path.join(_SCRIPT_PATH, '../../../util/'))
 
 try:
-    # IMPORT ONLY THE FUNCTION THAT EXISTS in the utility script
     from download_util_script import download_file
 except ImportError as e:
-    # Use logging.fatal for critical import errors and exit
     logging.fatal("Could not import 'download_file'. Please ensure the utility script is accessible. Original error: %s", e)
     raise RuntimeError(f"FATAL: Missing utility script dependency: {e}")
 
 
-def process_and_filter_zip(zip_path: str, output_dir: str, filter_pattern: re.Pattern) -> bool:
+def process_and_filter_zip(zip_path: str, output_dir: str) -> bool:
     """
-    Handles the custom unzipping, RV-pattern filtering, and cleanup.
-    If extraction fails unexpectedly, it is treated as a fatal error.
+    Unzips and filters contents of zip_path.
+    Extracts files matching RV_PATTERN if present; otherwise matches PROVISIONAL_PATTERN.
+    Returns True if matching file(s) were found and extracted, False otherwise.
     """
     zip_filename = os.path.basename(zip_path)
-    logging.info("  Unzipping and filtering contents (keeping only files matching pattern: %s)...", filter_pattern.pattern)
-
-    extraction_successful = False
 
     try:
-        # 1. Unzip and filter
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
             all_files = zip_ref.namelist()
-            files_to_extract = []
-
-            for file_name in all_files:
-                base_name = os.path.basename(file_name)
-                # Check if the file is at the root or within a single folder.
-                if filter_pattern.search(base_name):
-                    files_to_extract.append(file_name)
+
+            # Prefer revised files first
+            files_to_extract = [
+                f for f in all_files if RV_PATTERN.search(os.path.basename(f))
+            ]
+            # Fall back to provisional files if no revised file is in the zip
+            if not files_to_extract:
+                files_to_extract = [
+                    f for f in all_files if PROVISIONAL_PATTERN.search(os.path.basename(f))
+                ]
 
             if not files_to_extract:
-                # This is a warning, not a failure, as the process was clean.
-                logging.fatal("  Warning: No files matching the pattern found in %s. Skipping extraction.", zip_filename)
-                extraction_successful = True
+                return False
 
-            # Extract the filtered files
             for file_name in files_to_extract:
                 zip_ref.extract(file_name, output_dir)
                 logging.info("    Extracted: %s", file_name)
 
-            if files_to_extract:
-                logging.info("  Extraction successful.")
-            extraction_successful = True
+            return True
 
     except zipfile.BadZipFile:
-        # FATAL: Corrupted zip file means data is unavailable, and we must stop processing this file.
         logging.fatal("  FATAL ERROR: %s is a corrupted or empty zip file. Cannot proceed.", zip_filename)
         raise RuntimeError(f"Corrupted or empty zip file encountered: {zip_filename}")
     except Exception as e:
-        # FATAL: Any unexpected extraction error means partial data, which must be avoided.
         logging.fatal("  FATAL ERROR: An unexpected error occurred during unzipping/extraction of %s: %s", zip_filename, e)
-        # Raise an error to stop the script from proceeding with potentially partial unzipped files
         raise RuntimeError(f"Extraction failed for {zip_filename}: {e}")
     finally:
-        # 2. Clean up the downloaded zip file manually, regardless of success or failure
         if os.path.exists(zip_path):
             try:
                 os.remove(zip_path)
-                logging.info("  Removed zip file: %s", zip_path)
-            except OSError as e:
-                # Use info for non-critical file removal errors
-                logging.info("  Warning: Failed to remove zip file %s: %s", zip_path, e)
-
-    return extraction_successful
+            except OSError:
+                pass
 
 
-def main():
+def download_for_year(year: int) -> bool:
     """
-    Downloads IPEDS zip files using the utility, then handles custom unzipping/filtering locally.
-    Download failure is logged as a warning, allowing the script to proceed to the next year.
-    Extraction failure is logged as FATAL, which will stop the script for that year's file.
+    Downloads data for a given year by trying candidate URLs in order of preference.
+    Extracts and saves the first successful dataset found for that year.
     """
+    logging.info("\nProcessing year %d...", year)
 
-    # 1. Create the target directory if it doesn't exist
-    if not os.path.exists(DOWNLOAD_DIR):
-        try:
-            os.makedirs(DOWNLOAD_DIR)
-            logging.info("Created directory: %s", DOWNLOAD_DIR)
-        except OSError as e:
-            # Use logging.fatal for critical directory creation errors
-            logging.fatal("FATAL ERROR: Could not create directory %s: %s", DOWNLOAD_DIR, e)
-            raise RuntimeError(f"FATAL: Directory creation failed for {DOWNLOAD_DIR}: {e}")
-
-    # 2. Iterate through the required year range
-    for year in range(START_YEAR, END_YEAR + 1):
-        url = BASE_URL.format(year)
-        zip_filename = f"EF{year}D.zip"
-        download_path = os.path.join(DOWNLOAD_DIR, zip_filename)
-
-        logging.info("\nProcessing year %d...", year)
+    for url_template in BASE_URL_TEMPLATES:
+        if "{year}" in url_template:
+            url = url_template.format(year=year)
+        else:
+            url = url_template.format(year)
 
         try:
-            # 3. Call the utility function to DOWNLOAD ONLY (unzip=False)
             download_success = download_file(
                 url=url,
                 output_folder=DOWNLOAD_DIR,
-                unzip=False, # <-- CRITICAL: Do not let the utility unzip the file
-                tries=3,
-                delay=5,
+                unzip=False,
+                tries=4,
+                delay=2,
                 backoff=2
             )
 
             if download_success:
-                # 4. Handle custom processing (unzip, filter, and cleanup) locally
-                # If process_and_filter_zip encounters a fatal error, it will raise an exception
-                process_and_filter_zip(download_path, DOWNLOAD_DIR, RV_PATTERN)
+                # Determine the filename inferred by download_file
+                parsed_url = urlparse(url)
+                file_name = os.path.basename(parsed_url.path)
+                if not file_name:
+                    file_name = "downloaded_file"
+                elif '.' not in file_name:
+                    file_name = file_name + '.xlsx'
+
+                download_path = os.path.join(DOWNLOAD_DIR, file_name)
+                default_zip_path = os.path.join(DOWNLOAD_DIR, f"EF{year}D.zip")
+
+                actual_download_path = download_path if os.path.exists(download_path) else default_zip_path
+
+                if os.path.exists(actual_download_path):
+                    if zipfile.is_zipfile(actual_download_path):
+                        if process_and_filter_zip(actual_download_path, DOWNLOAD_DIR):
+                            logging.info("  Successfully fetched and extracted dataset for year %d.", year)
+                            return True
+                    else:
+                        # Direct CSV download (e.g. from data-generator)
+                        target_name = f"ef{year}d_rv.csv" if "HasRV=1" in url else f"ef{year}d.csv"
+                        target_path = os.path.join(DOWNLOAD_DIR, target_name)
+                        if os.path.exists(target_path):
+                            os.remove(target_path)
+                        os.rename(actual_download_path, target_path)
+                        logging.info("  Successfully fetched direct CSV dataset for year %d.", year)
+                        return True
+        except requests.exceptions.RequestException as e:
+            logging.info("  Network error for candidate URL %s: %s", url, e)
+        except Exception as e:
+            logging.info("  Candidate URL %s failed: %s", url, e)
 
-            else:
-                # Not available/download failed: Log as a warning and skip, as requested
-                logging.info("Warning: Download failed for year %d. Skipping processing for this year.", year)
+    logging.info("Warning: No dataset found for year %d across candidate URLs.", year)
+    return False
 
-        except Exception as e:
-            # Catch errors that process_and_filter_zip explicitly raises (BadZipFile, Extraction Error)
-            # or any other unexpected error during the loop. The error is already logged as FATAL.
-            logging.info("Execution failed for year %d, continuing to next year (if possible). Error: %s", year, e)
-            # The script will now proceed to the next iteration unless the error is outside the loop
 
-        # Optional: Add a pause between years to respect NCES server requests
-        time.sleep(5)
+def main():
+    """
+    Iterates through year range downloading datasets (preferring revised, falling back to provisional).
+    """
+
+    # 1. Create target directory if it doesn't exist
+    if not os.path.exists(DOWNLOAD_DIR):
+        try:
+            os.makedirs(DOWNLOAD_DIR)
+            logging.info("Created directory: %s", DOWNLOAD_DIR)
+        except OSError as e:
+            logging.fatal("FATAL ERROR: Could not create directory %s: %s", DOWNLOAD_DIR, e)
+            raise RuntimeError(f"FATAL: Directory creation failed for {DOWNLOAD_DIR}: {e}")
+
+    # 2. Iterate through required year range
+    for year in range(START_YEAR, END_YEAR + 1):
+        download_for_year(year)
+        time.sleep(1)
 
 
 if __name__ == "__main__":
@@ -164,5 +188,4 @@ def main():
         main()
         logging.info("\nScript finished. Filtered files extracted to the '%s' folder.", DOWNLOAD_DIR)
     except Exception as e:
-        # Catch errors that prevent main from starting or critical errors like directory creation
         logging.fatal("\nFATAL ERROR in main execution: %s", e)