Spaces:
Sleeping
Sleeping
File size: 2,799 Bytes
9bc2904 1124fbf 3ef63de 9bc2904 1124fbf b12a44c 9bc2904 0e05d21 eaeabeb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
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 """
<html>
<head><title>Chat API</title></head>
<body>
<h1>Hello HF!</h1>
<p>See the <a href="/docs">Swagger docs</a>.</p>
</body>
</html>
"""
@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/[email protected]/swagger-ui-bundle.js",
swagger_css_url="https://unpkg.com/[email protected]/swagger-ui.css",
)
|