Spaces:
Sleeping
Sleeping
| 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 ---------- | |
| async def root(): | |
| return """ | |
| <html> | |
| <head><title>Chat API</title></head> | |
| <body> | |
| <h1>Hello HF!</h1> | |
| <p>See the <a href="/docs">Swagger docs</a>.</p> | |
| </body> | |
| </html> | |
| """ | |
| 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} | |
| async def read_messages(username: str): | |
| expire_messages() | |
| not_mine = [m for m in messages if m["from"] != username] | |
| return {"messages": not_mine} | |
| async def read_all_messages(): | |
| expire_messages() | |
| return {"messages": list(messages)} | |
| 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 ---------- | |
| 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/[email protected]/swagger-ui-bundle.js", | |
| swagger_css_url="https://unpkg.com/[email protected]/swagger-ui.css", | |
| ) | |