"""Visitor log: one JSON line per page load or key action, into logs/visits.jsonl.

Two sites share this app: the main chat (openajaj.cz, `/`) and transcribe
(`/t`). Visitors arrive either through Cloudflare (the openajaj.cz domain,
which adds `CF-Connecting-IP` and `CF-IPCountry`) or straight to the hf.space
URL (the HF router puts the visitor first in `X-Forwarded-For`). Both headers
can be forged by a client — fine for counting, never for access decisions.

Human or bot: a request with the admin token is our own check ("internal");
a known bot user agent is a bot; a browser that solved the
captcha — ever, judged by its signed token of any age — is human; a browser
that never did is "unverified" (a headless scraper, or someone who gave up at
the captcha). A visitor keeps the best class it reached anywhere in the log,
so one who solved last month still counts as human today. A human who never
used a site (no chat on main, no transcription on /t) is "idle" there.

A visitor is its captcha token (logged only as a keyed hash), so a browser
update or a new IP stays the same person. Without a token it is the (ip, user
agent) pair, joined to the token that pair later shows — the hit before the
captcha belongs to whoever solved it.

Like refuse_log, this file lives in the container and is gone after a Space
restart, so the numbers cover the period since then.
"""

import json
import logging
import re
import threading
from collections import Counter, defaultdict
from collections.abc import Mapping
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from urllib.parse import urlsplit
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

import geo

LOG_FILE = Path("logs/visits.jsonl")

_TRACKED = {
    ("GET", "/"): ("main", "page"),
    ("POST", "/api/chat"): ("main", "chat"),
    ("GET", "/t"): ("t", "page"),
    ("GET", "/transcribe"): ("t", "page"),
    ("WS", "/api/stt"): ("t", "stt"),
}
_BOT_RE = re.compile(
    r"bot|spider|crawl|slurp|curl|wget|python|httpx|aiohttp|axios|go-http|"
    r"java/|okhttp|libwww|scrapy|headless|lighthouse|facebookexternalhit|"
    r"preview|monitor|uptime|node-fetch|postman",
    re.IGNORECASE,
)
_RANK = {"bot": 0, "unverified": 1, "human": 2, "internal": 3}
# First match wins, so the order matters: Android UAs say "Linux", iOS ones
# "like Mac OS X", Edge says "Chrome" and Chrome says "Safari".
_OS = (("Android", "Android"), ("iPhone", "iOS"), ("iPad", "iOS"), ("iPod", "iOS"),
       ("Windows", "Windows"), ("CrOS", "ChromeOS"), ("Mac OS X", "macOS"),
       ("Linux", "Linux"))
_BROWSERS = (("Edg", "Edge"), ("OPR/", "Opera"), ("SamsungBrowser", "Samsung"),
             ("Firefox/", "Firefox"), ("FxiOS", "Firefox"), ("CriOS", "Chrome"),
             ("Chrome/", "Chrome"), ("Safari/", "Safari"))
try:
    _LOCAL_TZ = ZoneInfo("Europe/Prague")
except ZoneInfoNotFoundError:  # an image without tzdata: hours in UTC, still serving
    _LOCAL_TZ = timezone.utc
_MAX_UA = 300
_PRAGUE = {"Prague", "Hlavní město Praha"}  # DB-IP and Cloudflare region names
_PRAHA_DISTRICT = re.compile(r"Praha \((Praha[^)]*)\)")  # Praha (Praha 5)
_CZECH_ISP = re.compile(r"(.+) Czech( Republic)?(,? (a\.s\.|s\.r\.o\.))?")
_VISIT_GAP = timedelta(minutes=30)  # Google Analytics' default session timeout

_log = logging.getLogger("ajaj.visits")
_write_lock = threading.Lock()


def tracked(method: str, path: str) -> tuple[str, str] | None:
    """(site, kind) when this request counts as a visit, else None."""
    return _TRACKED.get((method, path))


def client_ip(headers: Mapping[str, str], peer: str) -> tuple[str, str, str]:
    """(ip, via, country) of the visitor. Country is only known via Cloudflare."""
    cf_ip = headers.get("cf-connecting-ip")
    if cf_ip:
        return cf_ip, "cloudflare", headers.get("cf-ipcountry", "")
    forwarded = headers.get("x-forwarded-for", "")
    return forwarded.split(",")[0].strip() or peer, "direct", ""


def classify(user_agent: str, has_session: bool, is_internal: bool = False) -> str:
    """'internal' (our own checks, admin token), 'bot', 'human' (solved the
    captcha) or 'unverified'."""
    if is_internal:
        return "internal"
    if not user_agent or _BOT_RE.search(user_agent):
        return "bot"
    return "human" if has_session else "unverified"


def _match(user_agent: str, table: tuple[tuple[str, str], ...]) -> str:
    return next((name for needle, name in table if needle in user_agent), "Other")


def os_name(user_agent: str) -> str:
    """Operating system family from a user agent, 'Other' when unknown."""
    return _match(user_agent, _OS)


def browser(user_agent: str) -> str:
    """Browser family from a user agent, 'Other' when unknown."""
    return _match(user_agent, _BROWSERS)


def _utf8(value: str) -> str:
    """Cloudflare sends location headers as UTF-8; Starlette reads every header
    as Latin-1, so "Hlavní město" arrives garbled until it is re-decoded."""
    try:
        return value.encode("latin-1").decode("utf-8")
    except (UnicodeEncodeError, UnicodeDecodeError):
        return value


def _referrer_host(referer: str) -> str:
    """Host only: the full URL can carry search terms or tokens."""
    try:
        return urlsplit(referer).hostname or ""
    except ValueError:
        return ""


def record(site: str, kind: str, headers: Mapping[str, str], peer: str,
           has_session: bool, is_internal: bool = False,
           visitor_id: str = "", solved_ts: int = 0) -> None:
    """Append one visit. Never raises: stats must not break serving.
    `visitor_id` comes from the captcha token, never the token itself;
    `solved_ts` is when that token was issued, so a visitor who solved before
    the log began still counts as returning."""
    solved = datetime.fromtimestamp(solved_ts, _LOCAL_TZ).date().isoformat() if solved_ts else ""
    ip, via, country = client_ip(headers, peer)
    user_agent = headers.get("user-agent", "")[:_MAX_UA]
    line = json.dumps({
        "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "site": site, "kind": kind, "ip": ip, "via": via, "country": country,
        "ua": user_agent, "class": classify(user_agent, has_session, is_internal),
        "ref": _referrer_host(headers.get("referer", "")), "vid": visitor_id,
        "solved": solved,
        "cf_city": _utf8(headers.get("cf-ipcity", "")),
        "cf_region": _utf8(headers.get("cf-region", "")),
    })
    try:
        with _write_lock, LOG_FILE.open("a", encoding="utf-8") as file:
            file.write(line + "\n")
    except OSError as error:
        _log.warning("visit not recorded: %s", error)


def load() -> list[dict]:
    """Every recorded visit, oldest first."""
    # ponytail: reads the whole file per view; fine until it's tens of MB.
    try:
        with LOG_FILE.open(encoding="utf-8") as file:
            return [json.loads(line) for line in file if line.strip()]
    except FileNotFoundError:
        return []


def _place(city: str, region: str) -> str:
    """"city, region"; Prague names shortened to Czech ("Praha 5", not
    "Prague (Praha 5), Hlavní město Praha")."""
    if region in _PRAGUE:
        city = city.replace("Hlavní město Praha", "Praha").replace("Prague", "Praha")
        city = city.replace(" (Praha)", "")
        if match := _PRAHA_DISTRICT.fullmatch(city):
            city = match[1]
        region = "" if city.startswith("Praha") else "Praha"
    return ", ".join(part for part in (city, region) if part)


def short_isp(isp: str) -> str:
    """"Vodafone Czech Republic a.s." -> "Vodafone CZ"; others unchanged."""
    match = _CZECH_ISP.fullmatch(isp)
    return f"{match[1]} CZ" if match else isp


def fill_locations(records: list[dict]) -> None:
    """Add DB-IP's place ("dbip") and network owner ("isp"), and Cloudflare's
    place ("cf", only once its location headers are on), in place. Country
    comes from Cloudflare when it sent one, else DB-IP."""
    for r in records:
        found = geo.lookup(r["ip"])
        r["country"] = r["country"] or found["country"]
        r["dbip"] = _place(found["city"], found["region"])
        r["isp"] = short_isp(found["isp"])
        r["cf"] = _place(r.get("cf_city", ""), r.get("cf_region", ""))


def since_days(records: list[dict], days: int,
               now: datetime | None = None) -> list[dict]:
    """Records from Prague midnight `days - 1` days ago on; days=1 is today."""
    today = (now or datetime.now(timezone.utc)).astimezone(_LOCAL_TZ).date()
    start = datetime.combine(today - timedelta(days=days - 1), time.min, _LOCAL_TZ)
    return [r for r in records if datetime.fromisoformat(r["ts"]) >= start]


def _monday(day: str) -> str:
    start = date.fromisoformat(day)
    return (start - timedelta(days=start.weekday())).isoformat()


_UNITS = ("days", "weeks", "months", "hours")


def _buckets(ts: str) -> tuple[tuple[str, str], ...]:
    """(unit, bucket) pairs a timestamp falls in, Prague time: its day, week
    (its Monday), month and hour of day."""
    local = datetime.fromisoformat(ts).astimezone(_LOCAL_TZ)
    day = local.date().isoformat()
    return (("days", day), ("weeks", _monday(day)), ("months", day[:7]),
            ("hours", local.strftime("%H")))


def visit_starts(records: list[dict]) -> list[dict]:
    """The first record of each visit. A visitor's request 30 minutes or more
    after their previous one starts a new visit. Records oldest first."""
    last: dict[str, datetime] = {}
    starts = []
    for r in records:
        ts = datetime.fromisoformat(r["ts"])
        previous = last.get(r["visitor"])
        if previous is None or ts - previous >= _VISIT_GAP:
            starts.append(r)
        last[r["visitor"]] = ts
    return starts


def promote(records: list[dict]) -> None:
    """Set each record's "visitor", its order of first appearance ("n") and
    that visitor's best class, in place, so
    a captcha solve upgrades all its visits to human; a human who never used
    the record's site becomes "idle" there. Run it on the whole log before
    cutting a period, to see past solves and uses; summarize needs it done."""
    # ponytail: a pair seen with two tokens (same IP and browser, e.g. a shared
    # office) lends its token-less hits to the first one.
    token_of: dict[tuple[str, str], str] = {}
    for r in records:
        if r.get("vid"):
            token_of.setdefault((r["ip"], r["ua"]), r["vid"])
    best: dict[str, str] = {}
    order: dict[str, int] = {}
    for r in records:
        pair = (r["ip"], r["ua"])
        visitor: str = r.get("vid") or token_of.get(pair) or f"{r['ip']} {r['ua']}"
        r["visitor"] = visitor
        r["n"] = order.setdefault(visitor, len(order))  # stable across periods: picks the colour
        cls: str = r["class"]
        if _RANK[cls] >= _RANK[best.get(visitor, cls)]:
            best[visitor] = cls
    used = {(r["visitor"], r["site"]) for r in records if r["kind"] != "page"}
    humans: dict[str, int] = {}
    for r in records:
        r["class"] = best[r["visitor"]]
        if r["class"] == "human":  # "Visitor 1, 2, …": humans only, stable across periods
            r["hn"] = humans.setdefault(r["visitor"], len(humans) + 1)
        if r["class"] == "human" and (r["visitor"], r["site"]) not in used:
            r["class"] = "idle"


def summarize(records: list[dict]) -> dict[str, dict]:
    """Per-site numbers, every count split by visitor class. Records must have
    been through `promote`."""
    return {site: _summarize_site([r for r in records if r["site"] == site])
            for site in ("main", "t")}


def _summarize_site(records: list[dict]) -> dict:
    visitors: dict[str, dict] = {}
    for r in records:
        visitor = visitors.setdefault(r["visitor"], {
            "class": "bot", "via": r["via"], "country": "", "referrer": "",
            "dbip": "", "cf": "", "isp": "",
            "os": os_name(r["ua"]), "browser": browser(r["ua"]),
            "seen": set(), "engaged": False, "n": r["n"], "hn": r.get("hn", 0),
            "solved": ""})
        visitor["class"] = r["class"]
        visitor["referrer"] = visitor["referrer"] or r.get("ref", "")  # the first
        # Location: the latest known, so a visitor who moved (or switched VPN
        # exit) shows where they are now. dbip/cf/isp are set by fill_locations.
        for field in ("country", "dbip", "cf", "isp"):
            visitor[field] = r.get(field) or visitor[field]
        visitor["engaged"] |= r["kind"] != "page"
        visitor["solved"] = min(filter(None, (visitor["solved"], r.get("solved"))), default="")
        visitor["seen"].update(_buckets(r["ts"]))
    everyone = list(visitors.values())
    timeline: dict[str, dict[str, Counter]] = {unit: defaultdict(Counter) for unit in _UNITS}
    for visitor in everyone:
        visitor["country"] = visitor["country"] or "??"
        days = [bucket for unit, bucket in visitor["seen"] if unit == "days"]
        visitor["days"] = len(days)
        # Back on another day: seen on two, or solved the captcha before the first.
        visitor["returning"] = len(days) > 1 or "" < visitor["solved"] < min(days)
        for unit, bucket in visitor["seen"]:
            timeline[unit][bucket][visitor["class"]] += 1
    bots = Counter(r["ua"] for r in records if r["class"] == "bot")
    return {
        "visitors": dict(Counter(v["class"] for v in everyone)),
        "events": _by(records, "kind"),
        "engaged": dict(Counter(v["class"] for v in everyone if v["engaged"])),
        "returning": dict(Counter(v["class"] for v in everyone if v["returning"])),
        **{unit: _sorted(counts) for unit, counts in timeline.items()},
        **_visit_numbers(records, visitors),
        "via": _by(everyone, "via"),
        "countries": _by(everyone, "country"),
        "referrers": _by(everyone, "referrer"),
        "places_dbip": _by(everyone, "dbip"),
        "places_cf": _by(everyone, "cf"),
        "isps": _by(everyone, "isp"),
        "os": _by(everyone, "os"),
        "browsers": _by(everyone, "browser"),
        "bot_agents": [list(pair) for pair in bots.most_common(10)],
    }


def _visit_numbers(records: list[dict], visitors: dict[str, dict]) -> dict:
    """Visits by class per unit (day, week, month, hour of day), plus each
    human (or idle) visitor's own timeline
    ({unit: {bucket: visits}}, 0 for a bucket seen without a visit starting in
    it), humans in order of first appearance."""
    starts = visit_starts(records)
    timeline: dict[str, dict[str, Counter]] = {unit: defaultdict(Counter) for unit in _UNITS}
    humans: dict[str, dict[str, Counter]] = {
        key: {unit: Counter() for unit in timeline}
        for key, visitor in visitors.items() if visitor["class"] in ("human", "idle")}
    for key, units in humans.items():
        for unit, bucket in visitors[key]["seen"]:
            units[unit][bucket] = 0
    for r in starts:
        for unit, bucket in _buckets(r["ts"]):
            timeline[unit][bucket][r["class"]] += 1
            if r["visitor"] in humans:
                humans[r["visitor"]][unit][bucket] += 1
    uses = _uses(records, humans)
    return {
        "visits": dict(Counter(r["class"] for r in starts)),
        "visit_timeline": {unit: _sorted(counts) for unit, counts in timeline.items()},
        "humans": [{field: visitors[key][field]
                    for field in ("n", "hn", "solved", "country", "os", "browser", "dbip", "cf",
                                  "isp")}
                   | {"idle": visitors[key]["class"] == "idle"}
                   | {unit: dict(counts) for unit, counts in units.items()}
                   | {"uses": {unit: dict(c) for unit, c in uses[key].items()}}
                   for key, units in humans.items()],
    }


def _uses(records: list[dict], humans: dict[str, dict]) -> dict[str, dict[str, Counter]]:
    """Each human's uses (chats on main, transcriptions on /t) per unit bucket."""
    uses = {key: {unit: Counter() for unit in _UNITS} for key in humans}
    for r in records:
        if r["kind"] != "page" and r["visitor"] in uses:
            for unit, bucket in _buckets(r["ts"]):
                uses[r["visitor"]][unit][bucket] += 1
    return uses


def _by(visitors: list[dict], field: str) -> dict[str, dict[str, int]]:
    """{value of field: {class: count}}, of visitors or of records."""
    counts: dict[str, Counter] = defaultdict(Counter)
    for visitor in visitors:
        counts[visitor[field]][visitor["class"]] += 1
    return {value: dict(n) for value, n in counts.items()}


def _sorted(counts: dict[str, Counter]) -> dict[str, dict[str, int]]:
    return {key: dict(counts[key]) for key in sorted(counts)}