bsee-sodir-extraction（bsee 索迪尔萃取）

BSEE/SODIR Data Extraction Skill

Master data extraction from the Bureau of Safety and Environmental Enforcement (BSEE) and Norwegian Offshore Directorate (SODIR) for comprehensive offshore energy analysis.

When to Use This Skill

Use BSEE/SODIR data extraction when you need:

• Production data - Oil, gas, water production by field/well
• Well information - Directional surveys, completions, drilling data
• Field data - Reserves, operators, development status
• HSE data - Safety incidents, environmental compliance
• Economic analysis - NPV calculations using regulatory data
• Regulatory compliance - Track permits, violations, inspections

Data sources covered:

• BSEE (US Gulf of Mexico): Production, wells, platforms, safety
• SODIR (Norway): Fields, production, wells, discoveries
• NPD FactPages: Norwegian petroleum data (legacy)

Core Capabilities

1. BSEE Data Extraction

Available datasets:

• Production data (monthly oil/gas/water)
• Well data (API numbers, directional surveys)
• Platform/structure data
• Operator information
• Safety and incident data (OCS incidents)
• Environmental compliance

Base URLs:

BSEE_BASE_URLS = {
    "production": "https://www.data.bsee.gov/Production/",
    "well": "https://www.data.bsee.gov/Well/",
    "platform": "https://www.data.bsee.gov/Platform/",
    "company": "https://www.data.bsee.gov/Company/",
    "field": "https://www.data.bsee.gov/Field/",
    "incidents": "https://www.data.bsee.gov/Incidents/",
}

Production Data Extraction:

import pandas as pd
import requests
from pathlib import Path
from datetime import datetime
from typing import Optional

def fetch_bsee_production_data(
    year: int,
    output_dir: Path,
    area_code: Optional[str] = None
) -> pd.DataFrame:
    """
    Fetch BSEE production data for a given year.

    Args:
        year: Production year (e.g., 2024)
        output_dir: Directory to save downloaded data
        area_code: Optional area filter ('GC', 'MC', 'WR', etc.)

    Returns:
        DataFrame with production data
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # BSEE provides production data as downloadable files
    url = f"https://www.data.bsee.gov/Production/Files/ogoraan{year}.zip"

    # Download file
    response = requests.get(url, timeout=60)
    response.raise_for_status()

    zip_path = output_dir / f"production_{year}.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    # Extract and read
    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    # Read the extracted CSV
    csv_files = list(output_dir.glob(f"*{year}*.csv"))
    if not csv_files:
        raise FileNotFoundError(f"No CSV found for {year}")

    df = pd.read_csv(csv_files[0])

    # Filter by area if specified
    if area_code:
        df = df[df["AREA_CODE"] == area_code]

    # Clean column names
    df.columns = df.columns.str.strip().str.upper()

    # Add metadata
    df["EXTRACTION_DATE"] = datetime.now().isoformat()
    df["SOURCE"] = "BSEE"

    print(f"Fetched {len(df)} production records for {year}")

    return df

def aggregate_production_by_field(
    df: pd.DataFrame,
    time_period: str = "monthly"
) -> pd.DataFrame:
    """
    Aggregate production data by field.

    Args:
        df: Raw production DataFrame
        time_period: 'monthly', 'quarterly', or 'annual'

    Returns:
        Aggregated production DataFrame
    """
    # Group by field
    group_cols = ["FIELD_NAME", "AREA_CODE", "BLOCK_NUMBER"]

    if time_period == "monthly":
        group_cols.extend(["PRODUCTION_YEAR", "PRODUCTION_MONTH"])
    elif time_period == "quarterly":
        df["QUARTER"] = ((df["PRODUCTION_MONTH"] - 1) // 3) + 1
        group_cols.extend(["PRODUCTION_YEAR", "QUARTER"])
    else:  # annual
        group_cols.append("PRODUCTION_YEAR")

    # Aggregate
    agg_dict = {
        "OIL_BBL": "sum",
        "GAS_MCF": "sum",
        "WATER_BBL": "sum",
        "WELL_COUNT": "nunique" if "API_NUMBER" in df.columns else "count"
    }

    # Only aggregate columns that exist
    agg_dict = {k: v for k, v in agg_dict.items() if k in df.columns}

    aggregated = df.groupby(group_cols).agg(agg_dict).reset_index()

    return aggregated

# Example usage
production_2024 = fetch_bsee_production_data(
    year=2024,
    output_dir=Path("data/raw/bsee"),
    area_code="GC"  # Green Canyon
)

field_production = aggregate_production_by_field(
    production_2024,
    time_period="monthly"
)

print(field_production.head())

Well Data Extraction:

def fetch_bsee_well_data(
    api_number: Optional[str] = None,
    field_name: Optional[str] = None,
    output_dir: Path = Path("data/raw/bsee")
) -> pd.DataFrame:
    """
    Fetch BSEE well data.

    Args:
        api_number: Specific API number (14-digit)
        field_name: Filter by field name
        output_dir: Output directory

    Returns:
        DataFrame with well data
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # BSEE Well File download
    url = "https://www.data.bsee.gov/Well/Files/Well.zip"

    response = requests.get(url, timeout=120)
    response.raise_for_status()

    zip_path = output_dir / "well_data.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    # Read well data
    well_file = output_dir / "Well.csv"
    df = pd.read_csv(well_file)

    # Filter if specified
    if api_number:
        df = df[df["API_WELL_NUMBER"] == api_number]
    if field_name:
        df = df[df["FIELD_NAME"].str.contains(field_name, case=False, na=False)]

    return df

def fetch_directional_surveys(
    api_number: str,
    output_dir: Path = Path("data/raw/bsee")
) -> pd.DataFrame:
    """
    Fetch directional survey data for a well.

    Args:
        api_number: 14-digit API number
        output_dir: Output directory

    Returns:
        DataFrame with directional survey points
    """
    # BSEE provides directional survey data
    url = f"https://www.data.bsee.gov/Well/DirectionalSurvey/Files/DirectionalSurvey.zip"

    response = requests.get(url, timeout=120)
    response.raise_for_status()

    zip_path = output_dir / "directional_surveys.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    survey_file = output_dir / "DirectionalSurvey.csv"
    df = pd.read_csv(survey_file)

    # Filter by API
    df = df[df["API_WELL_NUMBER"] == api_number]

    # Sort by measured depth
    df = df.sort_values("MEASURED_DEPTH")

    return df

# Example: Get well data for a deepwater field
gom_wells = fetch_bsee_well_data(field_name="THUNDER HORSE")
print(f"Found {len(gom_wells)} wells in Thunder Horse field")

HSE Data Extraction:

def fetch_bsee_incident_data(
    start_year: int = 2020,
    end_year: int = 2024,
    output_dir: Path = Path("data/raw/bsee")
) -> pd.DataFrame:
    """
    Fetch BSEE incident/accident data.

    Args:
        start_year: Start year for data
        end_year: End year for data
        output_dir: Output directory

    Returns:
        DataFrame with incident records
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # Fetch incident data
    url = "https://www.data.bsee.gov/Incidents/Files/Accidents.zip"

    response = requests.get(url, timeout=120)
    response.raise_for_status()

    zip_path = output_dir / "incidents.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    incident_file = output_dir / "Accidents.csv"
    df = pd.read_csv(incident_file)

    # Convert date columns
    df["INCIDENT_DATE"] = pd.to_datetime(df["INCIDENT_DATE"], errors="coerce")

    # Filter by year range
    df = df[
        (df["INCIDENT_DATE"].dt.year >= start_year) &
        (df["INCIDENT_DATE"].dt.year <= end_year)
    ]

    return df

def calculate_operator_safety_score(
    incidents_df: pd.DataFrame,
    production_df: pd.DataFrame
) -> pd.DataFrame:
    """
    Calculate safety score per operator based on incidents per production.

    Args:
        incidents_df: Incident data
        production_df: Production data

    Returns:
        DataFrame with operator safety metrics
    """
    # Count incidents by operator
    incident_counts = incidents_df.groupby("OPERATOR_NAME").agg({
        "INCIDENT_ID": "count",
        "FATALITY_COUNT": "sum",
        "INJURY_COUNT": "sum"
    }).rename(columns={
        "INCIDENT_ID": "TOTAL_INCIDENTS",
        "FATALITY_COUNT": "TOTAL_FATALITIES",
        "INJURY_COUNT": "TOTAL_INJURIES"
    })

    # Sum production by operator
    production_totals = production_df.groupby("OPERATOR_NAME").agg({
        "OIL_BBL": "sum",
        "GAS_MCF": "sum"
    })

    # Merge
    safety_df = incident_counts.join(production_totals, how="outer").fillna(0)

    # Calculate incidents per million BOE
    safety_df["TOTAL_BOE"] = safety_df["OIL_BBL"] + safety_df["GAS_MCF"] / 6000
    safety_df["INCIDENTS_PER_MM_BOE"] = (
        safety_df["TOTAL_INCIDENTS"] / safety_df["TOTAL_BOE"] * 1e6
    )

    # Risk score (lower is better)
    safety_df["RISK_SCORE"] = (
        safety_df["INCIDENTS_PER_MM_BOE"] +
        safety_df["TOTAL_FATALITIES"] * 10 +
        safety_df["TOTAL_INJURIES"] * 2
    )

    return safety_df.sort_values("RISK_SCORE", ascending=False)

# Example: Safety analysis
incidents = fetch_bsee_incident_data(start_year=2020, end_year=2024)
production = fetch_bsee_production_data(year=2024, output_dir=Path("data/raw/bsee"))

safety_scores = calculate_operator_safety_score(incidents, production)
print("Operator Safety Scores (higher = more risk):")
print(safety_scores.head(10))

2. SODIR/NPD Data Extraction (Norway)

Available datasets:

• Field production (oil, gas, NGL, condensate)
• Well data (exploration, development)
• Discoveries and prospects
• Company information
• Pipeline and infrastructure

FactPages API:

import requests
import pandas as pd
from typing import Dict, List, Optional

class SODIRDataFetcher:
    """Fetch data from SODIR (Norwegian Offshore Directorate) FactPages."""

    BASE_URL = "https://factpages.sodir.no/api/v1"

    ENDPOINTS = {
        "fields": "/fields",
        "field_production": "/field-production-yearly",
        "wells": "/wells",
        "discoveries": "/discoveries",
        "companies": "/companies",
        "pipelines": "/pipelines",
        "facilities": "/facilities",
    }

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "EnergyDataAnalysis/1.0"
        })

    def _fetch(self, endpoint: str, params: Optional[Dict] = None) -> List[Dict]:
        """Fetch data from SODIR API."""
        url = f"{self.BASE_URL}{endpoint}"

        response = self.session.get(url, params=params, timeout=60)
        response.raise_for_status()

        return response.json()

    def get_all_fields(self) -> pd.DataFrame:
        """Get all Norwegian offshore fields."""
        data = self._fetch(self.ENDPOINTS["fields"])
        df = pd.DataFrame(data)
        return df

    def get_field_production(
        self,
        field_name: Optional[str] = None,
        start_year: Optional[int] = None,
        end_year: Optional[int] = None
    ) -> pd.DataFrame:
        """
        Get field production data.

        Args:
            field_name: Filter by field name
            start_year: Start year
            end_year: End year

        Returns:
            DataFrame with production data
        """
        data = self._fetch(self.ENDPOINTS["field_production"])
        df = pd.DataFrame(data)

        # Filter
        if field_name:
            df = df[df["fieldName"].str.contains(field_name, case=False, na=False)]
        if start_year:
            df = df[df["year"] >= start_year]
        if end_year:
            df = df[df["year"] <= end_year]

        return df

    def get_wells(
        self,
        well_type: Optional[str] = None,
        status: Optional[str] = None
    ) -> pd.DataFrame:
        """
        Get well data.

        Args:
            well_type: 'exploration', 'development', or 'other'
            status: Well status filter

        Returns:
            DataFrame with well data
        """
        data = self._fetch(self.ENDPOINTS["wells"])
        df = pd.DataFrame(data)

        if well_type:
            df = df[df["wellType"].str.lower() == well_type.lower()]
        if status:
            df = df[df["status"].str.contains(status, case=False, na=False)]

        return df

    def get_discoveries(self, status: Optional[str] = None) -> pd.DataFrame:
        """Get discoveries data."""
        data = self._fetch(self.ENDPOINTS["discoveries"])
        df = pd.DataFrame(data)

        if status:
            df = df[df["status"].str.contains(status, case=False, na=False)]

        return df

# Example usage
sodir = SODIRDataFetcher()

# Get all fields
fields = sodir.get_all_fields()
print(f"Total Norwegian fields: {len(fields)}")

# Get production for Johan Sverdrup
sverdrup_production = sodir.get_field_production(
    field_name="JOHAN SVERDRUP",
    start_year=2019
)
print(sverdrup_production)

# Get recent exploration wells
exploration_wells = sodir.get_wells(well_type="exploration")
print(f"Total exploration wells: {len(exploration_wells)}")

3. Combined Analysis

Cross-Basin Comparison:

import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from pathlib import Path

def compare_gom_norway_production(
    gom_data: pd.DataFrame,
    norway_data: pd.DataFrame,
    output_dir: Path = Path("reports")
) -> None:
    """
    Create comparative analysis of GOM vs Norway production.

    Args:
        gom_data: BSEE production data
        norway_data: SODIR production data
        output_dir: Report output directory
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # Aggregate by year
    gom_annual = gom_data.groupby("PRODUCTION_YEAR").agg({
        "OIL_BBL": "sum",
        "GAS_MCF": "sum"
    }).reset_index()
    gom_annual["REGION"] = "Gulf of Mexico"
    gom_annual["OIL_MM_BBL"] = gom_annual["OIL_BBL"] / 1e6
    gom_annual["GAS_BCF"] = gom_annual["GAS_MCF"] / 1e6

    norway_annual = norway_data.groupby("year").agg({
        "oilProduction": "sum",
        "gasProduction": "sum"
    }).reset_index()
    norway_annual.columns = ["PRODUCTION_YEAR", "OIL_MM_BBL", "GAS_BCF"]
    norway_annual["REGION"] = "Norway"

    # Create comparison chart
    fig = make_subplots(
        rows=1, cols=2,
        subplot_titles=["Oil Production (MM BBL)", "Gas Production (BCF)"]
    )

    # Oil production
    fig.add_trace(
        go.Bar(
            x=gom_annual["PRODUCTION_YEAR"],
            y=gom_annual["OIL_MM_BBL"],
            name="GOM Oil",
            marker_color="blue"
        ),
        row=1, col=1
    )
    fig.add_trace(
        go.Bar(
            x=norway_annual["PRODUCTION_YEAR"],
            y=norway_annual["OIL_MM_BBL"],
            name="Norway Oil",
            marker_color="red"
        ),
        row=1, col=1
    )

    # Gas production
    fig.add_trace(
        go.Bar(
            x=gom_annual["PRODUCTION_YEAR"],
            y=gom_annual["GAS_BCF"],
            name="GOM Gas",
            marker_color="lightblue"
        ),
        row=1, col=2
    )
    fig.add_trace(
        go.Bar(
            x=norway_annual["PRODUCTION_YEAR"],
            y=norway_annual["GAS_BCF"],
            name="Norway Gas",
            marker_color="pink"
        ),
        row=1, col=2
    )

    fig.update_layout(
        title="Gulf of Mexico vs Norway: Offshore Production Comparison",
        barmode="group",
        height=500
    )

    fig.write_html(output_dir / "gom_norway_comparison.html")
    print(f"Report saved to {output_dir / 'gom_norway_comparison.html'}")

4. NPV Analysis with Regulatory Data

import numpy as np
import numpy_financial as npf
import pandas as pd
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class EconomicAssumptions:
    """Economic assumptions for NPV calculation."""
    oil_price: float = 75.0      # $/bbl
    gas_price: float = 3.0       # $/mcf
    opex_per_boe: float = 15.0   # $/BOE
    capex_remaining: float = 0   # $ millions (for ongoing development)
    discount_rate: float = 0.10  # 10%
    royalty_rate: float = 0.125  # 12.5% federal royalty
    tax_rate: float = 0.21       # Corporate tax rate

def calculate_field_npv(
    production_df: pd.DataFrame,
    assumptions: EconomicAssumptions,
    forecast_years: int = 10
) -> Tuple[float, pd.DataFrame]:
    """
    Calculate NPV for a field based on BSEE production data.

    Args:
        production_df: Historical production data
        assumptions: Economic assumptions
        forecast_years: Years to forecast

    Returns:
        Tuple of (NPV, detailed cashflow DataFrame)
    """
    # Get latest year's production as baseline
    latest_year = production_df["PRODUCTION_YEAR"].max()
    baseline = production_df[production_df["PRODUCTION_YEAR"] == latest_year]

    annual_oil = baseline["OIL_BBL"].sum()
    annual_gas = baseline["GAS_MCF"].sum()

    # Simple decline curve (exponential decline)
    decline_rate = 0.10  # 10% annual decline

    cashflows = []

    for year in range(1, forecast_years + 1):
        # Decline production
        oil_prod = annual_oil * ((1 - decline_rate) ** year)
        gas_prod = annual_gas * ((1 - decline_rate) ** year)

        # Revenue
        oil_revenue = oil_prod * assumptions.oil_price
        gas_revenue = gas_prod * assumptions.gas_price
        gross_revenue = oil_revenue + gas_revenue

        # Royalties
        royalties = gross_revenue * assumptions.royalty_rate
        net_revenue = gross_revenue - royalties

        # Operating costs
        boe_produced = oil_prod + gas_prod / 6000
        opex = boe_produced * assumptions.opex_per_boe

        # EBITDA
        ebitda = net_revenue - opex

        # CapEx (if any)
        capex = assumptions.capex_remaining / forecast_years if year <= 3 else 0

        # Pre-tax income
        pretax_income = ebitda - capex

        # Taxes
        taxes = max(0, pretax_income * assumptions.tax_rate)

        # Net cash flow
        ncf = pretax_income - taxes

        cashflows.append({
            "Year": year,
            "Oil_BBL": oil_prod,
            "Gas_MCF": gas_prod,
            "Gross_Revenue_MM": gross_revenue / 1e6,
            "Royalties_MM": royalties / 1e6,
            "OPEX_MM": opex / 1e6,
            "CAPEX_MM": capex / 1e6,
            "Pre_Tax_MM": pretax_income / 1e6,
            "Taxes_MM": taxes / 1e6,
            "NCF_MM": ncf / 1e6
        })

    cashflow_df = pd.DataFrame(cashflows)

    # Calculate NPV
    ncf_series = [-assumptions.capex_remaining] + cashflow_df["NCF_MM"].tolist()
    npv = npf.npv(assumptions.discount_rate, ncf_series)

    return npv, cashflow_df

# Example: Calculate NPV for a GOM field
production = fetch_bsee_production_data(
    year=2024,
    output_dir=Path("data/raw/bsee")
)

# Filter to specific field
thunder_horse = production[
    production["FIELD_NAME"].str.contains("THUNDER HORSE", case=False, na=False)
]

assumptions = EconomicAssumptions(
    oil_price=75.0,
    gas_price=3.5,
    opex_per_boe=18.0,
    discount_rate=0.10
)

npv, cashflows = calculate_field_npv(thunder_horse, assumptions)

print(f"Thunder Horse NPV (10 year): ${npv:.1f} MM")
print("\nCashflow Summary:")
print(cashflows.to_string(index=False))

Complete Pipeline Example

"""
Complete BSEE/SODIR data extraction and analysis pipeline.
"""
import pandas as pd
from pathlib import Path
from datetime import datetime
import plotly.graph_objects as go

def run_extraction_pipeline(
    output_dir: Path = Path("data"),
    report_dir: Path = Path("reports")
) -> dict:
    """
    Run complete data extraction and analysis pipeline.

    Returns:
        Dictionary with extraction summary
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    report_dir.mkdir(parents=True, exist_ok=True)

    results = {
        "extraction_date": datetime.now().isoformat(),
        "datasets": {}
    }

    # 1. Extract BSEE Production Data
    print("Fetching BSEE production data...")
    try:
        bsee_production = fetch_bsee_production_data(
            year=2024,
            output_dir=output_dir / "raw" / "bsee"
        )
        bsee_production.to_csv(
            output_dir / "processed" / "bsee_production.csv",
            index=False
        )
        results["datasets"]["bsee_production"] = len(bsee_production)
    except Exception as e:
        print(f"BSEE production error: {e}")
        results["datasets"]["bsee_production"] = "error"

    # 2. Extract BSEE Well Data
    print("Fetching BSEE well data...")
    try:
        bsee_wells = fetch_bsee_well_data(
            output_dir=output_dir / "raw" / "bsee"
        )
        results["datasets"]["bsee_wells"] = len(bsee_wells)
    except Exception as e:
        print(f"BSEE wells error: {e}")
        results["datasets"]["bsee_wells"] = "error"

    # 3. Extract BSEE Incident Data
    print("Fetching BSEE incident data...")
    try:
        incidents = fetch_bsee_incident_data(start_year=2020, end_year=2024)
        incidents.to_csv(
            output_dir / "processed" / "bsee_incidents.csv",
            index=False
        )
        results["datasets"]["bsee_incidents"] = len(incidents)
    except Exception as e:
        print(f"BSEE incidents error: {e}")
        results["datasets"]["bsee_incidents"] = "error"

    # 4. Extract SODIR Data
    print("Fetching SODIR data...")
    try:
        sodir = SODIRDataFetcher()
        norway_fields = sodir.get_all_fields()
        norway_production = sodir.get_field_production(start_year=2020)

        norway_fields.to_csv(
            output_dir / "processed" / "sodir_fields.csv",
            index=False
        )
        norway_production.to_csv(
            output_dir / "processed" / "sodir_production.csv",
            index=False
        )

        results["datasets"]["sodir_fields"] = len(norway_fields)
        results["datasets"]["sodir_production"] = len(norway_production)
    except Exception as e:
        print(f"SODIR error: {e}")
        results["datasets"]["sodir"] = "error"

    # 5. Generate Summary Report
    print("Generating summary report...")
    generate_summary_report(results, report_dir)

    return results

def generate_summary_report(results: dict, report_dir: Path) -> None:
    """Generate HTML summary report."""

    html_content = f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>Energy Data Extraction Report</title>
        <style>
            body {{ font-family: Arial, sans-serif; margin: 40px; }}
            h1 {{ color: #2c3e50; }}
            table {{ border-collapse: collapse; width: 100%; }}
            th, td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
            th {{ background-color: #3498db; color: white; }}
            tr:nth-child(even) {{ background-color: #f2f2f2; }}
            .success {{ color: green; }}
            .error {{ color: red; }}
        </style>
    </head>
    <body>
        <h1>Energy Data Extraction Report</h1>
        <p><strong>Extraction Date:</strong> {results['extraction_date']}</p>

        <h2>Dataset Summary</h2>
        <table>
            <tr>
                <th>Dataset</th>
                <th>Records</th>
                <th>Status</th>
            </tr>
    """

    for dataset, count in results["datasets"].items():
        status_class = "error" if count == "error" else "success"
        status_text = "Error" if count == "error" else "Success"
        html_content += f"""
            <tr>
                <td>{dataset}</td>
                <td>{count if count != 'error' else 'N/A'}</td>
                <td class="{status_class}">{status_text}</td>
            </tr>
        """

    html_content += """
        </table>
    </body>
    </html>
    """

    report_path = report_dir / "extraction_summary.html"
    with open(report_path, "w") as f:
        f.write(html_content)

    print(f"Summary report saved to {report_path}")

# Run the pipeline
if __name__ == "__main__":
    results = run_extraction_pipeline()
    print("\nExtraction Complete!")
    print(f"Datasets extracted: {len(results['datasets'])}")

Best Practices

1. Rate Limiting

import time
from functools import wraps

def rate_limit(calls_per_minute: int = 30):
    """Decorator to rate limit API calls."""
    min_interval = 60.0 / calls_per_minute
    last_call = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=30)
def fetch_with_rate_limit(url: str) -> requests.Response:
    return requests.get(url)

2. Caching

from functools import lru_cache
from datetime import datetime, timedelta

@lru_cache(maxsize=100)
def cached_fetch(url: str, cache_hours: int = 24) -> pd.DataFrame:
    """Fetch with caching."""
    cache_file = Path(f".cache/{hash(url)}.parquet")

    if cache_file.exists():
        mtime = datetime.fromtimestamp(cache_file.stat().st_mtime)
        if datetime.now() - mtime < timedelta(hours=cache_hours):
            return pd.read_parquet(cache_file)

    # Fetch fresh data
    response = requests.get(url)
    df = pd.DataFrame(response.json())

    cache_file.parent.mkdir(exist_ok=True)
    df.to_parquet(cache_file)

    return df

3. Error Handling

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def robust_fetch(url: str) -> requests.Response:
    """Fetch with automatic retry on failure."""
    try:
        response = requests.get(url, timeout=60)
        response.raise_for_status()
        return response
    except requests.exceptions.RequestException as e:
        logger.error(f"Fetch failed for {url}: {e}")
        raise

Resources

• BSEE Data Center: https://www.data.bsee.gov/
• SODIR FactPages: https://factpages.sodir.no/
• BSEE API Documentation: https://www.data.bsee.gov/api-documentation
• NPD (legacy): https://www.npd.no/en/facts/

Use this skill for all energy regulatory data extraction in worldenergydata!