Indianss

statvar_imports/india_ndap/india_nss_health_ailments/README.md

@@ -1,12 +1,11 @@
-
 # India NSS Health Ailments
 
 ## 1. Import Overview
 
 This project processes and imports health ailment data from the **National Sample Survey (NSS) Report No. 556 on Health in India**. The dataset provides a profile of ailments, healthcare facility usage, and expenditures based on a survey.
 
 * **Source URL**: [https://ndap.niti.gov.in/dataset/7300](https://ndap.niti.gov.in/dataset/7300)
-* **Import Type**: Manual file-based import (CSV)
+* **Import Type**: Semi-Automated
 * **Source Data Availability**: Released by NDAP (NITI Aayog) based on periodic NSS survey rounds. Not updated on a regular cadence.
 * **Release Frequency**: Ad-hoc (typically once every 5 years); updated manually by team when a new round is published
 * **Notes**: The dataset includes metrics on various ailments as reported during NSS rounds. Each row represents health-related observations per ailment per state per year.
@@ -20,14 +19,14 @@ Before ingestion, the following preprocessing is done:
 * **Input files**:
 
   * `india_nss_health_ailments.csv`: Raw input data
-  * `pvmap.csv`: Property-value mapping
-  * `place_resolved.csv`: Geo resolution data for Indian states/UTs
-  * `metadata.csv`: StatVar metadata (used by `stat_var_processor.py`)
+  * `india_nss_health_ailments_pvmap.csv`: Property-value mapping
+  * `india_nss_health_ailments_place_resolved.csv`: Geo resolution data for Indian states/UTs
+  * `india_nss_health_ailments_metadata.csv`: StatVar metadata (used by `stat_var_processor.py`)
 * **Transformation pipeline**:
 
   * Columns are cleaned and standardized to match StatVar expectations.
   * StatVars are generated using `stat_var_processor.py`.
-  * Output is written to `india_nss_health_ailments.csv` and corresponding `india_nss_health_ailments.tmcf`.
+  * Output is written to `output/IndiaNSS_HealthAilments_output.csv` and corresponding `output/IndiaNSS_HealthAilments_output.tmcf`.
 * **Data Quality Checks**:
 
   * Linting is performed using the DataCommons import tool JAR
@@ -44,59 +43,97 @@ Before ingestion, the following preprocessing is done:
 * **Steps**:
 
   1. Monitor [NDAP Dataset 7300](https://ndap.niti.gov.in/dataset/7300) for new survey releases
-  2. Manually download the latest data
+  2. Download raw data using `download_script.py`
   3. Preprocess using `stat_var_processor.py` with updated CSV and mapping files
   4. Run linting and validation
   5. Upload final files to:
 
      * `gs://datcom-imports/india_ndap/NDAP_NSS_Health/latest/`
   6. Trigger `run_import.sh` manually for test/prod ingestion
-* **Note**: This pipeline is not fully automated due to manual file retrieval and preprocessing needs.
+* **Note**: This pipeline is semi-automated using a scheduled cron and GCS-staged credential configuration.
 
 ---
 
 ## 4. Script Execution Details
 
-### Script 1: `stat_var_processor.py`
+### Script 1: `download_script.py`
+
+**Usage**:
+
+```bash
+python3 download_script.py
+```
+
+**Output**: `india_nss_health_ailments.csv`
+
+**Purpose**: Downloads the raw data from the NDAP API and saves it as `india_nss_health_ailments.csv`. It retrieves query URL and credentials from GCS.
+
+---
+
+### Script 2: `stat_var_processor.py`
 
 **Usage**:
 
 ```bash
 python3 stat_var_processor.py \
-  --input_data='/path/to/india_nss_health_ailments.csv' \
-  --pv_map='/path/to/pvmap.csv' \
-  --places_resolved_csv='/path/to/place_resolved.csv' \
-  --config_file='/path/to/metadata.csv' \
-  --output_path='/path/to/output/health_nss' \
+  --input_data='india_nss_health_ailments.csv' \
+  --pv_map='india_nss_health_ailments_pvmap.csv' \
+  --places_resolved_csv='india_nss_health_ailments_place_resolved.csv' \
+  --config_file='india_nss_health_ailments_metadata.csv' \
+  --output_path=output/IndiaNSS_HealthAilments_output \
+  --output_counters=counters/IndiaNSS_HealthAilments_counters.csv \
   --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf
 ```
 
-**Purpose**: Generates StatVar MCF and cleaned observation CSV (`india_nss_health_ailments.csv`, `india_nss_health_ailments.tmcf`)
+**Purpose**: Generates StatVar MCF and cleaned observation CSV (`output/IndiaNSS_HealthAilments_output.csv`, `output/IndiaNSS_HealthAilments_output.tmcf`)
 
 ---
 
-### Script 2: Java Linting Tool
+### Script 3: Java Linting Tool
 
 **Usage**:
 
 ```bash
 java -jar '/path/to/datacommons-import-tool.jar' lint \
-  '/path/to/india_nss_health_ailments.csv' \
-  '/path/to/india_nss_health_ailments.tmcf'
+  'output/IndiaNSS_HealthAilments_output.csv' \
+  'output/IndiaNSS_HealthAilments_output.tmcf'
 ```
 
 **Purpose**: Validates final CSV+TMCF for formatting and semantic consistency before ingestion
 
 ---
 
-### Script 3: `download_script.py` (if used)
+## 5. Configuration & Troubleshooting
 
-**Usage**:
+### GCS Configuration Location
 
-```bash
-python3 download_script.py
-```
-**Output**: `india_nss_health_ailments.csv`
+`download_script.py` fetches the NDAP query URL and API credentials from Google Cloud Storage:
+* **GCS Path**: `gs://unresolved_mcf/india_ndap/NDAP_NSS_Health/latest/download_config.json`
 
-**Purpose**: Downloads the raw data from the NDAP API and saves it as `india_nss_health_ailments.csv`.
+**Expected JSON Structure**:
+```json
+{
+  "url": "<NDAP_API_QUERY_URL_WITH_CREDENTIALS>",
+  "input_files": [
+    "india_nss_health_ailments.csv"
+  ]
+}
+```
 
+### Troubleshooting NDAP API Key / Token Expiry
+
+If the download script fails or logs HTTP `401 Unauthorized` / `403 Forbidden` errors:
+
+1. **Obtain Fresh API Query/Key**:
+   * Navigate to [NDAP Dataset 7300](https://ndap.niti.gov.in/dataset/7300).
+   * Generate or copy the updated API query URL containing the valid access token / API key.
+2. **Update GCS Configuration**:
+   * Create or update the local `download_config.json` with the new query URL.
+   * Upload the updated configuration file to GCS:
+     ```bash
+     gcloud storage cp download_config.json gs://unresolved_mcf/india_ndap/NDAP_NSS_Health/latest/download_config.json
+     # or
+     gsutil cp download_config.json gs://unresolved_mcf/india_ndap/NDAP_NSS_Health/latest/download_config.json
+     ```
+3. **Verify Download**:
+   * Re-run `python3 download_script.py` and verify `india_nss_health_ailments.csv` downloads successfully.

statvar_imports/india_ndap/india_nss_health_ailments/download_script.py

@@ -34,11 +34,12 @@
 # --- ROBUST PATH RESOLUTION START ---
 _SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__))
 
+
 def _add_util_to_path():
     """Adds the repo 'util' directory to sys.path dynamically."""
     path_3_up = os.path.abspath(os.path.join(_SCRIPT_PATH, '../../../util/'))
     path_2_up = os.path.abspath(os.path.join(_SCRIPT_PATH, '../../util/'))
-    
+
     if os.path.exists(os.path.join(path_3_up, 'file_util.py')):
         sys.path.append(path_3_up)
     elif os.path.exists(os.path.join(path_2_up, 'file_util.py')):
@@ -53,6 +54,7 @@ def _add_util_to_path():
             curr = os.path.dirname(curr)
         logging.error("Could not find 'util' directory containing file_util.py")
 
+
 _add_util_to_path()
 
 try:
@@ -78,6 +80,7 @@ def _add_util_to_path():
     'Year',
 ]
 
+
 def load_config(path: str) -> dict:
     """Loads configuration from GCS or local disk."""
     if path.startswith('gs://'):
@@ -88,29 +91,29 @@ def load_config(path: str) -> dict:
         return json.loads(blob.download_as_string())
     return file_util.file_load_py_dict(path)
 
-def download_data(config_file_path: str) -> Tuple[List[Tuple], str]:
+
+def download_data(config_file_path: str) -> List[Tuple]:
     """Downloads data using configuration."""
     file_config = load_config(config_file_path)
     url = file_config.get('url')
-    # We ignore the 'input_files' from config to save in the current directory
-    output_dir = '' 
-
+
     if not url:
-        return [], ''
+        return []
 
     all_data = []
     page_num = 1
     while True:
         api_url = f'{url}&pageno={page_num}'
         response = _retry_method(api_url, None, 3, 5, 2)
         if not response:
-            logging.fatal('Failed to retrieve data from page %d', page_num)
+            logging.fatal('Failed to retrieve data from %s (page %d)', api_url,
+                          page_num)
 
         try:
             response_data = response.json()
-        except json.JSONDecodeError:
-            logging.error('Failed to parse JSON from page %d', page_num)
-            break
+        except json.JSONDecodeError as e:
+            logging.fatal('Failed to parse JSON from %s (page %d): %s', api_url,
+                          page_num, e)
 
         if response_data and 'Data' in response_data and response_data['Data']:
             for item in response_data['Data']:
@@ -135,42 +138,28 @@ def download_data(config_file_path: str) -> Tuple[List[Tuple], str]:
             logging.info('No more data found on page %d.', page_num)
             break
 
-    return all_data, output_dir
+    return all_data
 
-def preprocess_and_save(data: List[Tuple], output_dir: str) -> None:
+
+def preprocess_and_save(data: List[Tuple]) -> None:
     """Saves data to CSV directly in the script directory."""
     if not data:
         logging.info('No data was retrieved from the API.')
         return
 
     df = pd.DataFrame(data, columns=_OUTPUT_COLUMNS)
-    
+
     # Save directly in _SCRIPT_PATH (statvar_imports/india_ndap/india_nss_health_ailments)
-    output_path = os.path.join(_SCRIPT_PATH, 'india_nss_health_ailments.csv')  
+    output_path = os.path.join(_SCRIPT_PATH, 'india_nss_health_ailments.csv')
     df.to_csv(output_path, index=False)
     logging.info('Data saved to %s', output_path)
-  """Converts data to a DataFrame and saves it as a CSV file.
-
-  Args:
-    data: The data to be processed, as a list of tuples.
-    output_dir: The directory where the output CSV will be saved.
-  """
-  if not data:
-    logging.info('No data was retrieved from the API.')
-    return
-
-  df = pd.DataFrame(data, columns=_OUTPUT_COLUMNS)
-
-  os.makedirs(output_dir, exist_ok=True)
-  output_path = os.path.join(output_dir, 'india_nss_health_ailments.csv')  
-  df.to_csv(output_path, index=False)
-  logging.info('Data saved to %s', output_path)
 
 
 def main(_) -> None:
-    raw_data, output_dir = download_data(_FLAGS.config_file_path)
+    raw_data = download_data(_FLAGS.config_file_path)
     if raw_data:
-        preprocess_and_save(raw_data, output_dir)
+        preprocess_and_save(raw_data)
+
 
 if __name__ == '__main__':
-    app.run(main)
+    app.run(main)