scripts/fetch_bls_schema.py

#!/usr/bin/env python3
"""
Fetch raw BLS (Bureau of Labor Statistics) data and output the schema.
Shows raw API response structure for CPI and Unemployment data.
"""

import asyncio
import json
import os
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv
import httpx

# Load environment variables
env_paths = [
    Path.home() / ".env",
    Path(__file__).parent.parent / ".env",
]
for env_path in env_paths:
    if env_path.exists():
        load_dotenv(env_path)
        break

BLS_API_KEY = os.getenv("BLS_API_KEY")
BLS_BASE_URL = "https://api.bls.gov/publicAPI/v2/timeseries/data/"

# BLS Series IDs
SERIES = {
    "cpi": "CUUR0000SA0",        # CPI-U All items
    "unemployment": "LNS14000000"  # Unemployment rate
}


def print_table(title: str, rows: list, col_widths: list = None):
    """Print ASCII table."""
    if not rows:
        return

    # Calculate column widths
    if col_widths is None:
        col_widths = []
        for col in range(len(rows[0])):
            width = max(len(str(row[col])) for row in rows)
            col_widths.append(width)

    # Print header
    print(f"\n{title}")

    # Top border
    line = "┌" + "┬".join("─" * (w + 2) for w in col_widths) + "┐"
    print(line)

    # Header row
    header = rows[0]
    row_str = "│" + "│".join(f" {str(header[i]).ljust(col_widths[i])} " for i in range(len(header))) + "│"
    print(row_str)

    # Separator
    line = "├" + "┼".join("─" * (w + 2) for w in col_widths) + "┤"
    print(line)

    # Data rows
    for row in rows[1:]:
        row_str = "│" + "│".join(f" {str(row[i]).ljust(col_widths[i])} " for i in range(len(row))) + "│"
        print(row_str)

    # Bottom border
    line = "└" + "┴".join("─" * (w + 2) for w in col_widths) + "┘"
    print(line)


async def fetch_bls_data(series_ids: list) -> dict:
    """Fetch data from BLS API."""
    current_year = datetime.now().year

    try:
        async with httpx.AsyncClient() as client:
            payload = {
                "seriesid": series_ids,
                "startyear": str(current_year - 2),
                "endyear": str(current_year)
            }

            # Add API key if available (for v2 with higher limits)
            if BLS_API_KEY:
                payload["registrationkey"] = BLS_API_KEY

            headers = {"Content-Type": "application/json"}
            response = await client.post(BLS_BASE_URL, json=payload, headers=headers, timeout=15)
            return response.json()
    except Exception as e:
        return {"error": str(e)}


async def main():
    print("BLS Data Schema")
    print("=" * 60)
    print()
    print("Endpoint: https://api.bls.gov/publicAPI/v2/timeseries/data/")
    print("Method: POST with JSON payload")
    print()
    print("Series IDs:")
    print("  - CUUR0000SA0: CPI-U All items (Consumer Price Index)")
    print("  - LNS14000000: Unemployment Rate")
    print()

    print("Fetching CPI and Unemployment data...")
    data = await fetch_bls_data(list(SERIES.values()))

    if "error" in data:
        print(f"ERROR: {data}")
        return

    if data.get("status") != "REQUEST_SUCCEEDED":
        print(f"ERROR: {data.get('message', 'Unknown error')}")
        return

    print()
    print("=" * 60)
    print()

    # Print raw API response structure
    print("Raw API Response Structure")
    print("-" * 40)

    # Request payload
    rows = [["field", "description"]]
    rows.append(["seriesid[]", "Array of BLS series IDs to fetch"])
    rows.append(["startyear", "Start year for data range"])
    rows.append(["endyear", "End year for data range"])
    rows.append(["registrationkey", "Optional API key for higher limits"])
    print_table("Request Payload", rows)

    # Response metadata
    rows = [["field", "value"]]
    rows.append(["status", data.get("status", "")])
    rows.append(["responseTime", str(data.get("responseTime", ""))])
    rows.append(["message[]", "Array of status messages"])
    print_table("Response Metadata", rows)

    # Results structure
    results = data.get("Results", {})
    series_list = results.get("series", [])

    rows = [["field", "description"]]
    rows.append(["Results.series[]", f"Array of series data (count: {len(series_list)})"])
    print_table("Results Structure", rows)

    # Series data structure
    if series_list:
        sample_series = series_list[0]
        rows = [["field", "value"]]
        rows.append(["seriesID", sample_series.get("seriesID", "")])
        rows.append(["data[]", f"Array of observations (count: {len(sample_series.get('data', []))})"])
        print_table("series[0] (Series Structure)", rows)

        # Data observation structure
        data_obs = sample_series.get("data", [])
        if data_obs:
            sample_obs = data_obs[0]
            rows = [["field", "value"]]
            rows.append(["year", sample_obs.get("year", "")])
            rows.append(["period", sample_obs.get("period", "")])
            rows.append(["periodName", sample_obs.get("periodName", "")])
            rows.append(["value", sample_obs.get("value", "")])
            rows.append(["footnotes[]", str(sample_obs.get("footnotes", []))])
            print_table("data[0] (Observation Structure)", rows)

    # Field descriptions
    print()
    print()
    print("Field Descriptions")
    print("-" * 40)

    rows = [["field", "description"]]
    rows.append(["seriesID", "BLS series identifier"])
    rows.append(["year", "4-digit year (e.g., 2025)"])
    rows.append(["period", "Period code (M01-M12 for monthly, A01 for annual)"])
    rows.append(["periodName", "Human-readable period (January, February, etc.)"])
    rows.append(["value", "Data value as string"])
    rows.append(["footnotes", "Array of footnote codes"])
    print_table("Field Descriptions", rows)

    # Series data
    print()
    print()
    print("Series Data")
    print("-" * 40)

    for series in series_list:
        series_id = series.get("seriesID", "")
        series_name = "CPI-U All Items" if series_id == "CUUR0000SA0" else "Unemployment Rate"
        data_obs = series.get("data", [])

        rows = [["field", "value"]]
        rows.append(["series_id", series_id])
        rows.append(["name", series_name])

        if data_obs:
            latest = data_obs[0]
            rows.append(["period", f"{latest.get('year')}-{latest.get('periodName', latest.get('period'))}"])
            rows.append(["value", latest.get("value", "")])
        print_table(series_name, rows)

    # Save raw JSON
    output_path = Path(__file__).parent.parent / "docs" / "bls_raw.json"
    with open(output_path, 'w') as f:
        json.dump(data, f, indent=2, default=str)
    print(f"\nRaw JSON saved to: {output_path}")


if __name__ == "__main__":
    asyncio.run(main())