162 lines
9.6 KiB
Python
162 lines
9.6 KiB
Python
import asyncio
|
|
import base64
|
|
import hmac
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
from html import escape
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, Header, HTTPException, Query, Request, status
|
|
from fastapi.responses import HTMLResponse
|
|
from pydantic import BaseModel, Field, HttpUrl
|
|
|
|
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger("history-collector")
|
|
|
|
DATA_DIR = Path(os.getenv("DATA_DIR", "/data"))
|
|
DB_PATH = DATA_DIR / "history.sqlite3"
|
|
API_TOKEN = os.getenv("API_TOKEN", "")
|
|
VIEWER_USERNAME = os.getenv("VIEWER_USERNAME", "admin")
|
|
VIEWER_PASSWORD = os.getenv("VIEWER_PASSWORD", "")
|
|
DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "")
|
|
MAX_VISITS_PER_REQUEST = int(os.getenv("MAX_VISITS_PER_REQUEST", "100"))
|
|
|
|
class Visit(BaseModel):
|
|
url: HttpUrl
|
|
title: str = Field(default="", max_length=1000)
|
|
visit_time: datetime
|
|
transition: str = Field(default="observed", max_length=100)
|
|
|
|
class VisitBatch(BaseModel):
|
|
device_id: str = Field(min_length=1, max_length=128, pattern=r"^[a-zA-Z0-9._-]+$")
|
|
visits: list[Visit] = Field(min_length=1, max_length=MAX_VISITS_PER_REQUEST)
|
|
|
|
app = FastAPI(title="Local History Collector", docs_url=None, redoc_url=None)
|
|
|
|
def domain_from_url(url: str) -> str:
|
|
return (urlparse(url).hostname or "").lower()
|
|
|
|
def connect():
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("""CREATE TABLE IF NOT EXISTS visits (
|
|
id INTEGER PRIMARY KEY, device_id TEXT NOT NULL, url TEXT NOT NULL,
|
|
domain TEXT NOT NULL DEFAULT '', title TEXT NOT NULL DEFAULT '',
|
|
visit_time TEXT NOT NULL, transition TEXT NOT NULL, received_at TEXT NOT NULL,
|
|
UNIQUE(device_id, url, visit_time))""")
|
|
# Migration for databases created by collector v1.
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(visits)")}
|
|
if "domain" not in columns:
|
|
conn.execute("ALTER TABLE visits ADD COLUMN domain TEXT NOT NULL DEFAULT ''")
|
|
missing = conn.execute("SELECT id, url FROM visits WHERE domain = ''").fetchall()
|
|
conn.executemany("UPDATE visits SET domain = ? WHERE id = ?", [(domain_from_url(row["url"]), row["id"]) for row in missing])
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_visits_device_time ON visits(device_id, visit_time DESC)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_visits_domain ON visits(domain)")
|
|
conn.commit()
|
|
return conn
|
|
|
|
@app.on_event("startup")
|
|
def startup():
|
|
if len(API_TOKEN) < 24:
|
|
raise RuntimeError("API_TOKEN must contain at least 24 characters")
|
|
if len(VIEWER_PASSWORD) < 12:
|
|
raise RuntimeError("VIEWER_PASSWORD must contain at least 12 characters")
|
|
with closing(connect()):
|
|
pass
|
|
|
|
def require_token(authorization: str | None):
|
|
if not hmac.compare_digest(authorization or "", f"Bearer {API_TOKEN}"):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API token")
|
|
|
|
def require_viewer(request: Request):
|
|
header = request.headers.get("authorization", "")
|
|
try:
|
|
scheme, encoded = header.split(" ", 1)
|
|
username, password = base64.b64decode(encoded).decode("utf-8").split(":", 1)
|
|
except Exception:
|
|
scheme = username = password = ""
|
|
if scheme.lower() != "basic" or not (hmac.compare_digest(username, VIEWER_USERNAME) and hmac.compare_digest(password, VIEWER_PASSWORD)):
|
|
raise HTTPException(status_code=401, detail="Authentication required", headers={"WWW-Authenticate": 'Basic realm="History Collector"'})
|
|
|
|
def save_batch(batch: VisitBatch) -> int:
|
|
received_at = datetime.now(timezone.utc).isoformat()
|
|
rows = [(batch.device_id, str(v.url), domain_from_url(str(v.url)), v.title,
|
|
v.visit_time.astimezone(timezone.utc).isoformat(), v.transition, received_at) for v in batch.visits]
|
|
with closing(connect()) as conn:
|
|
before = conn.total_changes
|
|
conn.executemany("""INSERT OR IGNORE INTO visits
|
|
(device_id, url, domain, title, visit_time, transition, received_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""", rows)
|
|
inserted = conn.total_changes - before
|
|
conn.commit()
|
|
return inserted
|
|
|
|
async def post_discord(batch: VisitBatch):
|
|
if not DISCORD_WEBHOOK_URL:
|
|
return
|
|
lines = [f"• {v.visit_time.astimezone(timezone.utc):%Y-%m-%d %H:%M UTC} — {v.title or '(brez naslova)'}\n <{v.url}>" for v in batch.visits]
|
|
chunks, current = [], f"**Zgodovina brskanja · {batch.device_id}**\n"
|
|
for line in lines:
|
|
candidate = current + line + "\n"
|
|
if len(candidate) > 1900 and current.strip():
|
|
chunks.append(current)
|
|
current = f"**Zgodovina brskanja · {batch.device_id} (nadaljevanje)**\n{line}\n"
|
|
else:
|
|
current = candidate
|
|
if current.strip(): chunks.append(current)
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
for content in chunks:
|
|
response = await client.post(DISCORD_WEBHOOK_URL, json={"content": content})
|
|
response.raise_for_status()
|
|
except Exception:
|
|
logger.exception("Discord relay failed")
|
|
|
|
@app.get("/healthz")
|
|
def healthz():
|
|
return {"ok": True}
|
|
|
|
@app.post("/v1/visits", status_code=status.HTTP_202_ACCEPTED)
|
|
async def receive_visits(batch: VisitBatch, authorization: Annotated[str | None, Header()] = None):
|
|
require_token(authorization)
|
|
stored = save_batch(batch)
|
|
asyncio.create_task(post_discord(batch))
|
|
return {"accepted": len(batch.visits), "stored": stored}
|
|
|
|
def iso_from_local(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value).replace(tzinfo=timezone.utc).isoformat()
|
|
except ValueError:
|
|
return None
|
|
|
|
def viewer_page(rows, domains, devices, start, end, device, search):
|
|
opts = ['<option value="">Vse naprave</option>'] + [f'<option value="{escape(d)}" {"selected" if d == device else ""}>{escape(d)}</option>' for d in devices]
|
|
visit_rows = "".join(f"<tr><td>{escape(r['visit_time'].replace('+00:00', 'Z'))}</td><td>{escape(r['device_id'])}</td><td>{escape(r['domain'])}</td><td>{escape(r['title'])}</td><td><a href=\"{escape(r['url'], quote=True)}\" target=\"_blank\" rel=\"noreferrer\">{escape(r['url'])}</a></td></tr>" for r in rows) or '<tr><td colspan="5">Ni zadetkov.</td></tr>'
|
|
domain_rows = "".join(f"<tr><td>{escape(r['domain'])}</td><td>{r['visits']}</td><td>{escape(r['last_seen'].replace('+00:00', 'Z'))}</td></tr>" for r in domains) or '<tr><td colspan="3">Ni zadetkov.</td></tr>'
|
|
return f'''<!doctype html><html lang="sl"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Zgodovina brskanja</title><style>body{{font:14px system-ui,sans-serif;margin:2rem;background:#111827;color:#e5e7eb}}main{{max-width:1400px;margin:auto}}form,.box{{background:#1f2937;padding:1rem;border-radius:10px;margin:1rem 0}}input,select,button{{padding:.5rem;margin:.25rem;border-radius:5px;border:1px solid #4b5563}}input,select{{background:#111827;color:#e5e7eb}}button{{background:#2563eb;color:white}}table{{border-collapse:collapse;width:100%;background:#fff;color:#111827}}th,td{{padding:.55rem;border:1px solid #d1d5db;text-align:left;vertical-align:top}}th{{background:#e5e7eb}}a{{color:#075985;word-break:break-all}}.grid{{display:grid;grid-template-columns:2fr 1fr;gap:1rem}}@media(max-width:900px){{.grid{{grid-template-columns:1fr}}}}</style><main><h1>Zgodovina brskanja</h1><form method="get"><label>Od <input type="datetime-local" name="start" value="{escape(start or '')}"></label><label>Do <input type="datetime-local" name="end" value="{escape(end or '')}"></label><label>Naprava <select name="device">{''.join(opts)}</select></label><label>Išči <input name="q" value="{escape(search or '')}" placeholder="domena, naslov ali URL"></label><button>Filtriraj</button></form><div class="grid"><section class="box"><h2>Obiski (zadnjih 500)</h2><table><thead><tr><th>Čas (UTC)</th><th>Naprava</th><th>Domena</th><th>Naslov</th><th>URL</th></tr></thead><tbody>{visit_rows}</tbody></table></section><section class="box"><h2>Unikatne domene</h2><table><thead><tr><th>Domena</th><th>Obiski</th><th>Zadnji obisk</th></tr></thead><tbody>{domain_rows}</tbody></table></section></div></main></html>'''
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def viewer(request: Request, start: str | None = Query(None), end: str | None = Query(None), device: str | None = Query(None), q: str | None = Query(None)):
|
|
require_viewer(request)
|
|
filters, params = [], []
|
|
if value := iso_from_local(start): filters.append("visit_time >= ?"); params.append(value)
|
|
if value := iso_from_local(end): filters.append("visit_time <= ?"); params.append(value)
|
|
if device: filters.append("device_id = ?"); params.append(device)
|
|
if q: filters.append("(domain LIKE ? OR title LIKE ? OR url LIKE ?)"); params.extend([f"%{q}%"] * 3)
|
|
where = (" WHERE " + " AND ".join(filters)) if filters else ""
|
|
with closing(connect()) as conn:
|
|
rows = conn.execute(f"SELECT visit_time, device_id, domain, title, url FROM visits{where} ORDER BY visit_time DESC LIMIT 500", params).fetchall()
|
|
domains = conn.execute(f"SELECT domain, COUNT(*) visits, MAX(visit_time) last_seen FROM visits{where} GROUP BY domain ORDER BY visits DESC, last_seen DESC LIMIT 200", params).fetchall()
|
|
devices = [r[0] for r in conn.execute("SELECT DISTINCT device_id FROM visits ORDER BY device_id")]
|
|
return viewer_page(rows, domains, devices, start, end, device, q)
|