from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from pydantic import BaseModel from collections import deque from datetime import datetime, timedelta from typing import Optional import uuid app = FastAPI(title="Chat Handshake API") # ---------- Configuration ---------- EXPIRATION_MINUTES = 5 MAX_MESSAGES = 1000 LOG_FILE = "activity.log" # ---------- In-memory message storage ---------- messages = deque(maxlen=MAX_MESSAGES) # ---------- Pydantic model ---------- class Message(BaseModel): type: str # "offer", "answer", "chat", "connect" from_user: str to_user: Optional[str] = None content: Optional[str] = None # ---------- Helper functions ---------- def expire_messages(): cutoff = datetime.utcnow() - timedelta(minutes=EXPIRATION_MINUTES) while messages and messages[0]["timestamp"] < cutoff: messages.popleft() def log_message(msg: dict, ip: str): with open(LOG_FILE, "a") as f: f.write(f"{datetime.utcnow()} {ip} {msg}\n") # ---------- API Endpoints ---------- @app.get("/", response_class=HTMLResponse) async def root(): return """ Chat API

Hello HF!

See the Swagger docs.

""" @app.post("/write") async def write_message(msg: Message, request: Request): expire_messages() message_id = str(uuid.uuid4()) entry = { "id": message_id, "type": msg.type, "from": msg.from_user, "to": msg.to_user, "content": msg.content, "timestamp": datetime.utcnow() } messages.append(entry) log_message(entry, request.client.host) return {"status": "ok", "id": message_id} @app.get("/read") async def read_messages(username: str): expire_messages() not_mine = [m for m in messages if m["from"] != username] return {"messages": not_mine} @app.get("/readall") async def read_all_messages(): expire_messages() return {"messages": list(messages)} @app.post("/retract") async def retract_message(message_id: str): global messages messages = deque([m for m in messages if m["id"] != message_id], maxlen=MAX_MESSAGES) return {"status": "ok"} # ---------- 4️⃣ Explicit Swagger docs endpoint ---------- @app.get("/docs", include_in_schema=False) async def custom_swagger_ui_html(): return get_swagger_ui_html( openapi_url=app.openapi_url, title=app.title + " - Swagger UI", oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url, swagger_js_url="https://unpkg.com/swagger-ui-dist@5.9.0/swagger-ui-bundle.js", swagger_css_url="https://unpkg.com/swagger-ui-dist@5.9.0/swagger-ui.css", )