KIMB-technologies
cc4bb9a7e8
All checks were successful
Build and push Docker image at git tag / build (push) Successful in 55s
125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
# Agenten Plattform
|
||
#
|
||
# (c) 2024 Magnus Bender
|
||
# Institute of Humanities-Centered Artificial Intelligence (CHAI)
|
||
# Universitaet Hamburg
|
||
# https://www.chai.uni-hamburg.de/~bender
|
||
#
|
||
# source code released under the terms of GNU Public License Version 3
|
||
# https://www.gnu.org/licenses/gpl-3.0.txt
|
||
|
||
import os
|
||
|
||
from typing import List
|
||
from datetime import datetime
|
||
|
||
from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
|
||
from fastapi.responses import HTMLResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
|
||
from jinja2.runtime import Undefined as JinjaUndefined
|
||
|
||
from ums.management.interface import Interface
|
||
from ums.management.db import DB, MessageDbRow
|
||
from ums.management.process import MessageProcessor
|
||
|
||
from ums.utils import AgentMessage, AgentResponse, TEMPLATE_PATH
|
||
|
||
class WebMain():
|
||
|
||
_TIME_FORMAT = "%H:%M:%S %d.%m.%Y"
|
||
|
||
def __init__(self):
|
||
self._init_app()
|
||
self._init_templates()
|
||
|
||
self.db = DB()
|
||
self.msg_process = MessageProcessor(self.db)
|
||
|
||
self._add_routes()
|
||
self._add_routers()
|
||
|
||
|
||
def _init_app(self):
|
||
self.app = FastAPI(
|
||
title="Agenten Plattform",
|
||
description="Agenten Plattform – Management",
|
||
openapi_url="/api/schema.json",
|
||
docs_url='/api',
|
||
redoc_url=None
|
||
)
|
||
|
||
def _init_templates(self):
|
||
self.template = Jinja2Templates(
|
||
directory=TEMPLATE_PATH,
|
||
auto_reload=True
|
||
)
|
||
|
||
def timestamp2date(t:int|JinjaUndefined) -> str:
|
||
return "" if isinstance(t, JinjaUndefined) \
|
||
else datetime.fromtimestamp(t).strftime(self._TIME_FORMAT)
|
||
self.template.env.globals["timestamp2date"] = timestamp2date
|
||
|
||
def date2timestamp(d:str|JinjaUndefined) -> int|str:
|
||
return "" if isinstance(d, JinjaUndefined) \
|
||
else int(datetime.strptime(d, self._TIME_FORMAT).timestamp())
|
||
self.template.env.globals["date2timestamp"] = date2timestamp
|
||
|
||
|
||
def _add_routers(self):
|
||
interface_router = Interface(self.template, self.db)
|
||
self.app.include_router(interface_router.router)
|
||
|
||
def _add_routes(self):
|
||
|
||
@self.app.get("/index", response_class=HTMLResponse, summary="Link list")
|
||
def index(request: Request):
|
||
return self.template.TemplateResponse(
|
||
'index.html',
|
||
{"request" : request}
|
||
)
|
||
|
||
@self.app.post("/message", summary="Send a message to the management", tags=['agents'])
|
||
def message(request: Request, message:AgentMessage, background_tasks: BackgroundTasks) -> AgentResponse:
|
||
|
||
receiver = request.headers['host']
|
||
if ':' in receiver:
|
||
receiver = receiver[:receiver.rindex(':')]
|
||
|
||
sender = request.headers['x-forwarded-for']
|
||
|
||
return self.msg_process.new_message(sender, receiver, message, background_tasks)
|
||
|
||
@self.app.get("/list", summary="Get list of messages (like table)", tags=["cli, agents"])
|
||
def list(id:str|None=None, sender:str|None=None, recipient:str|None=None,
|
||
processed:bool|None=None, solution:bool|None=None,
|
||
time_after:int|None=None, time_before:int|None=None,
|
||
limit:int=10, offset:int=0
|
||
) -> List[MessageDbRow]:
|
||
|
||
db_args = {
|
||
"limit" : limit,
|
||
"offset" : offset
|
||
}
|
||
|
||
for v,n in (
|
||
(id,'id'), (sender,'sender'), (recipient,'recipient'),
|
||
(processed,'processed'), (solution,'solution'),
|
||
(time_after, 'time_after'), (time_before, 'time_before')
|
||
):
|
||
if not v is None:
|
||
db_args[n] = v
|
||
|
||
return [row for row in self.db.iterate(**db_args)]
|
||
|
||
@self.app.get("/list/single", summary="Get a single message", tags=["cli, agents"])
|
||
def status(count:int) -> MessageDbRow:
|
||
msg = self.db.by_count(count)
|
||
if msg is None:
|
||
raise HTTPException(status_code=404, detail="Message not found")
|
||
|
||
return msg
|
||
|
||
if __name__ == "ums.management.main" and os.environ.get('SERVE', 'false').lower() == 'true':
|
||
main = WebMain()
|
||
app = main.app |