app.py

import os
import uvicorn
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, status, HTTPException
from fastapi.responses import JSONResponse, FileResponse
from fastapi.exceptions import RequestValidationError
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
from utils.database import SQLiteDb
from routes import health, authentication, adb, tokens
from routes.authentication import ADMIN_USERNAME, ADMIN_PASSWORD
from routes.adb import adb as adb_library
from models.errors import ErrorResponse
from models.authentication import DEFAULT_PASSWORD
from utils.logger import create_logger
from utils.adb_wireless import start_terminal_pairing_session
from utils.scheduler import start_scheduler
from colorama import Fore, Style

load_dotenv()

VERSION = os.getenv("VERSION", "0.5")
ADB_QR_DEVICE_PAIRING = os.getenv("ADB_QR_DEVICE_PAIRING", "true").lower() == "true"
ADB_AUTO_CONNECT = os.getenv("ADB_AUTO_CONNECT", "false").lower() == "true"
ADB_DEFAULT_DEVICE = os.getenv("ADB_DEFAULT_DEVICE")
PLAN_RESET_DAY_OF_MONTH = int(os.getenv("PLAN_RESET_DAY_OF_MONTH", "0"))
DATABASE_PATH = os.getenv("DATABASE_PATH", "data/Android-SMS-API.db")
STATIC_DIR = Path(__file__).resolve().parent / "static"

db_helper = SQLiteDb(database_path=DATABASE_PATH)
database = db_helper.connect()

log = create_logger(alias="APP", logger_name="ASA_APP")

@asynccontextmanager
async def lifespan(app: FastAPI):

    log.info("Application startup: Initializing background services.")

    connection_failed = False

    if ADB_AUTO_CONNECT and ADB_DEFAULT_DEVICE:

        log.info(f"ADB Auto-Connect enabled. Attempting to connect to default device: {ADB_DEFAULT_DEVICE}")
        try:
            await adb_library.connect_device(ADB_DEFAULT_DEVICE)
        except Exception as e:
            log.error(f"Auto-connect failed: {str(e)}")
            connection_failed = True

    if (connection_failed and ADB_QR_DEVICE_PAIRING) or (not ADB_AUTO_CONNECT and ADB_QR_DEVICE_PAIRING):

        try:
            existing_devices = await adb_library.get_devices()

            if existing_devices:
                log.info(f"Skipping QR pairing — {len(existing_devices)} device(s) already connected to ADB.")

            else:
                log.debug("Starting terminal-based QR pairing session as per configuration.")
                start_terminal_pairing_session(300)

        except Exception:
            log.debug("Starting terminal-based QR pairing session as per configuration.")
            start_terminal_pairing_session(300)

    start_scheduler()

    current_password = ADMIN_PASSWORD
    is_default = current_password == DEFAULT_PASSWORD
    host = "localhost"
    port = 8000

    if is_default:

        log.warning(f"""\n
{Fore.YELLOW}{Style.BRIGHT}{'=' * 60}
  ⚠  DEFAULT PASSWORD DETECTED — MUST BE CHANGED
{'=' * 60}{Style.RESET_ALL}
  {Fore.CYAN}Username:{Style.RESET_ALL}  {Style.BRIGHT}{ADMIN_USERNAME}{Style.RESET_ALL}
  {Fore.CYAN}Password:{Style.RESET_ALL}  {Style.BRIGHT}{current_password}{Style.RESET_ALL}

  {Fore.YELLOW}The default password must be changed before login.{Style.RESET_ALL}
  {Fore.YELLOW}Set a new password via the web UI or update:{Style.RESET_ALL}
  {Fore.WHITE}{Style.BRIGHT}  .env → ADMIN_PASSWORD=YourNewPassword{Style.RESET_ALL}

  {Fore.YELLOW}Password requirements:{Style.RESET_ALL}
  {Fore.WHITE}  • At least 8 characters{Style.RESET_ALL}
  {Fore.WHITE}  • At least one letter (a-z, A-Z){Style.RESET_ALL}
  {Fore.WHITE}  • At least one digit (0-9){Style.RESET_ALL}
  {Fore.WHITE}  • At least one special character (!@#$%^&* etc.){Style.RESET_ALL}

  {Fore.CYAN}Dashboard:{Style.RESET_ALL}  {Style.BRIGHT}http://{host}:{port}{Style.RESET_ALL}
  {Fore.CYAN}API Docs:{Style.RESET_ALL}   {Style.BRIGHT}http://{host}:{port}/docs{Style.RESET_ALL}
{Fore.YELLOW}{Style.BRIGHT}{'=' * 60}{Style.RESET_ALL}
""")
    else:
        log.info(f"""\n
{Fore.GREEN}{Style.BRIGHT}{'─' * 50}
  ✓  Admin Credentials
{'─' * 50}{Style.RESET_ALL}
  {Fore.CYAN}Username:{Style.RESET_ALL}  {Style.BRIGHT}{ADMIN_USERNAME}{Style.RESET_ALL}
  {Fore.CYAN}Password:{Style.RESET_ALL}  {Style.BRIGHT}{current_password}{Style.RESET_ALL}

  {Fore.CYAN}Dashboard:{Style.RESET_ALL}  {Style.BRIGHT}http://{host}:{port}{Style.RESET_ALL}
  {Fore.CYAN}API Docs:{Style.RESET_ALL}   {Style.BRIGHT}http://{host}:{port}/docs{Style.RESET_ALL}
{Fore.GREEN}{Style.BRIGHT}{'─' * 50}{Style.RESET_ALL}
""")

    yield


app = FastAPI(
    title="Android-SMS-API",
    version=VERSION,
    description="Turn your Android phone into a programmable SMS server. A lightweight HTTP API wrapper around ADB for sending text messages over cellular network",
    lifespan=lifespan,
    responses={
        400: {
            "model": ErrorResponse,
            "description": "Bad Request - Invalid input or validation error",
            "content": {
                "application/json": {
                    "example": {
                        "detail": "Username already registered",
                        "status_code": 400,
                        "error_type": "BadRequest"
                    }
                }
            }
        },
        401: {
            "model": ErrorResponse,
            "description": "Unauthorized - Invalid or missing authentication credentials",
            "content": {
                "application/json": {
                    "example": {
                        "detail": "Could not validate credentials",
                        "status_code": 401,
                        "error_type": "Unauthorized"
                    }
                }
            }
        },
        403: {
            "model": ErrorResponse,
            "description": "Forbidden - Insufficient permissions or operation not allowed",
            "content": {
                "application/json": {
                    "example": {
                        "detail": "You are not authorized to perform this action!",
                        "status_code": 403,
                        "error_type": "Forbidden"
                    }
                }
            }
        },
        404: {
            "model": ErrorResponse,
            "description": "Not Found - Resource does not exist",
            "content": {
                "application/json": {
                    "example": {
                        "detail": "Account not found",
                        "status_code": 404,
                        "error_type": "NotFound"
                    }
                }
            }
        },
        500: {
            "model": ErrorResponse,
            "description": "Internal Server Error - Server-side error occurred",
            "content": {
                "application/json": {
                    "example": {
                        "detail": "Internal server error occurred while processing your request",
                        "status_code": 500,
                        "error_type": "InternalServerError"
                    }
                }
            }
        }
    }
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Global handler for HTTP exceptions"""

    log.warning(f"HTTP Exception encountered. Status: {exc.status_code}, Detail: {exc.detail}, Path: {request.url.path}")

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "detail": exc.detail,
            "status_code": exc.status_code,
            "error_type": _get_error_type(exc.status_code)
        }
    )


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Global handler for validation errors"""

    log.warning(f"Request validation failed. Path: {request.url.path}, Errors: {str(exc.errors())}")

    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "detail": "Validation error: " + str(exc.errors()),
            "status_code": 400,
            "error_type": "ValidationError"
        }
    )


def _get_error_type(status_code: int) -> str:
    """Map HTTP status codes to error types"""
    error_types = {
        400: "BadRequest",
        401: "Unauthorized",
        403: "Forbidden",
        404: "NotFound",
        500: "InternalServerError"
    }
    return error_types.get(status_code, "Error")


app.include_router(
    router=health.router,
    tags=["Health"]
)

app.include_router(
    router=authentication.router,
    prefix="/auth"
)

app.include_router(
    router=tokens.router
)

app.include_router(
    router=adb.router,
    prefix="/adb"
)

if STATIC_DIR.is_dir():

    log.info(f"Serving frontend static files from: {STATIC_DIR}")

    assets_dir = STATIC_DIR / "assets"

    if assets_dir.is_dir():
        app.mount("/assets", StaticFiles(directory=str(assets_dir)), name="assets")

    @app.get("/{full_path:path}", include_in_schema=False)
    async def serve_frontend(request: Request, full_path: str):

        file_path = STATIC_DIR / full_path
        if full_path and file_path.is_file():
            return FileResponse(str(file_path))
        return FileResponse(str(STATIC_DIR / "index.html"))

else:

    log.warning(f"Static directory not found at {STATIC_DIR}. Frontend will not be served.")

if __name__ == "__main__":

    log.info("Starting Uvicorn server environment...")
    uvicorn.run(app, host="0.0.0.0", port=8001, log_config=None)