HackInPovo/geolog
9
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Delete backend.py.bacca

Browse source
This commit is contained in:
lichene 2025-08-30 21:53:37 +00:00
parent b2a5f85e4e
commit b204e32a49
1 changed files with 0 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

88
backend.py.bacca
View file

@ -1,88 +0,0 @@
# backend.py
from fastapi import FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional
import pandas as pd
import geoip2.database
from datetime import datetime
# ----------------------------
# 1. Load Access Logs
# ----------------------------
logs = []
with open("access.log") as f:
for line in f:
# Example log format: "2025-08-28T12:34:56Z 192.0.2.1 GET /api/service1"
parts = line.strip().split(" ", 3)
if len(parts) != 4:
continue # skip malformed lines
timestamp, ip, method, path = parts
logs.append({
"timestamp": timestamp,
"ip": ip,
"method": method,
"path": path
})
df = pd.DataFrame(logs)
df["timestamp"] = pd.to_datetime(df["timestamp"])
# ----------------------------
# 2. GeoIP Lookup
# ----------------------------
reader = geoip2.database.Reader("GeoLite2-City.mmdb")
def ip_to_geo(ip):
try:
response = reader.city(ip)
return response.location.latitude, response.location.longitude
except Exception as e:
print(f"GeoIP lookup failed for {ip}: {e}")
return None, None
df["lat"], df["lon"] = zip(*df["ip"].apply(ip_to_geo))
print(df)
df = df.dropna(subset=["lat", "lon"])
# ----------------------------
# 3. FastAPI Setup
# ----------------------------
app = FastAPI(title="Reverse Proxy Connections Map API")
# Allow frontend to query API from any origin
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"]
)
# ----------------------------
# 4. API Endpoint
# ----------------------------
@app.get("/connections")
def get_connections(
service: Optional[str] = Query(None, description="Filter by service path"),
start: Optional[str] = Query(None, description="Start datetime in ISO format"),
end: Optional[str] = Query(None, description="End datetime in ISO format")
):
data = df.copy()
if service:
data = data[data["path"].str.contains(service)]
if start:
data = data[data["timestamp"] >= pd.to_datetime(start)]
if end:
data = data[data["timestamp"] <= pd.to_datetime(end)]
return data[["timestamp", "path", "lat", "lon"]].to_dict(orient="records")
# ----------------------------
# 5. Healthcheck Endpoint
# ----------------------------
@app.get("/health")
def health():
return {"status": "ok", "total_connections": len(df)}
if __name__ == "__main__":
import uvicorn
uvicorn.run("backend:app", host="0.0.0.0", port=8000, reload=True)