service_monitoring/main.py

import json
import os
import pprint
import requests
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
import logging
from helpers import flatten_dict

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d %(funcName)s %(message)s",
)
logger = logging.getLogger(__name__)

origins = [
    "http://localhost",
    "http://localhost:3010",
    "http://localhost:4173",
    "http://localhost:5173",
    "http://localhost:8000",
]

vtk = json.loads(os.getenv("VTK_KEYS"))
vtk_org = {}
BX_API_CALL = os.getenv("BX_API_CALL")
TG_TOKEN = os.getenv("TG_TOKEN")

api_app = FastAPI(title="api app")


@api_app.get("/")
def test():
    return {"app": "service monitoring api"}


@api_app.get("/vtk/{item_id}")
def get_vtk(item_id):
    if not item_id in vtk:
        raise HTTPException(status_code=404, detail="Item not found")
    try:
        if not item_id in vtk_org:
            res = requests.get(
                "https://my.vendotek.com/api/v1/org",
                headers={"Authorization": f"Bearer {vtk[item_id]}"},
            )
            data = res.json()
            org_id = data[0]["name"]
            vtk_org[item_id] = org_id

        res_unit = requests.get(
            f"https://my.vendotek.com/api/v1/org/{vtk_org[item_id]}/unit",
            headers={"Authorization": f"Bearer {vtk[item_id]}"},
        )
        data_unit = res_unit.json()
        result = [
            {
                "sn": d["sn"],
                "last_online": d["last_seen_at"],
                "name": ", ".join(
                    list(filter(None, [d["location_name"], d["address"], d["city"]]))
                ),
                "ip": list(
                    filter(
                        lambda item: item["name"] == "IP address",
                        list(
                            filter(
                                lambda item: item["id"] == "LAN",
                                list(
                                    filter(
                                        lambda item: item["id"] == "Communications",
                                        d["modules"],
                                    )
                                )[0]["modules"],
                            )
                        )[0]["details"],
                    )
                )[0]["value"],
            }
            for d in data_unit
        ]
        return {"status": "success", "data": result}
    except Exception as e:
        logger.error(e)
        raise HTTPException(status_code=500, detail=str(e))


@api_app.get("/tg/get_me")
async def get_tg_data():
    try:
        url = f"https://api.telegram.org/bot{TG_TOKEN}/getMe"
        res = requests.get(url)
        logger.info(url)
        data = res.json()
        if data["ok"] == False:
            raise HTTPException(status_code=500, detail=data["description"])
        return {"status": "success", "data": data["result"]}
    except Exception as e:
        logger.error(e)
        raise HTTPException(status_code=500, detail=str(e))


@api_app.get("/tg/get_updates")
async def get_tg_data():
    try:
        url = f"https://api.telegram.org/bot{TG_TOKEN}/getUpdates?offset=-1"
        res = requests.get(url)
        logger.info(url)
        data = res.json()
        if data["ok"] == False:
            raise HTTPException(status_code=500, detail=data["description"])
        return {"status": "success", "data": data["result"]}
    except Exception as e:
        logger.error(e)
        raise HTTPException(status_code=500, detail=str(e))
    
@api_app.get("/bx/get_users")
async def get_bx_users():
    try:
        url = f"{BX_API_CALL}user.get.json"
        res = requests.get(url)
        logger.info(url)
        data = res.json()
        return {"status": "success", "data": [v for v in data["result"] if v["UF_DEPARTMENT"][0] not in [15,16]]}
    except Exception as e:
        logger.error(e)
        raise HTTPException(status_code=500, detail=str(e))
    
@api_app.get("/bx/test_webhook/{bx_id}")
async def bx_test_webhook(bx_id):
    try:
        url = f"{BX_API_CALL}task.commentitem.add?taskId=447&fields[POST_MESSAGE]=[USER={bx_id}]Упоминание[/USER]"
        res = requests.get(url)
        logger.info(url)
        data = res.json()
        return {"status": "success", "data": data}
    except Exception as e:
        logger.error(e)
        raise HTTPException(status_code=500, detail=str(e))


app = FastAPI(title="main app")
app.exception_handler(404)


async def return_error():
    return {"error": "custom error"}


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

app.mount("/api", api_app)
app.mount("/", StaticFiles(directory="front/.output/public", html=True), name="front")