res_levis_API/app.py

import json
import subprocess
import logging
import ipaddress
#import api_utils.carddav_util as carddav_util
#from .api_utils import carddav_util
from enum import Enum
from typing import Any, Dict, List, Optional
# import wave
import os
import shutil
import re
# import enviroment variables
import config_env
#other
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Callable
import base64
from datetime import datetime, timedelta
import csv

from api_utils import api_utils, coerce_methods
from collections import OrderedDict

from pydantic import BaseModel, Field

from fastapi import Depends, FastAPI, HTTPException, File, UploadFile, Query
from fastapi.encoders import jsonable_encoder
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi.openapi.utils import get_openapi

from starlette.status import HTTP_403_FORBIDDEN
from starlette.responses import RedirectResponse, Response, JSONResponse
from starlette.requests import Request
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import FileResponse
from starlette.types import ASGIApp, Message, Receive, Scope, Send

from models.cellular_hardware import cellularHardware
from models.cellular_hardwares import cellularHardwares
from models.call import call, post_call
from models.calls import calls
from models.httpresponse import httpResponse400, httpResponse200, httpResponse500
from schemas.reslevis import CalibrationMetadata

from fastapi_login import LoginManager

from core.security import manager, NotAuthenticatedException
from security import get_current_user

from routes import auth as _auth
auth_router = _auth.router
from routes import user as _user
user_router = _user.router
#from routes.posts import router as posts_router
from routes import majornet as _majornet
majornet_router = _majornet.router
from routes import presence as _presence
presence_router = _presence.router
from routes import contacts as _contacts
contacts_router = _contacts.router
from routes import reslevis as _reslevis
reslevis_router = _reslevis.router

#security
from fastapi import FastAPI, Security
from fastapi.security import OAuth2AuthorizationCodeBearer

#Proxy al CORE ResLevis
import httpx
from mqtt_gateway_monitor import MqttGatewayMonitor


AUTH_URL = config_env.KEYCLOAK_AUTH_URL
TOKEN_URL = config_env.KEYCLOAK_TOKEN_URL

oauth2 = OAuth2AuthorizationCodeBearer(
    authorizationUrl=AUTH_URL,
    tokenUrl=TOKEN_URL,
    scopes={"items:read": "Read items", "items:write": "Write items"},
)

log = logging.getLogger(__name__)  # pylint: disable=invalid-name

DEBUG = True

app = FastAPI(title="MajorNet API",  redoc_url=None, docs_url=None, openapi_url=None)
####DEV
##app = FastAPI(title=PROJECT_NAME)
app.debug = True

#logging
from audit import AuditMiddleware
app.add_middleware(AuditMiddleware)
logging.basicConfig(filename='/data/var/log/FastAPI/AuthenticatedAPI.log', filemode='w', format='%(name)s - %(levelname)s - %(message)s')
logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)s:%(name)s:%(message)s"
)

#ResLevis CORE Proxying
CORE_BASE_URL = config_env.CORE_API_URL.rstrip("/")

HOP_BY_HOP = {
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
    "te", "trailers", "transfer-encoding", "upgrade",
}

def _filter_headers(headers: dict) -> dict:
    return {k: v for k, v in headers.items() if k.lower() not in HOP_BY_HOP}

async def _forward_to_core(request: Request, body: bytes) -> Response:
    url = f"{CORE_BASE_URL}{request.url.path}"
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.request(
            request.method,
            url,
            params=request.query_params,
            content=body,
            headers=_filter_headers(dict(request.headers)),
        )
    return Response(
        content=resp.content,
        status_code=resp.status_code,
        headers=_filter_headers(dict(resp.headers)),
        media_type=resp.headers.get("content-type"),
    )

ALLOWED_HOSTS = ["*"]

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

# MQTT gateway monitor
mqtt_monitor = MqttGatewayMonitor()

@app.on_event("startup")
async def start_mqtt_monitor():
    await mqtt_monitor.start()

@app.on_event("shutdown")
async def stop_mqtt_monitor():
    await mqtt_monitor.stop()

#ResLevis CORE middleware
@app.middleware("http")
async def local_then_core(request: Request, call_next):
    # only proxy CRUD for Reslevis (change prefix or methods if needed)
    if request.url.path.startswith("/reslevis/") and request.method in {"POST", "PUT", "DELETE", "PATCH"}:
        body = await request.body()          # raw body preserved
        local_resp = await call_next(request)  # local storage runs here
        if local_resp.status_code >= 400:
            return local_resp                 # stop if local failed
        try:
            core_resp = await _forward_to_core(request, body)
        except httpx.RequestError:
            return local_resp
        if core_resp.status_code >= 400:
            return local_resp
        return core_resp
    return await call_next(request)

@app.exception_handler(NotAuthenticatedException)
def auth_exception_handler(request: Request, exc: NotAuthenticatedException):
    """
    Redirect the user to the login page if not logged in
    """
    return RedirectResponse(url='/login')

# Nasconde i dettagli degli errori 5xx
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    if exc.status_code >= 500:
        log.exception("HTTP %s: %s", exc.status_code, exc.detail)
        return JSONResponse(status_code=exc.status_code, content={"detail": "Internal Server Error"})
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})

@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    log.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

app.include_router(auth_router)
app.include_router(user_router)
#app.include_router(posts_router)
app.include_router(presence_router)
app.include_router(majornet_router)
app.include_router(contacts_router)
app.include_router(reslevis_router, prefix="/reslevis",  tags=["Reslevis"])

META_FILE_PATTERN = re.compile(r"^meta_(-?\d+)\.json$")

def _model_to_dict(model: Any) -> Dict[str, Any]:
    if hasattr(model, "model_dump"):
        return model.model_dump()
    return model.dict()

def _validate_metadata(payload: Dict[str, Any]) -> Dict[str, Any]:
    if hasattr(CalibrationMetadata, "model_validate"):
        return _model_to_dict(CalibrationMetadata.model_validate(payload))
    return _model_to_dict(CalibrationMetadata.parse_obj(payload))

def _read_ble_ai_metadata(path: str) -> Dict[str, Any]:
    with open(path) as f:
        payload = json.load(f)
    return _validate_metadata(payload)


@app.get("/")
async def root():
    #return {"url": "/docs"}
    return get_swagger_ui_html(openapi_url="/openapi.json", title="docs")


@app.get("/ble-ai/infer", tags=["BLE-AI"], dependencies=[Depends(get_current_user)])
async def get_ble_ai_infer(mac: Optional[List[str]] = Query(default=None)):
    path = config_env.BLE_AI_INFER_CSV
    if not os.path.isfile(path):
        raise HTTPException(status_code=404, detail="CSV not found")

    mac_filter = None
    if mac:
        mac_filter = set()
        for item in mac:
            if not item:
                continue
            for part in str(item).split(","):
                part = part.strip()
                if part:
                    mac_filter.add(part.lower())

    items = []
    with open(path, newline="") as f:
        reader = csv.DictReader(f, delimiter=";")
        for row in reader:
            row_mac = row.get("mac")
            if mac_filter is not None:
                if not row_mac or row_mac.lower() not in mac_filter:
                    continue
            try:
                items.append(
                    {
                        "mac": row_mac,
                        "z": int(row["z"]) if row.get("z") not in (None, "") else None,
                        "x": int(row["x"]) if row.get("x") not in (None, "") else None,
                        "y": int(row["y"]) if row.get("y") not in (None, "") else None,
                    }
                )
            except (KeyError, ValueError):
                continue

    return {"items": items, "count": len(items)}


@app.get("/ble-ai/metadata", tags=["BLE-AI"], dependencies=[Depends(get_current_user)])
async def get_ble_ai_metadata(floor: Optional[int] = Query(default=None)):
    meta_dir = config_env.BLE_AI_META_DIR
    if not os.path.isdir(meta_dir):
        raise HTTPException(status_code=404, detail="Metadata directory not found")

    if floor is not None:
        file_path = os.path.join(meta_dir, f"meta_{floor}.json")
        if not os.path.isfile(file_path):
            raise HTTPException(status_code=404, detail="Metadata file not found")
        try:
            metadata = _read_ble_ai_metadata(file_path)
        except Exception:
            log.exception("Invalid metadata file: %s", file_path)
            raise HTTPException(status_code=500, detail="Invalid metadata file")
        return {"floor": floor, "metadata": metadata}

    items = []
    for file_name in os.listdir(meta_dir):
        match = META_FILE_PATTERN.match(file_name)
        if not match:
            continue

        file_floor = int(match.group(1))
        file_path = os.path.join(meta_dir, file_name)
        if not os.path.isfile(file_path):
            continue

        try:
            metadata = _read_ble_ai_metadata(file_path)
        except Exception:
            log.exception("Invalid metadata file: %s", file_path)
            raise HTTPException(status_code=500, detail="Invalid metadata file")
        items.append({"floor": file_floor, "metadata": metadata})

    items.sort(key=lambda item: item["floor"])
    return {"items": items, "count": len(items)}


@app.get("/openapi.json/", tags=["Documentation"])
async def get_open_api_endpoint():
#async def get_open_api_endpoint(current_user: User = Depends(get_current_active_user)):
    return JSONResponse(get_openapi(title="MajorNet APIs", version="1.0", routes=app.routes))


@app.get("/docs/", tags=["Documentation"])
#async def get_documentation(current_user: User = Depends(get_current_active_user)):
async def get_documentation():
    if DEBUG: print("SONO IN /DOCS")
    return get_swagger_ui_html(openapi_url="/openapi.json", title="docs")


@app.post("/majortel/call/", tags=["Majortel"])
async def route_call(active_user=Depends(manager),callerNumber=None, calledNumber=None):
    try:
        if DEBUG: print("Entro nel TRY")
        # Check if the callerNumber sent is an ip address and retrieve the associated number
        ipaddress.ip_address(callerNumber)
        callerNumberRetrieved = os.popen('asterisk -rx "sip show peers" | grep ' + callerNumber).read()
        callerNumberRetrieved = callerNumberRetrieved.split("/")
        callerNumber = callerNumberRetrieved[0]
    except:
        if DEBUG: print("EXCEPT")
    mode = "callFromTo"
    from_ = callerNumber
    to_ = calledNumber
    if DEBUG: print("DATI ARRIVATI: ")
    if DEBUG: print(callerNumber)
    if DEBUG: print(calledNumber)
    subprocess.Popen(['perl', '/usr/local/bin/ast/voice.pl', mode, from_, to_], stdout=subprocess.PIPE)
    return


@app.post("/majortel/ring/", tags=["Majortel"])
async def route_ring(active_user=Depends(manager),calledNumber=None, calledId=None, ringTime=None):
    try:
        if DEBUG: print("Entro nel TRY")
        # Check if the callerNumber sent is an ip address and retrieve the associated number
        ipaddress.ip_address(calledNumber)
        calledNumberRetrieved = os.popen('asterisk -rx "sip show peers" | grep ' + calledNumber).read()
        calledNumberRetrieved = calledNumberRetrieved.split("/")
        calledNumber = calledNumberRetrieved[0]
    except:
        if DEBUG: print("EXCEPT")
    mode = "ringAlert"
    to_ = calledNumber
    calledId_ = calledId
    ringTime_ = ringTime
    if DEBUG: print("DATI ARRIVATI: ")
    if DEBUG: print(calledNumber)
    if DEBUG: print(calledId)
    if DEBUG: print(ringTime)
    subprocess.Popen(['perl', '/usr/local/bin/ast/voice.pl', mode, to_, calledId_, ringTime_], stdout=subprocess.PIPE)
    return


@app.get("/majortel/hardware/", response_model=cellularHardwares, tags=["Majortel"])
#async def majortel_hardware_get(current_user: User = Depends(get_current_active_user)):
async def majortel_hardware_get():
    gsm_temp_list = "GSM span1: Provisioned, Up, Active"

    response = {"CellularHardwares": []}
    hardware_dict = {}
    myCmd = os.popen('asterisk -rx "gsm show spans"').read()
    myCmd = myCmd.split("\n")
    # cancello l'ultimo item della lista poichè, risultando vuoto dopo lo split,
    # genera errore "index out of range" nei successivi accessi alla lista
    if DEBUG: print("spans: ")
    if DEBUG: print(myCmd)
    myCmd = os.popen('asterisk -rx "dongle show devices"').read()
    myCmd = myCmd.split("\n")
    # cancello il primo item della lista poichè contiene la legenda
    del myCmd[0]
    del myCmd[-1]
    # costruisco la response
    for device in myCmd:
        device = device.split()
        hardware_id = device[0]
        current_device = os.popen('asterisk -rx "dongle show device state ' + hardware_id + '"').read()
        current_device = current_device.split("\n")
        # cancello il primo e gli ultimi item della lista poichè, risultando vuoti dopo lo split,
        # generano errore "index out of range" nei successivi accessi alla lista
        del current_device[0]
        del current_device[-1]
        del current_device[-1]
        # costruisco un dizionario a partire dall'output della system call
        for row in current_device:
            row = row.split(":")
            row[0] = row[0].strip()
            row[1] = row[1].strip()
            hardware_dict[row[0]] = row[1]
        hardware_id = hardware_dict["Device"]
        status = hardware_dict["State"]
        signal = hardware_dict["RSSI"]
        signal = int(signal[0:2])
        operator = hardware_dict["Provider Name"]
        device_obj = {"id": hardware_id, "description": "description", "status": status, "signal_level": signal,
                      "registered_number": "To Do", "operator": operator}
        response["CellularHardwares"].append(device_obj)

    return response


@app.get("/majortel/hardware/{item_id}", response_model=cellularHardware, tags=["Majortel"], responses={400: {"model": httpResponse400}})
#async def majortel_hardware_id_get(item_id: str, current_user: User = Depends(get_current_active_user)):
async def majortel_hardware_id_get(item_id: str,active_user=Depends(manager)):
    hardware_response = {}
    hardware_dict = {}
    current_device = os.popen('asterisk -rx "dongle show device state ' + item_id + '"').read()
    current_device = current_device.split("\n")
    # cancello il primo e gli ultimi item della lista poichè, risultando vuoti dopo lo split, 
    # generano errore "index out of range" nei successivi accessi alla lista
    del current_device[0]
    if (current_device[0]):
        del current_device[-1]
        del current_device[-1]
        # costruisco un dizionario a partire dall'output della system call
        for row in current_device:
            row = row.split(":")
            row[0] = row[0].strip()
            row[1] = row[1].strip()
            hardware_dict[row[0]] = row[1]
        hardware_id = hardware_dict["Device"]
        status = hardware_dict["State"]
        signal = hardware_dict["RSSI"]
        signal = int(signal[0:2])
        operator = hardware_dict["Provider Name"]
        hardware_response = {"id": hardware_id, "description": "description", "status": status, "signal_level": signal,
                             "registered_number": "To Do", "operator": operator}
        return hardware_response
    else:
        return JSONResponse(status_code=404, content={"message": "Device not found"})


@app.get("/majortel/calls/", response_model=calls, tags=["Majortel"])
#async def majortel_calls_get(current_user: User = Depends(get_current_active_user)):
async def majortel_calls_get(active_user=Depends(manager)):
    response = {"Calls": []}
    p = subprocess.Popen(['perl', '/opt/api_project_python/addCallQueue.pl', "get_calls"], stdout=subprocess.PIPE)
    calls = str(p.stdout.read())
    calls = calls.split("'")
    response1 = calls
    calls = calls[1].split("||")
    # cancello l'ultimo item della lista poichè, risultando vuoto dopo lo split,
    # genera errore "index out of range" nei successivi accessi alla lista
    del calls[-1]
    for call in calls:
        call = call.split("|")
        call_id = call[0]
        status = call[1]
        history = call[2]
        ntel = call[3]
        ackid = call[4]
        call_obj = {"id": call_id, "status": status, "history": history, "called_number": ntel, "ack_id": ackid}
        response["Calls"].append(call_obj)
    return response


@app.get("/majortel/calls/{call_id}/", response_model=call, tags=["Majortel"])
async def majortel_calls_id_get(call_id,active_user=Depends(manager)):
    p = subprocess.Popen(['perl', '/var/opt/FastAPI/addCallQueue.pl', "call_id", call_id],
                         stdout=subprocess.PIPE)
    call = str(p.stdout.read())
    call = call.split("|")
    call_id = call[0].split("'")[1]
    status = call[1]
    # history = call[2]
    ntel = call[3]
    ackid = call[4]
    ackid = int(ackid.split("'")[0])
    response = {"id": call_id, "status": status, "called_number": ntel, "ack_id": ackid}

    return response


# response_model=post_call,
@app.post("/majortel/calls", response_model=post_call, tags=["Majortel"])
async def majortel_calls_post(hw_id, ack_id, called_number, text_message="codice di riscontro ", timeout=30,
                              retry=1, file: UploadFile = File(None),active_user=Depends(manager)):
    # controllo se l'hw_id corrisponde ai devices connessi
    current_device = os.popen('asterisk -rx "dongle show device state ' + hw_id + '"').read()
    current_device = current_device.split("\n")
    del current_device[0]
    if (current_device[0]):
        curr_time = datetime.now()
        call_id = str(curr_time.strftime('%y')) + str(curr_time.strftime('%m')) + str(curr_time.strftime('%d')) + str(
            curr_time.strftime('%H')) + str(curr_time.strftime('%M')) + str(curr_time.strftime('%S')) + str(
            curr_time.strftime('%f'))
        file_path = ""
        if (file):
            nomefile = file.filename
            extension = nomefile.split(".")[1]

            if (extension == "wav"):
                file_path = "/data/service/voip.majornet/ivr/" + call_id
                renamed_file = file_path + ".wav"
                with open(renamed_file, "wb") as f:
                    shutil.copyfileobj(file.file, f)
                    f.close()
                # p = subprocess.Popen(['(cd /tmp/audioCalls;/usr/local/bin/sox_wav2gsm)'],stdout=subprocess.PIPE)
                # os.popen('/usr/local/bin/sox_wav2gsm 1>/dev/null 2>/dev/null')
        subprocess.Popen(
            ['perl', '/opt/api_project_python/addCallQueue.pl', "voicegateway", "digiovine@afasystems.it", call_id,
             called_number, ack_id, text_message, "retry message here", timeout, retry, "yes", "notification_to",
             "notification_bcc", "setup", file_path, "fromemailssss", hw_id], stdout=subprocess.PIPE)
        response = {"id": call_id}
        return response
    else:
        return JSONResponse(status_code=404, content={"message": "Device not found"})


@app.delete("/majortel/calls/", tags=["Majortel"],
            responses={200: {"model": httpResponse200}, 400: {"model": httpResponse400},
                       500: {"model": httpResponse500}})
#async def majortel_calls_id_delete(call_id, current_user: User = Depends(get_current_active_user)):
async def majortel_calls_id_delete(call_id,active_user=Depends(manager)):
    p = subprocess.Popen(['perl', '/opt/api_project_python/addCallQueue.pl', "delete_call", call_id],
                         stdout=subprocess.PIPE)
    response = str(p.stdout.read())
    response = response.split("'")[1]
    if (response == "Success"):
        return JSONResponse(status_code=200, content={"message": "Success"})
    elif (response == "Not found"):
        return JSONResponse(status_code=404, content={"message": "Call not found"})
    else:
        return JSONResponse(status_code=500, content={"message": "Server error"})