Spaces:
Running
Running
| from fastapi import FastAPI, Body, Header, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from utils import beijing, parse_token | |
| from logging_helper import LoggingHelper | |
| from fastapi.templating import Jinja2Templates | |
| app = FastAPI( | |
| title=f"Log Displayer", | |
| description=f"Updated at {beijing()}", | |
| ) | |
| print("Adding middlewares...", end="") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| print("Done\n") | |
| LOGS_REPO_ID = "pgsoft/logs" | |
| local_dir = "data/logs" | |
| logger = LoggingHelper( | |
| repo_id=LOGS_REPO_ID, | |
| local_dir=local_dir, | |
| ) | |
| async def add_log( | |
| end: str, | |
| message: str = Body(..., embed=True), | |
| token: str | None = Header(None), | |
| source: str | None = Header("web"), | |
| ): | |
| print("Type: ", end) | |
| print("From: ", source) | |
| print("Token: ", token) | |
| uid, username = parse_token(token) | |
| timestamp = beijing().isoformat() | |
| print("Timestamp: ", timestamp) | |
| print("Message: ", message) | |
| log = { | |
| "type": end, | |
| "source": source, | |
| "uid": uid, | |
| "username": username, | |
| "token": token, | |
| "content": message, | |
| "timestamp": timestamp, | |
| } | |
| logger.addlog(log) | |
| return True | |
| async def health_check(): | |
| return True | |
| templates = Jinja2Templates(directory="static") | |
| async def root(request: Request): | |
| data = logger.refresh() | |
| return templates.TemplateResponse( | |
| "index.html", | |
| {"request": request, "data": data}, | |
| ) | |