Northern-Thailand-Ping-River-Monitor/src/web_api.py

#!/usr/bin/env python3
"""
FastAPI web interface for water monitoring system
"""

import asyncio
import os
import secrets
import time
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from threading import Lock
from typing import Any, Dict, List, Optional

import requests
from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles

from .config import Config
from .health_check import (
    APIHealthCheck,
    DatabaseHealthCheck,
    HealthCheckManager,
    MemoryHealthCheck,
)
from .logging_config import get_logger, setup_logging
from .metrics import get_metrics_collector, increment_counter, set_gauge
from .postgres_history import PostgresHistory
from .schemas import (
    HealthResponse,
    MeasurementResponse,
    MetricsResponse,
    ScrapingStatusResponse,
    StationCreateModel,
    StationResponse,
    StationUpdateModel,
)
from .thaiwater import ThaiWaterClient
from .water_scraper_v3 import EnhancedWaterMonitorScraper

logger = get_logger(__name__)

# Simple thread-safe TTL cache for PostgreSQL history queries
HISTORY_CACHE: Dict[str, tuple] = {}
HISTORY_CACHE_LOCK = Lock()
HISTORY_TTL = 300  # 5 minutes

FORECAST_CACHE: Dict[str, tuple] = {}
FORECAST_CACHE_LOCK = Lock()
FORECAST_TTL = 900  # 15 minutes

DB_STATS_CACHE: Dict[str, tuple] = {}
DB_STATS_CACHE_LOCK = Lock()
DB_STATS_TTL = 300  # 5 minutes

# Admin API protection. Read/dashboard endpoints stay public; anything that
# mutates state or leaks configuration requires the X-API-Key header matching
# ADMIN_API_KEY. Secure by default: with no key configured, those endpoints
# are disabled entirely rather than open.
ADMIN_API_KEY = os.getenv("ADMIN_API_KEY")


def require_admin_key(x_api_key: Optional[str] = Header(None, alias="X-API-Key")):
    if not ADMIN_API_KEY:
        raise HTTPException(
            status_code=503,
            detail="Admin API disabled: ADMIN_API_KEY is not configured on the server",
        )
    if not x_api_key or not secrets.compare_digest(x_api_key, ADMIN_API_KEY):
        raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")


# Dashboard HTML is loaded once at import from src/static/dashboard.html.
_DASHBOARD_HTML_PATH = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "static", "dashboard.html"
)
try:
    with open(_DASHBOARD_HTML_PATH, encoding="utf-8") as _dashboard_file:
        DASHBOARD_HTML = _dashboard_file.read()
except OSError as _dashboard_error:  # pragma: no cover - defensive fallback
    logger.error(f"Could not load dashboard HTML: {_dashboard_error}")
    DASHBOARD_HTML = "<h1>Northern Thailand Ping River Monitor API</h1><p>See <code>/docs</code>.</p>"


# Global application state
app_state = {
    "scraper": None,
    "health_manager": None,
    "scraping_task": None,
    "is_scraping": False,
    "scraping_stats": {
        "total_runs": 0,
        "successful_runs": 0,
        "failed_runs": 0,
        "last_run": None,
        "next_run": None,
    },
}


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager"""
    # Startup
    logger.info("Starting Water Monitor API...")

    # Initialize configuration
    try:
        Config.validate_config()
        logger.info("Configuration validated successfully")
    except Exception as e:
        logger.error(f"Configuration validation failed: {e}")
        raise

    # Initialize scraper
    db_config = Config.get_database_config()
    app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)

    # Initialize health checks
    health_manager = HealthCheckManager()
    health_manager.add_check(DatabaseHealthCheck(app_state["scraper"].db_adapter))
    health_manager.add_check(
        APIHealthCheck(Config.API_URL, app_state["scraper"].session)
    )
    health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000))
    app_state["health_manager"] = health_manager

    # Start background scraping task
    app_state["scraping_task"] = asyncio.create_task(background_scraping_task())

    logger.info("Water Monitor API started successfully")

    yield

    # Shutdown
    logger.info("Shutting down Water Monitor API...")

    if app_state["scraping_task"]:
        app_state["scraping_task"].cancel()
        try:
            await app_state["scraping_task"]
        except asyncio.CancelledError:
            pass

    logger.info("Water Monitor API shutdown complete")


# Create FastAPI app
app = FastAPI(
    title="Northern Thailand Ping River Monitor API",
    description="Real-time water level monitoring system for Northern Thailand's Ping River Basin stations",
    version="3.1.3",
    lifespan=lifespan,
)
app.mount(
    "/static",
    StaticFiles(directory=os.path.dirname(_DASHBOARD_HTML_PATH)),
    name="static",
)

# Add CORS middleware.
# Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured
# we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination);
# credentials are only enabled when explicit origins are provided.
_cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"]
_cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS)
app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_credentials=_cors_allow_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)


async def background_scraping_task():
    """Background task for periodic data scraping"""
    while True:
        try:
            if not app_state["is_scraping"]:
                app_state["is_scraping"] = True

                # Run scraping cycle
                scraper = app_state["scraper"]
                if scraper:
                    logger.info("Starting background scraping cycle")
                    start_time = datetime.now()

                    try:
                        # run_scraping_cycle() does blocking network/DB I/O and time.sleep
                        # retries; run it in a thread so it doesn't freeze the event loop.
                        result = await asyncio.get_event_loop().run_in_executor(
                            None, scraper.run_scraping_cycle
                        )

                        # Update stats
                        app_state["scraping_stats"]["total_runs"] += 1
                        app_state["scraping_stats"]["last_run"] = start_time

                        if result:
                            app_state["scraping_stats"]["successful_runs"] += 1
                            increment_counter("scraping_cycles_successful")
                            logger.info(
                                "Background scraping cycle completed successfully"
                            )
                        else:
                            app_state["scraping_stats"]["failed_runs"] += 1
                            increment_counter("scraping_cycles_failed")
                            logger.warning(
                                "Background scraping cycle completed with no new data"
                            )

                        # Update metrics
                        set_gauge("last_scraping_timestamp", start_time.timestamp())

                    except Exception as e:
                        app_state["scraping_stats"]["failed_runs"] += 1
                        increment_counter("scraping_cycles_failed")
                        logger.error(f"Background scraping cycle failed: {e}")

                app_state["is_scraping"] = False

            # Calculate next run time
            interval_seconds = Config.SCRAPING_INTERVAL_HOURS * 3600
            app_state["scraping_stats"]["next_run"] = datetime.now() + timedelta(
                seconds=interval_seconds
            )

            # Wait for next cycle
            await asyncio.sleep(interval_seconds)

        except asyncio.CancelledError:
            logger.info("Background scraping task cancelled")
            break
        except Exception as e:
            logger.error(f"Error in background scraping task: {e}")
            await asyncio.sleep(60)  # Wait a minute before retrying


# API Routes


@app.get("/", response_class=HTMLResponse)
async def root():
    """Root endpoint with basic dashboard"""
    return HTMLResponse(content=DASHBOARD_HTML)


@app.get("/health", response_model=HealthResponse)
async def get_health():
    """Get system health status"""
    increment_counter("api_requests", labels={"endpoint": "health"})

    health_manager = app_state["health_manager"]
    if not health_manager:
        raise HTTPException(status_code=503, detail="Health manager not initialized")

    # Run health checks (populates state read by get_health_summary)
    health_manager.run_all_checks()
    summary = health_manager.get_health_summary()

    return HealthResponse(**summary)


@app.get("/metrics", response_model=MetricsResponse)
async def get_metrics():
    """Get application metrics"""
    increment_counter("api_requests", labels={"endpoint": "metrics"})

    metrics_collector = get_metrics_collector()
    metrics = metrics_collector.get_all_metrics()

    return MetricsResponse(**metrics)


@app.get("/stations", response_model=List[StationResponse])
async def get_stations():
    """Get list of all monitoring stations"""
    increment_counter("api_requests", labels={"endpoint": "stations"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    stations = []
    for station_id, station_info in scraper.station_mapping.items():
        stations.append(
            StationResponse(
                station_id=int(station_id),
                station_code=station_info["code"],
                thai_name=station_info["thai_name"],
                english_name=station_info["english_name"],
                latitude=station_info.get("latitude"),
                longitude=station_info.get("longitude"),
                status="active",
            )
        )

    return stations


@app.post(
    "/stations",
    response_model=StationResponse,
    dependencies=[Depends(require_admin_key)],
)
async def create_station(station: StationCreateModel):
    """Create a new monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "create_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    try:
        # Find next available station ID
        existing_ids = [int(sid) for sid in scraper.station_mapping.keys()]
        new_station_id = max(existing_ids) + 1 if existing_ids else 1

        # Add to station mapping and persist
        new_key = str(new_station_id)
        scraper.station_mapping[new_key] = {
            "code": station.station_code,
            "thai_name": station.thai_name,
            "english_name": station.english_name,
            "latitude": station.latitude,
            "longitude": station.longitude,
            "geohash": station.geohash,
        }
        if not scraper.save_stations():
            scraper.station_mapping.pop(new_key, None)
            raise HTTPException(status_code=500, detail="Failed to persist new station")

        logger.info(
            f"Created new station: {station.station_code} ({station.english_name})"
        )

        return StationResponse(
            station_id=new_station_id,
            station_code=station.station_code,
            thai_name=station.thai_name,
            english_name=station.english_name,
            latitude=station.latitude,
            longitude=station.longitude,
            geohash=station.geohash,
            status=station.status,
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error creating station: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.put(
    "/stations/{station_id}",
    response_model=StationResponse,
    dependencies=[Depends(require_admin_key)],
)
async def update_station(station_id: int, updates: StationUpdateModel):
    """Update an existing monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "update_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    try:
        station_info = scraper.station_mapping[station_key]
        original = dict(station_info)  # snapshot for rollback if persistence fails

        # Update fields if provided
        if updates.thai_name is not None:
            station_info["thai_name"] = updates.thai_name
        if updates.english_name is not None:
            station_info["english_name"] = updates.english_name
        if updates.latitude is not None:
            station_info["latitude"] = updates.latitude
        if updates.longitude is not None:
            station_info["longitude"] = updates.longitude
        if updates.geohash is not None:
            station_info["geohash"] = updates.geohash

        if not scraper.save_stations():
            scraper.station_mapping[station_key] = original
            raise HTTPException(
                status_code=500, detail="Failed to persist station update"
            )

        logger.info(f"Updated station {station_id}: {station_info['code']}")

        return StationResponse(
            station_id=station_id,
            station_code=station_info["code"],
            thai_name=station_info["thai_name"],
            english_name=station_info["english_name"],
            latitude=station_info.get("latitude"),
            longitude=station_info.get("longitude"),
            geohash=station_info.get("geohash"),
            status=updates.status or "active",
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error updating station {station_id}: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/stations/{station_id}", dependencies=[Depends(require_admin_key)])
async def delete_station(station_id: int):
    """Delete a monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "delete_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    try:
        station_info = scraper.station_mapping.pop(station_key)

        if not scraper.save_stations():
            scraper.station_mapping[station_key] = station_info  # restore
            raise HTTPException(
                status_code=500, detail="Failed to persist station deletion"
            )

        logger.info(f"Deleted station {station_id}: {station_info['code']}")

        return {"message": f"Station {station_info['code']} deleted successfully"}

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error deleting station {station_id}: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/stations/{station_id}", response_model=StationResponse)
async def get_station(station_id: int):
    """Get details of a specific monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "get_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    station_info = scraper.station_mapping[station_key]

    return StationResponse(
        station_id=station_id,
        station_code=station_info["code"],
        thai_name=station_info["thai_name"],
        english_name=station_info["english_name"],
        latitude=station_info.get("latitude"),
        longitude=station_info.get("longitude"),
        geohash=station_info.get("geohash"),
        status="active",
    )


def _to_measurement_response(measurement: Dict[str, Any]) -> MeasurementResponse:
    """Map a raw measurement dict from a DB adapter to the API response model.

    ``discharge`` is optional in the data (some stations report only level), so
    it is read with ``.get`` rather than assumed present.
    """
    return MeasurementResponse(
        timestamp=measurement["timestamp"],
        station_code=measurement["station_code"],
        station_name_en=measurement["station_name_en"],
        station_name_th=measurement["station_name_th"],
        water_level=measurement["water_level"],
        discharge=measurement.get("discharge"),
        discharge_percent=measurement.get("discharge_percent"),
        status=measurement.get("status", "active"),
    )


@app.get("/sensors/thaiwater")
async def get_thaiwater_sensors():
    """Get current ThaiWater water-level sensors in the Ping basin."""
    increment_counter("api_requests", labels={"endpoint": "thaiwater_sensors"})
    try:
        client = ThaiWaterClient(
            api_key=Config.THAIWATER_API_KEY,
            timeout=Config.REQUEST_TIMEOUT,
        )
        return await asyncio.to_thread(client.fetch_ping_sensors)
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except requests.RequestException as error:
        logger.error(f"Error fetching ThaiWater sensors: {error}")
        raise HTTPException(status_code=502, detail="ThaiWater API unavailable")


@app.get("/measurements/history/{station_code}")
async def get_postgres_history(
    station_code: str,
    hours: int = Query(168, ge=1),
    limit: int = Query(50000, ge=1, le=100000),
):
    """Get historical measurements for a station from the configured database."""
    cache_key = f"{station_code}:{hours}:{limit}"
    now = time.monotonic()
    with HISTORY_CACHE_LOCK:
        cached = HISTORY_CACHE.get(cache_key)
        if cached and now - cached[0] < HISTORY_TTL:
            return cached[1]
    try:
        db_config = Config.get_database_config()
        end_time = datetime.now()
        if db_config["type"] == "postgresql":
            history = PostgresHistory(db_config["connection_string"])
            data = await asyncio.to_thread(
                history.station_history,
                station_code,
                end_time - timedelta(hours=hours),
                end_time,
                limit,
            )
        else:
            scraper = app_state["scraper"]
            if not scraper or not scraper.db_adapter:
                raise RuntimeError("Database not available")
            rows = await asyncio.to_thread(
                scraper.db_adapter.get_measurements_by_timerange,
                end_time - timedelta(hours=hours),
                end_time,
                [station_code],
            )
            # adapter returns newest-first; keep the newest `limit` rows, chart wants ascending
            data = list(reversed(rows[:limit]))
        with HISTORY_CACHE_LOCK:
            HISTORY_CACHE[cache_key] = (now, data)
        return data
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except Exception as error:
        logger.error(f"Error fetching measurement history: {error}")
        raise HTTPException(status_code=502, detail="Measurement history unavailable")


@app.get("/forecast")
async def get_flood_forecasts():
    """Flood-risk forecasts per station for the 6/12/24 h horizons."""
    increment_counter("api_requests", labels={"endpoint": "forecast"})
    now = time.monotonic()
    with FORECAST_CACHE_LOCK:
        cached = FORECAST_CACHE.get("all")
        if cached and now - cached[0] < FORECAST_TTL:
            return cached[1]
    try:
        from .ml.predict import get_latest_forecasts
    except ImportError as error:
        raise HTTPException(status_code=503, detail=f"Forecasting unavailable: {error}")
    try:
        data = await asyncio.to_thread(get_latest_forecasts)
    except FileNotFoundError:
        raise HTTPException(status_code=503, detail="No trained flood models found")
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except Exception as error:
        logger.error(f"Error computing flood forecasts: {error}")
        raise HTTPException(status_code=502, detail="Flood forecast unavailable")
    with FORECAST_CACHE_LOCK:
        FORECAST_CACHE["all"] = (now, data)
    return data


@app.get("/measurements/latest", response_model=List[MeasurementResponse])
async def get_latest_measurements(limit: int = 100):
    """Get latest measurements from all stations"""
    increment_counter("api_requests", labels={"endpoint": "measurements_latest"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    try:
        measurements = scraper.get_latest_data(limit=limit)

        return [_to_measurement_response(m) for m in measurements]

    except Exception as e:
        logger.error(f"Error fetching latest measurements: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get(
    "/measurements/station/{station_code}", response_model=List[MeasurementResponse]
)
async def get_station_measurements(
    station_code: str, hours: int = 24, limit: int = 1000
):
    """Get measurements for a specific station"""
    increment_counter("api_requests", labels={"endpoint": "measurements_station"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    try:
        # Get measurements for the specified time range
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours)

        measurements = scraper.db_adapter.get_measurements_by_timerange(
            start_time, end_time, station_codes=[station_code]
        )

        # Limit results
        measurements = measurements[:limit]

        return [_to_measurement_response(m) for m in measurements]

    except Exception as e:
        logger.error(f"Error fetching station measurements: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/stats")
async def get_database_stats():
    """Get database coverage statistics (totals, date range, hourly coverage)"""
    increment_counter("api_requests", labels={"endpoint": "api_stats"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    now = time.monotonic()
    with DB_STATS_CACHE_LOCK:
        cached = DB_STATS_CACHE.get("all")
        if cached and now - cached[0] < DB_STATS_TTL:
            return cached[1]

    try:
        stats = await asyncio.to_thread(scraper.db_adapter.get_database_stats)
    except Exception as e:
        logger.error(f"Error fetching database stats: {e}")
        raise HTTPException(status_code=500, detail=str(e))

    if stats is None:
        raise HTTPException(status_code=503, detail="Database statistics unavailable")

    first_ts = stats["first_timestamp"]
    last_ts = stats["last_timestamp"]
    data = {
        "total_measurements": stats["total_measurements"],
        "station_count": stats["station_count"],
        "first_timestamp": first_ts.isoformat(),
        "last_timestamp": last_ts.isoformat(),
        "days_spanned": (last_ts.date() - first_ts.date()).days + 1,
        "coverage_percent": stats["coverage_percent"],
    }
    with DB_STATS_CACHE_LOCK:
        DB_STATS_CACHE["all"] = (now, data)
    return data


@app.post("/scrape/trigger", dependencies=[Depends(require_admin_key)])
async def trigger_scraping(background_tasks: BackgroundTasks):
    """Trigger manual data scraping"""
    increment_counter("api_requests", labels={"endpoint": "scrape_trigger"})

    if app_state["is_scraping"]:
        raise HTTPException(status_code=409, detail="Scraping already in progress")

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    def run_scraping():
        """Background task to run scraping"""
        try:
            app_state["is_scraping"] = True
            logger.info("Manual scraping triggered via API")

            result = scraper.run_scraping_cycle()

            # Update stats
            app_state["scraping_stats"]["total_runs"] += 1
            app_state["scraping_stats"]["last_run"] = datetime.now()

            if result:
                app_state["scraping_stats"]["successful_runs"] += 1
                increment_counter("manual_scraping_successful")
            else:
                app_state["scraping_stats"]["failed_runs"] += 1
                increment_counter("manual_scraping_failed")

        except Exception as e:
            app_state["scraping_stats"]["failed_runs"] += 1
            increment_counter("manual_scraping_failed")
            logger.error(f"Manual scraping failed: {e}")
        finally:
            app_state["is_scraping"] = False

    background_tasks.add_task(run_scraping)

    return {"message": "Scraping triggered", "status": "started"}


@app.get("/scraping/status", response_model=ScrapingStatusResponse)
async def get_scraping_status():
    """Get current scraping status"""
    increment_counter("api_requests", labels={"endpoint": "scraping_status"})

    stats = app_state["scraping_stats"]

    return ScrapingStatusResponse(
        is_running=app_state["is_scraping"],
        last_run=stats["last_run"],
        next_run=stats["next_run"],
        total_runs=stats["total_runs"],
        successful_runs=stats["successful_runs"],
        failed_runs=stats["failed_runs"],
    )


@app.get("/config", dependencies=[Depends(require_admin_key)])
async def get_config():
    """Get current configuration (sensitive data masked)"""
    increment_counter("api_requests", labels={"endpoint": "config"})

    config = Config.get_all_settings()

    # Mask sensitive information
    for key in config:
        if "password" in key.lower() or "secret" in key.lower():
            if config[key]:
                config[key] = "*" * 8

    return config


if __name__ == "__main__":
    import uvicorn

    # Setup logging
    setup_logging(
        log_level=Config.LOG_LEVEL,
        log_file=Config.LOG_FILE,
        enable_console=True,
        enable_colors=True,
    )

    # Run the API server
    uvicorn.run(
        "web_api:app", host="0.0.0.0", port=8000, reload=False, log_config=None
    )  # Use our custom logging