74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
import uvicorn
|
|
import logging
|
|
import requests
|
|
from fastapi import FastAPI
|
|
from fastapi import Response
|
|
from fastapi import WebSocket
|
|
from fastapi import WebSocketDisconnect
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
from src.websocket_manager import ThreadedConnectionManager
|
|
|
|
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)d\r\n\t%(message)s', datefmt='%d-%b-%y %H:%M:%S', level=logging.INFO)
|
|
|
|
PORT = 50001
|
|
|
|
# * * * * * * * * * * * *
|
|
# INIT FASTAPI CLASSES
|
|
# * * * * * * * * * * * *
|
|
app = FastAPI()
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
# * * * * * * * * * * * *
|
|
# INIT CUSTOM CLASSES
|
|
# * * * * * * * * * * * *
|
|
manager = ThreadedConnectionManager()
|
|
|
|
# * * * * * * * * * * * *
|
|
# WEBSOCKET ROUTES
|
|
# * * * * * * * * * * * *
|
|
@app.websocket("/ws/{client_id}")
|
|
async def websocket_endpoint(websocket: WebSocket, client_id: int):
|
|
await manager.connect(websocket)
|
|
try:
|
|
while True:
|
|
text = await websocket.receive_text()
|
|
print(text)
|
|
# await manager.send_personal_message(f"You wrote: {data}", websocket)
|
|
await manager.broadcast({"data": text})
|
|
except WebSocketDisconnect:
|
|
manager.disconnect(websocket)
|
|
await manager.broadcast({"data": f"Client #{client_id} left the chat"})
|
|
|
|
# * * * * * * * * * * * *
|
|
# HTTP ROUTES
|
|
# * * * * * * * * * * * *
|
|
@app.get("/")
|
|
async def index():
|
|
filename = f"html/index.html"
|
|
return FileResponse(filename)
|
|
|
|
@app.get("/alive")
|
|
async def get_alive():
|
|
return {}
|
|
|
|
class AdviceBody(BaseModel):
|
|
topic: str
|
|
|
|
@app.post("/advice")
|
|
async def post_advice(body: AdviceBody):
|
|
topic = body.topic
|
|
result = requests.get("https://api.adviceslip.com/advice/search/"+topic, timeout=20)
|
|
data = result.json()
|
|
if "slips" in data:
|
|
return data
|
|
return Response(content=data["message"]["text"], status_code=404, media_type="text/html", headers={})
|
|
|
|
def main():
|
|
uvicorn.run(app, host="0.0.0.0", port=PORT)
|
|
manager.running = False
|
|
|
|
if __name__ == "__main__":
|
|
main() |