import asyncio import json import logging import os import sqlite3 from contextlib import closing from datetime import datetime, timezone from pathlib import Path from typing import Literal import httpx from fastapi import FastAPI, Header, HTTPException, Request, status 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", "") 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 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, 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) )""") conn.execute("CREATE INDEX IF NOT EXISTS idx_visits_device_time ON visits(device_id, visit_time DESC)") return conn @app.on_event("startup") def startup(): if len(API_TOKEN) < 24: raise RuntimeError("API_TOKEN must contain at least 24 characters") with closing(connect()): pass def require_token(authorization: str | None): expected = f"Bearer {API_TOKEN}" if authorization != expected: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API token") def save_batch(batch: VisitBatch) -> int: received_at = datetime.now(timezone.utc).isoformat() rows = [(batch.device_id, 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, 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] content = f"**Zgodovina brskanja · {batch.device_id}**\n" + "\n".join(lines) # Discord accepts message content up to 2,000 characters; split safely by visit. chunks, current = [], "**Zgodovina brskanja · %s**\n" % batch.device_id for line in lines: candidate = current + line + "\n" if len(candidate) > 1900 and current.strip(): chunks.append(current) current = "**Zgodovina brskanja · %s (nadaljevanje)**\n%s\n" % (batch.device_id, line) 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: # The visits are already safely persisted; don't reject them because Discord is unavailable. 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: str | None = Header(default=None)): require_token(authorization) stored = save_batch(batch) asyncio.create_task(post_discord(batch)) return {"accepted": len(batch.visits), "stored": stored}