Initial commit

fingerp/main.py

@@ -0,0 +1,75 @@
+from fastapi import FastAPI, Request, Depends
+from fastapi.responses import HTMLResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from pathlib import Path
+
+from .db import AsyncSessionLocal, engine, Base
+from .models import Fingerprint
+from .utils import hash_fingerprint
+
+app = FastAPI()
+
+
+@app.get("/", response_class=HTMLResponse)
+async def root():
+    static_path = Path(__file__).parent / "static" / "index.html"
+    return static_path.read_text()
+
+
+# Dependency
+async def get_db():
+    async with AsyncSessionLocal() as session:
+        yield session
+
+
+@app.on_event("startup")
+async def startup():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+@app.post("/fingerprint")
+async def collect_fingerprint(request: Request, db: AsyncSession = Depends(get_db)):
+    body = await request.json()
+
+    fp_hash = hash_fingerprint(body)
+
+    ip = request.client.host
+    user_agent = request.headers.get("user-agent", "")
+
+    record = Fingerprint(
+        fingerprint_hash=fp_hash,
+        raw_data=body,
+        ip_address=ip,
+        user_agent=user_agent,
+    )
+
+    db.add(record)
+    await db.commit()
+
+    return {
+        "status": "stored",
+        "fingerprint_hash": fp_hash,
+    }
+
+
+@app.get("/fingerprint/{fp_hash}")
+async def get_fingerprint(fp_hash: str, db: AsyncSession = Depends(get_db)):
+    result = await db.execute(
+        select(Fingerprint).where(Fingerprint.fingerprint_hash == fp_hash)
+    )
+
+    rows = result.scalars().all()
+
+    return {
+        "count": len(rows),
+        "records": [
+            {
+                "ip": r.ip_address,
+                "created_at": r.created_at,
+                "user_agent": r.user_agent,
+            }
+            for r in rows
+        ],
+    }