home-automation/app/main.py

import logging
import os
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy.orm import Session

from app import models  # noqa: F401
from app.api.routes.api.config import router as api_config_router
from app.api.routes.api.data import router as api_data_router
from app.api.routes.api.energy import router as api_energy_router
from app.api.routes.api.energy_contracts import router as api_energy_contracts_router
from app.api.routes.api.expose import router as api_expose_router
from app.api.routes.api.meters import router as api_meters_router
from app.api.routes.api.modbus import router as api_modbus_router
from app.api.routes.api.session import router as api_session_router
from app.api.routes import status
from app.db import get_session_local
from app.api.routes.homeassistant import router as homeassistant_router
from app.api.routes.location import router as location_router
from app.api.routes.poo import router as poo_router
from app.api.routes.public_ip import router as public_ip_router
from app.api.routes.ticktick import router as ticktick_router
from app.config import get_settings
from app.integrations.mqtt import mqtt_manager
from app.services.auth import AuthBootstrapError, initialize_auth_schema
from app.services.config_page import build_runtime_settings, seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap
from app.services.dsmr_ingest import apply_dsmr_subscription
from app.services.public_ip import check_public_ipv4_and_notify
from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
from app.services.ha_discovery import publish_discovery, publish_states
from app.services.tibber_prices import refresh_prices
from app.services.energy_cost import compute_closed_periods
from app.services.timezone import local_tz
from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db

logger = logging.getLogger(__name__)

_REPO_ROOT = Path(__file__).resolve().parents[1]


def _get_spa_dist_dir() -> Path:
    env_val = os.environ.get("SPA_DIST_DIR")
    if env_val:
        return Path(env_val)
    return _REPO_ROOT / "frontend" / "dist"


def _run_scheduled_public_ip_check() -> None:
    session_local = get_session_local()
    session: Session = session_local()
    try:
        check_public_ipv4_and_notify(session, bootstrap_settings=get_settings())
    finally:
        session.close()


def _run_scheduled_modbus_poll() -> None:
    session_local = get_session_local()
    session: Session = session_local()
    try:
        poll_all_enabled_devices(session, bootstrap_settings=get_settings())
    finally:
        session.close()


def _run_scheduled_tibber_refresh() -> None:
    """Scheduled job: fetch Tibber 15-minute prices and upsert into tibber_price.

    Runs every hour so that:
    - Today's prices are available from startup.
    - Tomorrow's prices (published by Tibber around 13:00 CET / 11:00 UTC) are
      picked up within an hour of publication without requiring a server restart.

    The job is a no-op when:
    - No active energy contract with kind="tibber" exists.
    - The Tibber API token is empty in the runtime settings.

    Any client exceptions (auth failures, network errors) are caught and logged
    so that a single failed fetch does not crash the scheduler or affect the
    other background jobs.
    """
    session_local = get_session_local()
    session: Session = session_local()
    try:
        runtime_settings = build_runtime_settings(session, get_settings())
        refresh_prices(session, runtime_settings)
    except Exception:
        logger.exception("_run_scheduled_tibber_refresh: unexpected error")
    finally:
        session.close()


def _run_scheduled_energy_cost() -> None:
    """Scheduled job: compute billing records for all uncalculated closed 15-minute periods.

    Runs every minute so that a new period is picked up within 1 minute of
    closing.  The job is a no-op when:
    - No active energy contract with a version covering the period exists.
    - DSMR data has not yet arrived for the period boundaries.
    - The period's billing record already exists and is not degraded.

    Any unexpected exceptions are caught and logged so that a single failure
    does not crash the scheduler or affect the other background jobs.
    """
    session_local = get_session_local()
    session: Session = session_local()
    try:
        compute_closed_periods(session)
        # After billing periods are computed, push fresh energy-cost state values
        # to MQTT/HA.  publish_states is internally guarded by _should_publish
        # (MQTT disabled / not connected → no-op), so this never raises due to
        # unconfigured MQTT and does not block the billing job.
        try:
            from app.services.ha_discovery import publish_states
            publish_states(session)
        except Exception:
            logger.exception("_run_scheduled_energy_cost: publish_states failed (non-fatal)")
    except Exception:
        logger.exception("_run_scheduled_energy_cost: unexpected error")
    finally:
        session.close()


def _run_scheduled_ha_state_publish() -> None:
    """Periodic job: publish discovery configs + state + availability for all enabled exposed entities.

    Runs every 60 seconds.  When the MQTT broker is connected:
    - Publishes (or re-publishes) all HA Discovery configs (retained, idempotent).
    - Publishes the current state / availability for all enabled entities.

    This also serves as the reliable "publish discovery after connect" mechanism:
    because paho connects asynchronously, a synchronous call immediately after
    ``mqtt_manager.connect()`` would fire before the TCP handshake completes and
    be a no-op.  Instead, this periodic job picks it up within 60 seconds of the
    broker becoming available — retained payloads make repeated publishes harmless.

    Additionally, if MQTT is configured in DB but the manager is not yet connected
    (e.g. MQTT was enabled via UI after startup), this job attempts to connect so
    the user does not need to restart the server.
    """
    session_local = get_session_local()
    session: Session = session_local()
    try:
        runtime_settings = build_runtime_settings(session, get_settings())
        # Reconnect if MQTT is configured (in DB) but not yet connected.
        if mqtt_manager.is_configured(runtime_settings) and not mqtt_manager.is_connected:
            logger.info("_run_scheduled_ha_state_publish: MQTT configured but not connected — attempting connect.")
            mqtt_manager.connect(runtime_settings)
        publish_discovery(session)
        publish_states(session)
    except Exception:
        logger.exception("_run_scheduled_ha_state_publish: unexpected error")
    finally:
        session.close()


def _run_midnight_state_publish() -> None:
    """本地午夜后不久专门发布一次状态，让 *_today 的每日归零稳稳落在午夜之后
    （对 HA 钟慢几秒鲁棒）。best-effort：失败仅记日志，不影响调度器。"""
    session_local = get_session_local()
    session = session_local()
    try:
        from app.services.ha_discovery import publish_states
        publish_states(session)
    except Exception:
        logger.exception("_run_midnight_state_publish: failed (non-fatal)")
    finally:
        session.close()


def ensure_auth_db_ready() -> None:
    session_local = get_session_local()
    session: Session = session_local()
    try:
        validate_app_runtime_db(get_settings().app_database_url)
        initialize_auth_schema(session, get_settings())
        seed_missing_config_from_bootstrap(session, get_settings())
        sync_app_hostname_from_bootstrap(session, get_settings())
    except AppDatabaseAdoptionError as exc:
        raise RuntimeError(str(exc)) from exc
    except AuthBootstrapError as exc:
        raise RuntimeError(str(exc)) from exc
    finally:
        session.close()


def ensure_runtime_dirs() -> None:
    settings = get_settings()
    if settings.app_sqlite_path is not None:
        settings.app_sqlite_path.parent.mkdir(parents=True, exist_ok=True)


@asynccontextmanager
async def lifespan(_: FastAPI):
    ensure_runtime_dirs()
    ensure_auth_db_ready()
    scheduler = BackgroundScheduler(timezone="UTC")
    scheduler.add_job(
        _run_scheduled_public_ip_check,
        trigger=IntervalTrigger(hours=4),
        id="public-ip-check",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(
        _run_scheduled_modbus_poll,
        trigger=IntervalTrigger(seconds=BASE_POLL_TICK_SECONDS),
        id="modbus-poll",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Periodic HA state / availability publish (60-second fallback sweep).
    scheduler.add_job(
        _run_scheduled_ha_state_publish,
        trigger=IntervalTrigger(seconds=60),
        id="ha-state-publish",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Tibber price refresh: fetch today + tomorrow every hour.
    # The job is a no-op when no active tibber contract or token is configured,
    # so it is safe to register unconditionally.
    scheduler.add_job(
        _run_scheduled_tibber_refresh,
        trigger=IntervalTrigger(hours=1),
        id="tibber-refresh",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Energy cost billing: compute uncalculated closed 15-minute periods every minute.
    # The job is a no-op when no active contract or DSMR data is present, so it is
    # safe to register unconditionally.
    scheduler.add_job(
        _run_scheduled_energy_cost,
        trigger=IntervalTrigger(minutes=1),
        id="energy-cost",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Dedicated midnight publish: fire at local 00:00:10 so *_today grace (5 s) has
    # already elapsed and the day-rolled value is pushed to HA immediately, rather
    # than waiting for the next 60-second ha-state-publish sweep.
    scheduler.add_job(
        _run_midnight_state_publish,
        trigger=CronTrigger(hour=0, minute=0, second=10, timezone=local_tz()),
        id="midnight-today-publish",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.start()

    # MQTT: connect using DB-merged runtime settings so broker configured via UI
    # is picked up on restart (not just from env/bootstrap settings).
    # Discovery will be published by the first run of _run_scheduled_ha_state_publish
    # (within 60 s of startup), after the async paho handshake completes.
    _startup_session_local = get_session_local()
    _startup_session: Session = _startup_session_local()
    try:
        _startup_runtime_settings = build_runtime_settings(_startup_session, get_settings())
    finally:
        _startup_session.close()
    mqtt_manager.connect(_startup_runtime_settings)

    # DSMR ingest: subscribe to the configured MQTT topic when enabled.  The same
    # applier is called from PUT /api/config, so toggling DSMR via the UI takes
    # effect without an app restart.
    apply_dsmr_subscription(_startup_runtime_settings)

    yield

    # MQTT: clean shutdown before the process exits.
    mqtt_manager.disconnect()

    scheduler.shutdown(wait=False)


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(
        title=settings.app_name,
        debug=settings.app_debug,
        version="0.1.0",
        lifespan=lifespan,
        description=(
            "Home automation backend with auth, runtime config, Home Assistant "
            "integrations, TickTick integration, and SQLite-backed recorders."
        ),
    )

    static_dir = Path(__file__).parent / "static"
    app.mount("/static", StaticFiles(directory=static_dir), name="static")

    app.include_router(status.router)
    app.include_router(api_config_router)
    app.include_router(api_data_router)
    app.include_router(api_energy_router)
    app.include_router(api_energy_contracts_router)
    app.include_router(api_meters_router)
    app.include_router(api_expose_router)
    app.include_router(api_modbus_router)
    app.include_router(api_session_router)
    app.include_router(homeassistant_router)
    app.include_router(location_router)
    app.include_router(poo_router)
    app.include_router(public_ip_router)
    app.include_router(ticktick_router)

    # SPA hosting: mount frontend/dist if it exists and has index.html.
    # If the SPA dist is absent (e.g. backend-only CI), skip SPA serving entirely
    # so that pytest stays green with only the API routes registered.
    spa_dist = _get_spa_dist_dir()
    spa_index = spa_dist / "index.html"
    if spa_dist.is_dir() and spa_index.is_file():
        spa_assets = spa_dist / "assets"
        if spa_assets.is_dir():
            app.mount("/assets", StaticFiles(directory=spa_assets), name="spa-assets")

        # Resolve the dist root once so the containment check is fast and consistent.
        _spa_root = spa_dist.resolve()

        @app.get("/{full_path:path}", include_in_schema=False)
        async def spa_fallback(full_path: str, request: Request) -> FileResponse:  # noqa: RUF029
            # Explicit 404 for unmatched /api/* — never return index.html for API paths.
            if full_path.startswith("api/"):
                raise HTTPException(status_code=404, detail="not found")
            # Resolve candidate to an absolute path and verify it stays within the SPA
            # dist root.  Without this check, URL-encoded ".." sequences (e.g. "..%2f")
            # bypass Starlette's path parameter handling and allow arbitrary file reads.
            candidate = (spa_dist / full_path).resolve()
            if candidate.is_file() and candidate.is_relative_to(_spa_root):
                return FileResponse(candidate)
            # For any path outside the dist root, or for SPA client routes that don't
            # correspond to a real file, return index.html so the SPA router handles it.
            return FileResponse(spa_index)
    else:
        logger.warning(
            "SPA dist not found at %s — SPA hosting disabled (API-only mode).", spa_dist
        )

    return app


app = create_app()