"""Varly server-side A/B tests for Python: a small, dependency-free example.

    varly = Varly(site_key="pk_…", server_key=os.environ["VARLY_SERVER_KEY"])
    d = varly.decide("<experiment-id>", cookies=request.cookies, user_agent=request.headers.get("User-Agent"))
    if d["variant"] == "b": ...

- Fetches the site's config at most every 30 seconds (with an ETag), so pausing a test in Varly
  reaches your server within a minute.
- Assigns visitors exactly like the Varly snippet: the same visitor gets the same version.
- Without the _varly_id cookie (no consent yet, or a new visitor): the original, nothing recorded.
- Exposures are sent in the background every 5 seconds, once per visitor and test.
"""

import json
import math
import re
import threading
import time
import urllib.error
import urllib.request

BUCKETS = 10000
VISITOR_ID = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
BOT_UA = re.compile(
    r"(varlyreview|bot|crawler|spider|crawl|slurp|bingpreview|facebookexternalhit|embedly|quora link preview"
    r"|headless|phantom|puppeteer|playwright|lighthouse|pagespeed|gtmetrix|pingdom|uptime|monitor|curl|wget"
    r"|python-requests|aiohttp|httpx|axios|go-http-client|java/|okhttp|scrapy|semrush|ahrefs|screaming frog)",
    re.I,
)


def fnv1a(text):
    h = 0x811C9DC5
    for ch in text:
        h ^= ord(ch)
        h = (h * 0x01000193) & 0xFFFFFFFF
    return h


def bucket(experiment_id, visitor_id):
    return fnv1a(experiment_id + ":" + visitor_id) % BUCKETS


def js_round(x):
    # Same as JavaScript's Math.round for positive numbers (Python's round() rounds half to even).
    return int(math.floor(x + 0.5))


def assign(test, visitor_id):
    variants = [v for v in test.get("variants", []) if v.get("weight", 0) > 0]
    if not variants:
        return None
    if len(variants) == 1:
        return variants[0]["key"]
    total = sum(v["weight"] for v in variants)
    b = bucket(test["id"], visitor_id)
    acc = 0
    for v in variants:
        acc += v["weight"]
        if b < js_round((acc / total) * BUCKETS):
            return v["key"]
    return variants[-1]["key"]


def device_from_user_agent(ua):
    if re.search(r"iPad|Tablet|PlayBook|Silk", ua, re.I):
        return "tablet"
    if re.search(r"Mobi|Android|iPhone|iPod", ua, re.I):
        return "mobile"
    return "desktop"


def in_test(experiment_id, targeting, visitor_id, device=None, returning=None, channel=None):
    t = targeting or {}
    if t.get("devices") and device not in t["devices"]:
        return False
    if t.get("visitor") and (returning is None or (t["visitor"] == "returning") != bool(returning)):
        return False
    if t.get("channels") and channel not in t["channels"]:
        return False
    share = t.get("share", 1)
    if not share > 0:
        return False
    return share >= 1 or bucket(experiment_id + ":traffic", visitor_id) < js_round(share * BUCKETS)


class Varly:
    def __init__(self, site_key, server_key, endpoint="https://cdn.varly.ai", max_stale=600):
        self.site_key = site_key
        self.server_key = server_key
        self.endpoint = endpoint.rstrip("/")
        self.max_stale = max_stale
        self._config, self._etag, self._fetched = None, None, 0.0
        self._queue, self._sent = [], {}
        self._lock = threading.Lock()
        threading.Thread(target=self._loop, daemon=True).start()

    def _request(self, method, path, body=None, etag=None):
        req = urllib.request.Request(
            "%s%s?site=%s" % (self.endpoint, path, self.site_key),
            data=None if body is None else json.dumps(body).encode(),
            method=method,
            headers={"Authorization": "Bearer " + self.server_key, "Content-Type": "application/json"},
        )
        if etag:
            req.add_header("If-None-Match", etag)
        try:
            with urllib.request.urlopen(req, timeout=3) as res:
                return res.status, res.read(), res.headers.get("ETag")
        except urllib.error.HTTPError as e:
            return e.code, b"", None

    def refresh(self):
        try:
            status, raw, etag = self._request("GET", "/v1/server-config", etag=self._etag)
            if status == 304:
                self._fetched = time.time()
            elif status == 200:
                self._config, self._etag, self._fetched = json.loads(raw), etag, time.time()
        except Exception:  # Varly not reachable: keep the last config for max_stale seconds
            pass

    def flush(self):
        with self._lock:
            batch, self._queue = self._queue[:500], self._queue[500:]
        if batch:
            try:
                self._request("POST", "/v1/exposures", {"exposures": batch})
            except Exception:
                with self._lock:
                    self._queue = batch + self._queue

    def _loop(self):
        while True:
            if time.time() - self._fetched >= (self._config or {}).get("refresh", 30):
                self.refresh()
            self.flush()
            time.sleep(5)

    def decide(self, experiment_id, cookies=None, visitor_id=None, user_agent=None,
               device=None, returning=None, channel=None, track=True,
               anonymous_id=None, opt_out=False):
        """opt_out: True when the request has DNT: 1 or Sec-GPC: 1.
        anonymous_id: only with privacy mode on, the _varly_anon ID your page sends along."""
        out = {"experimentId": experiment_id, "variant": "control", "inTest": False,
               "assigned": None, "dryRun": False, "reason": "not_running", "privacy": False}
        config = self._config if time.time() - self._fetched <= self.max_stale else None
        test = next((t for t in (config or {}).get("tests", []) if t["id"] == experiment_id), None)
        if not test or len(test.get("variants", [])) < 2:
            return out
        visitor = visitor_id if visitor_id is not None else (cookies or {}).get("_varly_id")
        privacy = False
        if not visitor or not VISITOR_ID.match(visitor):
            # Privacy mode: a visitor without consent takes part anonymously, never with DNT or GPC.
            if (config or {}).get("privacy") and not opt_out and anonymous_id and VISITOR_ID.match(anonymous_id):
                visitor, privacy = anonymous_id, True
            else:
                return dict(out, reason="no_visitor_id")
        if user_agent is not None and (not user_agent.strip() or BOT_UA.search(user_agent)):
            return dict(out, reason="bot")
        if device is None and user_agent:
            device = device_from_user_agent(user_agent)
        if not in_test(test["id"], test.get("targeting"), visitor, device, returning, channel):
            return dict(out, reason="not_targeted")
        assigned = assign(test, visitor)
        dry_run = test.get("mode") != "live"
        decision = {"experimentId": experiment_id, "variant": "control" if dry_run else assigned,
                    "inTest": True, "assigned": assigned, "dryRun": dry_run, "reason": "in_test",
                    "privacy": privacy}
        key = experiment_id + ":" + visitor
        if track and self._sent.get(key, 0) < time.time():
            self._sent[key] = time.time() + 1800
            with self._lock:
                self._queue.append({"experiment_id": experiment_id, "variant_key": assigned,
                                    "visitor_id": visitor, "ts": int(time.time() * 1000), "device": device,
                                    "privacy_mode": 1 if privacy else 0})
        return decision