Data Analyst &
Aspiring Data Engineer.

import json
import os
from datetime import datetime, timezone
from io import BytesIO

import boto3
import requests
from dotenv import load_dotenv


# ------------------------------------------------------------
# STEP 1: Load environment variables from the .env file.
# ------------------------------------------------------------
# The .env file stores private settings such as API keys, bucket names,
# and credentials. Keeping them outside the code is safer and cleaner.
load_dotenv()


# ------------------------------------------------------------
# STEP 2: Define the locations we want to fetch weather data for.
# ------------------------------------------------------------
# Each location has a city name and coordinates. The API uses the
# latitude and longitude to return weather for the correct place.
LOCATIONS = [
    {
        "city": "Cavite",
        "latitude": 14.4791,
        "longitude": 120.8960
    },
    {
        "city": "Manila",
        "latitude": 14.5995,
        "longitude": 120.9842
    },
    {
        "city": "Tokyo",
        "latitude": 35.6762,
        "longitude": 139.6503
    }
]


def get_r2_client():
    """
    Create a connection to Cloudflare R2.

    In simple terms:
    - R2 is where we store the raw weather JSON files.
    - boto3 is used because R2 works like an S3-compatible storage service.
    """
    # STEP 3: Read the R2 credentials from environment variables.
    account_id = os.getenv("R2_ACCOUNT_ID")
    access_key = os.getenv("R2_ACCESS_KEY_ID")
    secret_key = os.getenv("R2_SECRET_ACCESS_KEY")

    # STEP 4: Validate that all required R2 credentials are available.
    # This gives a clear error message instead of failing later in the upload.
    if not account_id or not access_key or not secret_key:
        raise ValueError("Missing R2 credentials in .env file.")

    # STEP 5: Build and return the R2 client.
    # This client is what the script uses to upload files to the bucket.
    return boto3.client(
        service_name="s3",
        endpoint_url=f"https://{account_id}.r2.cloudflarestorage.com",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name="auto"
    )


def fetch_raw_weather(city: str, latitude: float, longitude: float) -> dict:
    """
    Fetch the raw weather response for one city.

    In simple terms:
    - We ask Open-Meteo for the current weather.
    - We keep the full raw API response for traceability.
    - This is useful because raw data can be reprocessed later if needed.
    """
    # STEP 6: Set the API endpoint.
    url = "https://api.open-meteo.com/v1/forecast"

    # STEP 7: Tell the API what data we need.
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "current": ",".join([
            "temperature_2m",
            "relative_humidity_2m",
            "wind_speed_10m",
            "wind_direction_10m",
            "weather_code"
        ]),
        "timezone": "Asia/Manila"
    }

    # STEP 8: Call the API and stop if the request fails.
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()

    # STEP 9: Return the raw response with useful metadata.
    # Metadata helps us understand where the data came from and when it was fetched.
    return {
        "city": city,
        "latitude": latitude,
        "longitude": longitude,
        "fetched_at_utc": datetime.now(timezone.utc).isoformat(),
        "source_api": "Open-Meteo",
        "raw_response": response.json()
    }


def upload_json_to_r2(data: dict, object_key: str):
    """
    Upload a JSON payload to Cloudflare R2.

    In simple terms:
    - Convert the Python dictionary into a JSON file.
    - Upload that JSON file to the selected R2 bucket.
    """
    # STEP 10: Read the bucket name from the environment.
    bucket_name = os.getenv("R2_BUCKET_NAME")

    # STEP 11: Make sure the bucket name is configured.
    if not bucket_name:
        raise ValueError("Missing R2_BUCKET_NAME in .env file.")

    # STEP 12: Create the R2 client that will handle the upload.
    s3_client = get_r2_client()

    # STEP 13: Convert the dictionary into formatted JSON bytes.
    # BytesIO lets boto3 treat the JSON data like a file in memory.
    json_bytes = json.dumps(data, indent=2).encode("utf-8")
    file_obj = BytesIO(json_bytes)

    # STEP 14: Upload the JSON file to the R2 bucket.
    s3_client.upload_fileobj(
        file_obj,
        bucket_name,
        object_key
    )

    # STEP 15: Return the uploaded file path for logging and confirmation.
    return f"s3://{bucket_name}/{object_key}"


def main():
    """
    Run the full raw-data upload step.

    This step fetches weather data for all configured locations and saves
    the raw API result to Cloudflare R2.
    """
    # STEP 16: Create a timestamp for this pipeline run.
    # This helps make every uploaded file unique and easy to trace.
    run_time = datetime.now(timezone.utc)
    date_path = run_time.strftime("%Y/%m/%d")
    timestamp = run_time.strftime("%Y%m%d_%H%M%S")

    # STEP 17: Collect raw weather data for all cities.
    all_raw_data = []

    for location in LOCATIONS:
        raw_data = fetch_raw_weather(
            city=location["city"],
            latitude=location["latitude"],
            longitude=location["longitude"]
        )
        all_raw_data.append(raw_data)

    # STEP 18: Wrap the raw data in one pipeline payload.
    # This adds pipeline-level information such as run time and record count.
    payload = {
        "pipeline_name": "portfolio_weather_pipeline",
        "run_timestamp_utc": run_time.isoformat(),
        "record_count": len(all_raw_data),
        "data": all_raw_data
    }

    # STEP 19: Create the folder and file name in R2.
    # Example path: raw/weather/2026/06/23/weather_raw_20260623_120000.json
    object_key = f"raw/weather/{date_path}/weather_raw_{timestamp}.json"

    # STEP 20: Upload the payload to R2.
    uploaded_path = upload_json_to_r2(payload, object_key)

    # STEP 21: Print confirmation so the pipeline log is easy to understand.
    print("Raw weather data uploaded successfully.")
    print(f"Location: {uploaded_path}")


# This makes sure main() runs only when this file is executed directly.
if __name__ == "__main__":
    main()

import json
import os
from datetime import datetime, timezone

import boto3
import pymysql
from dotenv import load_dotenv


# ------------------------------------------------------------
# STEP 1: Load environment variables from the .env file.
# ------------------------------------------------------------
# The .env file stores private settings such as R2 credentials and
# database credentials. This keeps sensitive values out of the code.
load_dotenv()


def get_r2_client():
    """
    Create a connection to Cloudflare R2.

    In simple terms:
    - R2 stores the raw weather JSON files.
    - This function creates the client needed to read those files.
    """
    # STEP 2: Read R2 credentials from the environment.
    account_id = os.getenv("R2_ACCOUNT_ID")
    access_key = os.getenv("R2_ACCESS_KEY_ID")
    secret_key = os.getenv("R2_SECRET_ACCESS_KEY")

    # STEP 3: Check that the required R2 credentials exist.
    # This gives a clear error message if the .env file is incomplete.
    if not account_id or not access_key or not secret_key:
        raise ValueError("Missing R2 credentials in .env file.")

    # STEP 4: Build and return the R2 client.
    return boto3.client(
        service_name="s3",
        endpoint_url=f"https://{account_id}.r2.cloudflarestorage.com",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name="auto"
    )


def get_mysql_connection():
    """
    Create a connection to the MySQL database.

    In simple terms:
    - This is the bridge between the Python script and the database.
    - The script uses this connection to insert the cleaned weather rows.
    """
    # STEP 5: List the database settings required to connect to MySQL.
    required_env = [
        "DB_HOST",
        "DB_PORT",
        "DB_NAME",
        "DB_USER",
        "DB_PASSWORD"
    ]

    # STEP 6: Check if any required database settings are missing.
    missing = [key for key in required_env if not os.getenv(key)]

    if missing:
        raise ValueError(f"Missing DB credentials in .env: {missing}")

    # STEP 7: Connect to MySQL and return the connection object.
    # DictCursor returns rows as dictionaries, which are easier to work with.
    return pymysql.connect(
        host=os.getenv("DB_HOST"),
        port=int(os.getenv("DB_PORT", "3306")),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD"),
        database=os.getenv("DB_NAME"),
        cursorclass=pymysql.cursors.DictCursor
    )


def get_latest_raw_weather_object_key():
    """
    Find the newest raw weather file in the R2 bucket.

    In simple terms:
    - The pipeline saves raw files in the raw/weather/ folder.
    - This function checks that folder and picks the most recently uploaded file.
    """
    # STEP 8: Read the R2 bucket name and create the R2 client.
    bucket_name = os.getenv("R2_BUCKET_NAME")
    s3_client = get_r2_client()

    # STEP 9: List files inside the raw weather folder.
    response = s3_client.list_objects_v2(
        Bucket=bucket_name,
        Prefix="raw/weather/"
    )

    # STEP 10: Get the list of files returned by R2.
    objects = response.get("Contents", [])

    # STEP 11: Stop the script if no raw weather files are available.
    if not objects:
        raise FileNotFoundError("No raw weather files found in R2 bucket.")

    # STEP 12: Sort the files by upload time and select the newest one.
    latest_object = sorted(
        objects,
        key=lambda obj: obj["LastModified"],
        reverse=True
    )[0]

    # STEP 13: Return the file path/key of the newest raw file.
    return latest_object["Key"]


def read_json_from_r2(object_key):
    """
    Read a JSON file from Cloudflare R2.

    In simple terms:
    - Download the selected raw weather file.
    - Convert the JSON text into a Python dictionary.
    """
    # STEP 14: Read the bucket name and create the R2 client.
    bucket_name = os.getenv("R2_BUCKET_NAME")
    s3_client = get_r2_client()

    # STEP 15: Download the selected object from R2.
    response = s3_client.get_object(
        Bucket=bucket_name,
        Key=object_key
    )

    # STEP 16: Read the file content and convert it from JSON text to a dictionary.
    content = response["Body"].read().decode("utf-8")
    return json.loads(content)


def format_datetime_for_mysql(iso_datetime):
    """
    Convert an ISO timestamp into a MySQL-friendly datetime format.

    In simple terms:
    - APIs often return timestamps like 2026-06-23T12:00:00+00:00.
    - MySQL usually stores datetime values like 2026-06-23 12:00:00.
    """
    # STEP 17: Return None when the source timestamp is empty.
    if not iso_datetime:
        return None

    # STEP 18: Normalize timestamps ending in Z so Python can parse them.
    clean_value = iso_datetime.replace("Z", "+00:00")
    dt = datetime.fromisoformat(clean_value)

    # STEP 19: Format the datetime for MySQL.
    return dt.strftime("%Y-%m-%d %H:%M:%S")


def transform_weather_payload(raw_payload):
    """
    Convert raw API data into clean database rows.

    In simple terms:
    - The raw JSON contains nested data from the API.
    - This function flattens it into simple rows that match the MySQL table.
    """
    # STEP 20: Keep the original pipeline run timestamp for traceability.
    pipeline_run_id = raw_payload.get("run_timestamp_utc")

    # STEP 21: Record when this transform step processed the data.
    processed_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    # STEP 22: Prepare a list that will hold the cleaned rows.
    transformed_rows = []

    # STEP 23: Process each city's raw weather record.
    for item in raw_payload.get("data", []):
        raw_response = item.get("raw_response", {})
        current = raw_response.get("current", {})

        # STEP 24: Build one clean row for the database.
        # Each key here matches a column in the weather_pipeline_results table.
        row = {
            "pipeline_run_id": pipeline_run_id,
            "city": item.get("city"),
            "latitude": item.get("latitude"),
            "longitude": item.get("longitude"),
            "temperature": current.get("temperature_2m"),
            "humidity": current.get("relative_humidity_2m"),
            "wind_speed": current.get("wind_speed_10m"),
            "wind_direction": current.get("wind_direction_10m"),
            "weather_code": str(current.get("weather_code")),
            "source_api": item.get("source_api", "Open-Meteo"),
            "fetched_at": format_datetime_for_mysql(item.get("fetched_at_utc")),
            "processed_at": processed_at
        }

        transformed_rows.append(row)

    # STEP 25: Return all cleaned rows so they can be inserted into MySQL.
    return transformed_rows


def load_rows_to_mysql(rows):
    """
    Insert the transformed weather rows into MySQL.

    In simple terms:
    - Take the clean rows created by transform_weather_payload().
    - Save them into the weather_pipeline_results table.
    """
    # STEP 26: If there is no data, skip the database insert.
    if not rows:
        print("No rows to load.")
        return

    # STEP 27: Prepare the SQL insert statement.
    # The %(column_name)s placeholders are filled using the row dictionaries.
    insert_sql = """
        INSERT INTO weather_pipeline_results (
            pipeline_run_id,
            city,
            latitude,
            longitude,
            temperature,
            humidity,
            wind_speed,
            wind_direction,
            weather_code,
            source_api,
            fetched_at,
            processed_at
        )
       LUES (
            %(pipeline_run_id)s,
            %(city)s,
            %(latitude)s,
            %(longitude)s,
            %(temperature)s,
            %(humidity)s,
            %(wind_speed)s,
            %(wind_direction)s,
            %(weather_code)s,
            %(source_api)s,
            %(fetched_at)s,
            %(processed_at)s
        );
    """

    # STEP 28: Open the database connection.
    connection = get_mysql_connection()

    try:
        # STEP 29: Insert all rows in one batch for better performance.
        with connection.cursor() as cursor:
            cursor.executemany(insert_sql, rows)

        # STEP 30: Save the changes permanently in the database.
        connection.commit()
        print(f"Loaded {len(rows)} rows into MySQL.")

    finally:
        # STEP 31: Always close the database connection.
        # This prevents open connections from being left behind.
        connection.close()


def main():
    """
    Run the full transform-and-load step.

    This step reads the latest raw weather file from R2, cleans it,
    and loads the result into MySQL.
    """
    # STEP 32: Find the newest raw weather file in R2.
    latest_key = get_latest_raw_weather_object_key()
    print(f"Latest raw file found: {latest_key}")

    # STEP 33: Read the raw JSON payload from R2.
    raw_payload = read_json_from_r2(latest_key)

    # STEP 34: Convert the raw JSON into clean database-ready rows.
    transformed_rows = transform_weather_payload(raw_payload)
    print(f"Transformed {len(transformed_rows)} rows.")

    # STEP 35: Load the cleaned rows into MySQL.
    load_rows_to_mysql(transformed_rows)

    # STEP 36: Print a success message for the pipeline log.
    print("Transform and load step completed successfully.")


# This makes sure main() runs only when this file is executed directly.
if __name__ == "__main__":
    main()