import os
from datetime import datetime, timedelta
from fastapi import FastAPI, Request, HTTPException, Query
from fastapi.responses import HTMLResponse, Response, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from .database import init_db, get_db, get_domain_id
from .visitors import process_visitor
app = FastAPI(title="Simple Analytics")
app.mount("/static", StaticFiles(directory="/app/static"), name="static")
templates = Jinja2Templates(directory="/app/templates")
SECRET_KEY = os.environ.get("SECRET_KEY", "change-me-in-production")
@app.on_event("startup")
def startup():
init_db()
# ─── Tracking Endpoint ───────────────────────────────────────────────
@app.get("/c")
async def collect(request: Request):
"""Image pixel endpoint — called by the tracking script."""
client_ip = request.client.host if request.client else "unknown"
url = request.query_params.get("u", "/")
referrer = request.query_params.get("r", "")
headers = dict(request.headers)
if not url or url == "/":
headers["Referer"] = referrer
visitor = process_visitor(client_ip, headers, url)
domain = extract_domain(visitor["referrer"] or visitor["page"])
if not domain:
return Response(GIF_1X1, media_type="image/gif")
with get_db() as conn:
domain_id = get_domain_id(conn, domain)
conn.execute(
"INSERT INTO visits "
"(domain_id, page, referrer, country, city, browser, os, device, is_bot, ip_hash) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(domain_id, visitor["page"], visitor["referrer"],
visitor["country"], visitor["city"], visitor["browser"],
visitor["os"], visitor["device"], visitor["is_bot"],
visitor["ip_hash"]),
)
return Response(GIF_1X1, media_type="image/gif")
def extract_domain(url: str) -> str:
"""Extract domain from a URL."""
if not url or url.startswith("/"):
return None
try:
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.hostname
except Exception:
return None
# 1x1 transparent GIF
GIF_1X1 = bytes([
0x47, 0x49, 0x46, 0x38, 0x39, 0x61, 0x01, 0x00, 0x01, 0x00,
0x80, 0x00, 0x00, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x21,
0xf9, 0x04, 0x01, 0x00, 0x00, 0x00, 0x00, 0x2c, 0x00, 0x00,
0x00, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0x02, 0x02, 0x44,
0x01, 0x00, 0x3b
])
# ─── Beacon Script ───────────────────────────────────────────────────
BEACON_JS = r"""
(function(){
var d=document,l=location;
if(typeof d.referrer==='undefined')d.referrer='';
var p=encodeURIComponent(l.href);
var r=encodeURIComponent(d.referrer);
var n=Math.round((new Date()).getTime()/3600000);
var img=new Image(1,1);
img.src=location.protocol+'//'+location.hostname+'/c?u='+p+'&r='+r+'&n='+n;
img.onload=function(){this.remove()};
})();
"""
@app.get("/beacon.js")
async def beacon_script():
return Response(BEACON_JS, media_type="application/javascript")
# ─── Admin Dashboard ─────────────────────────────────────────────────
@app.get("/")
async def index(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/login")
async def login_page(request: Request):
return templates.TemplateResponse("login.html", {"request": request})
@app.post("/login")
async def handle_login(request: Request):
form = await request.form()
password = form.get("password", "")
if password == SECRET_KEY:
response = HTMLResponse("""<script>
document.cookie="auth=true; path=/; max-age=86400; SameSite=Lax";
window.location.href='/dashboard';
</script>""")
return response
return HTMLResponse(
'<script>alert("Wrong password");window.history.back();</script>',
status_code=401,
)
def check_auth(request: Request):
"""Simple cookie-based auth check."""
token = request.cookies.get("auth")
if token != "true":
raise HTTPException(status_code=302, headers={"Location": "/login"})
@app.get("/dashboard")
async def dashboard(request: Request):
check_auth(request)
return templates.TemplateResponse("dashboard.html", {"request": request})
# ─── API Endpoints ───────────────────────────────────────────────────
@app.get("/api/domains")
async def list_domains(request: Request):
check_auth(request)
with get_db() as conn:
rows = conn.execute(
"SELECT d.domain, d.created_at, "
"COUNT(v.id) as total_visits, "
"COUNT(DISTINCT DATE(v.timestamp)) as active_days "
"FROM domains d "
"LEFT JOIN visits v ON d.id = v.domain_id AND v.is_bot = 0 "
"GROUP BY d.id "
"ORDER BY total_visits DESC"
).fetchall()
return [dict(row) for row in rows]
@app.get("/api/stats")
async def get_stats(
request: Request,
domain: str = Query(None),
days: int = Query(7, ge=1, le=365),
):
check_auth(request)
with get_db() as conn:
if domain:
domain_row = conn.execute(
"SELECT id FROM domains WHERE domain = ?", (domain,)
).fetchone()
if not domain_row:
raise HTTPException(status_code=404, detail="Domain not found")
domain_id = domain_row["id"]
where = "WHERE v.domain_id = ? AND v.is_bot = 0"
params = (domain_id,)
else:
where = "WHERE v.is_bot = 0"
params = ()
since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
# Total visits
total = conn.execute(
f"SELECT COUNT(*) as cnt FROM visits v {where} "
f"AND v.timestamp >= ? AND v.ip_hash IS NOT NULL",
params + (since,),
).fetchone()["cnt"]
# Unique visitors (by ip_hash)
unique = conn.execute(
f"SELECT COUNT(DISTINCT ip_hash) as cnt FROM visits v {where} "
f"AND v.timestamp >= ?",
params + (since,),
).fetchone()["cnt"]
# Daily visits for chart
daily = conn.execute(
f"SELECT DATE(timestamp) as day, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? "
f"GROUP BY day ORDER BY day",
params + (since,),
).fetchall()
# Top pages
pages = conn.execute(
f"SELECT page, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? "
f"GROUP BY page ORDER BY cnt DESC LIMIT 10",
params + (since,),
).fetchall()
# Top referrers
referrers = conn.execute(
f"SELECT referrer, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? AND referrer != '' "
f"GROUP BY referrer ORDER BY cnt DESC LIMIT 10",
params + (since,),
).fetchall()
# Countries
countries = conn.execute(
f"SELECT country, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? AND country != '' AND country != 'Unknown' "
f"GROUP BY country ORDER BY cnt DESC LIMIT 10",
params + (since,),
).fetchall()
# Browsers
browsers = conn.execute(
f"SELECT browser, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? "
f"GROUP BY browser ORDER BY cnt DESC LIMIT 10",
params + (since,),
).fetchall()
# Devices
devices = conn.execute(
f"SELECT device, COUNT(*) as cnt "
f"FROM visits v {where} "
f"AND v.timestamp >= ? "
f"GROUP BY device ORDER BY cnt DESC",
params + (since,),
).fetchall()
return {
"total_visits": total,
"unique_visitors": unique,
"daily": [dict(r) for r in daily],
"pages": [dict(r) for r in pages],
"referrers": [dict(r) for r in referrers],
"countries": [dict(r) for r in countries],
"browsers": [dict(r) for r in browsers],
"devices": [dict(r) for r in devices],
}