fix missing addresses in teslamate

fix_addresses.py

#!/usr/bin/env python3

import os
import psycopg2
import requests
import json
import time

# --- Environment Loading ---
try:
    from dotenv import load_dotenv
    dotenv_path = os.path.join(os.path.dirname(__file__), '..', '.env')
    if os.path.exists(dotenv_path):
        load_dotenv(dotenv_path)
        print("Loaded environment variables from .env file.")
except ImportError:
    print("Warning: python-dotenv not found. Skipping .env file. Ensure environment variables are set.")

# --- Database Configuration ---
DB_HOST = os.environ.get("DATABASE_HOST", "localhost")
DB_PORT = os.environ.get("DATABASE_PORT", "5432")
DB_USER = os.environ.get("DATABASE_USER", "teslamate")
DB_NAME = os.environ.get("DATABASE_NAME", "teslamate")
DB_PASS = os.environ.get("DATABASE_PASS")

# --- OSM Configuration ---
OSM_TIMEOUT = int(os.environ.get("OSM_TIMEOUT", 5))
PROXY = os.environ.get("PROXY")

# --- Script Configuration ---
INTERVAL = int(os.environ.get("INTERVAL", 0))

OSM_REVERSE_URL = "https://nominatim.openstreetmap.org/reverse?lat={}&lon={}&format=json"

def get_db_connection():
    """Establishes a connection to the PostgreSQL database."""
    if not DB_PASS:
        raise ValueError("DATABASE_PASS environment variable must be set.")
    conn = psycopg2.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASS,
        dbname=DB_NAME,
    )
    return conn

def get_address_from_osm(lat, lon):
    """Fetches address data from OpenStreetMap Nominatim API."""
    url = OSM_REVERSE_URL.format(lat, lon)
    proxies = {"http": PROXY, "https": PROXY} if PROXY else None
    headers = {
        "User-Agent": "curl/7.79.1"
    }
    try:
        response = requests.get(url, timeout=OSM_TIMEOUT, proxies=proxies, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching address from OSM for lat={lat}, lon={lon}: {e}")
        return None

def find_and_fix_missing_addresses():
    """Finds and fixes missing addresses in the Teslamate database."""
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            # Find drives with missing start or end addresses
            cur.execute("""
                SELECT d.id, p_start.latitude, p_start.longitude, p_end.latitude, p_end.longitude, d.start_address_id, d.end_address_id
                FROM drives d
                JOIN positions p_start ON d.start_position_id = p_start.id
                JOIN positions p_end ON d.end_position_id = p_end.id
                WHERE d.start_address_id IS NULL OR d.end_address_id IS NULL;
            """)
            drives = cur.fetchall()

            for drive in drives:
                drive_id, start_lat, start_lon, end_lat, end_lon, start_address_id, end_address_id = drive

                if start_address_id is None:
                    fix_address(cur, "drives", "start_address_id", drive_id, start_lat, start_lon)

                if end_address_id is None:
                    fix_address(cur, "drives", "end_address_id", drive_id, end_lat, end_lon)

            conn.commit()
            
            # Find charging processes with missing addresses
            cur.execute("""
                SELECT cp.id, p.latitude, p.longitude
                FROM charging_processes cp
                JOIN positions p ON cp.position_id = p.id
                WHERE cp.address_id IS NULL;
            """)
            charges = cur.fetchall()

            for charge in charges:
                charge_id, lat, lon = charge
                fix_address(cur, "charging_processes", "address_id", charge_id, lat, lon)

            conn.commit()
            print("Address fixing process completed.")

    finally:
        conn.close()


def fix_address(cur, table_name, column_name, record_id, lat, lon):
    """Helper function to fix a single address."""
    osm_data = get_address_from_osm(lat, lon)
    if not osm_data:
        return

    osm_id = osm_data.get("osm_id")
    osm_type = osm_data.get("osm_type")

    if not osm_id or not osm_type:
        print(f"Could not get OSM ID or type for lat={lat}, lon={lon}")
        return

    # Check if address already exists
    cur.execute("SELECT id FROM addresses WHERE osm_id = %s AND osm_type = %s", (osm_id, osm_type))
    address_id_row = cur.fetchone()

    if address_id_row:
        address_id = address_id_row[0]
    else:
        # Insert new address
        address = osm_data.get("address", {})
        display_name = osm_data.get("display_name", "")
        name = display_name.split(",")[0].strip() if display_name else ""
        raw_data = json.dumps(osm_data)

        cur.execute("""
            INSERT INTO addresses (display_name, latitude, longitude, name, house_number, road, neighbourhood, city, county, postcode, state, state_district, country, raw, inserted_at, updated_at, osm_id, osm_type)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s)
            RETURNING id;
        """, (
            display_name, lat, lon, name,
            address.get("house_number"), address.get("road"), address.get("neighbourhood"),
            address.get("city"), address.get("county"), address.get("postcode"),
            address.get("state"), address.get("state_district"), address.get("country"),
            raw_data, osm_id, osm_type
        ))
        address_id = cur.fetchone()[0]
        print(f"Inserted new address with ID {address_id} for OSM ID {osm_id}")

    # Update the drive or charge record
    cur.execute(f"UPDATE {table_name} SET {column_name} = %s WHERE id = %s", (address_id, record_id))
    print(f"Updated {table_name} record {record_id} with address ID {address_id}")


if __name__ == "__main__":
    if not DB_PASS:
        print("Error: DATABASE_PASS environment variable is not set.")
    else:
        if INTERVAL > 0:
            while True:
                print(f"Running address fix job... next run in {INTERVAL} minutes.")
                find_and_fix_missing_addresses()
                time.sleep(INTERVAL * 60)
        else:
            find_and_fix_missing_addresses()

originated from https://github.com/WayneJz/teslamate-addr-fix, rewrote by gemini.