Build your own AI assistant for Octo Browser: manage profiles with text commands

python -m pip install requests playwright

irm https://claude.ai/install.ps1 | iex

[Environment]::SetEnvironmentVariable("Path", [Environment]::GetEnvironmentVariable("Path", "User") + ";$env:USERPROFILE\.local\bin", "User")

cd D:\
mkdir project_claude
cd project_claude

mkdir octo_api
cd octo_api
mkdir plans
mkdir scripts
mkdir proxies_in
mkdir .claude

Get-Clipboard | Set-Content -NoNewline token.txt

(Get-Item token.txt).Length

import csv
import hashlib
import json
import os
import re
import subprocess
import sys
import signal
import time
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
import requests

DIR = Path(__file__).parent


def token():
    """The API token from token.txt, read only when a cloud request is sent. utf-8-sig: Notepad may add a BOM."""
    if "token" not in STATS:
        try:
            STATS["token"] = (DIR / "token.txt").read_text(encoding="utf-8-sig").strip()
        except FileNotFoundError:
            sys.exit("token.txt not found: put the API token from the Octo client (Settings, Additional) into token.txt.")
        if not STATS["token"] or not STATS["token"].isascii() or any(c.isspace() for c in STATS["token"]):
            sys.exit("token.txt must contain only the API token, one line without spaces.")
    return STATS["token"]

LOCAL_PORT = 58888  # Octo client settings, "Additional" tab, "Local HTTP server": change it here if yours differs
BASES = {"public": "https://app.octobrowser.net/api/v2/automation", "local": f"http://localhost:{LOCAL_PORT}/api"}
MAX_REQUESTS = 100  # max requests per plan
MAX_DELETE = 20  # max profiles deleted by one plan: a mass deletion is confirmed in parts
PAGE_LENS = ("10", "25", "50", "100")  # the only page sizes the API accepts
METHODS = ("GET", "POST", "PATCH", "DELETE")
ALLOW_DELETE = True  # False forbids every deletion (profiles, tags, folders, proxies), whoever asks for it
PROTECTED_TAG = "protected"  # profiles with this tag can't be changed, deleted or transferred by the assistant
AUDIT_TAG, AUDIT_FOLDER = "Audit-check", "For check"  # how the audit marks profiles that need a look
CHECK_PROFILE = "check-proxy"  # a dedicated empty profile, the only one used to test proxies
# Octo and people name some countries differently: both sides of the warm-up country check are compared by these
COUNTRY_ALIASES = {"czech republic": "czechia", "cesko": "czechia", "česko": "czechia", "česká republika": "czechia",
                   "uk": "united kingdom", "great britain": "united kingdom", "britain": "united kingdom",
                   "england": "united kingdom", "usa": "united states", "us": "united states", "united states of america": "united states",
                   "america": "united states", "holland": "netherlands", "nederland": "netherlands",
                   "turkey": "türkiye", "turkiye": "türkiye", "russian federation": "russia", "south korea": "korea",
                   "republic of korea": "korea", "korea republic of": "korea", "viet nam": "vietnam",
                   "deutschland": "germany", "österreich": "austria", "osterreich": "austria", "schweiz": "switzerland",
                   "suisse": "switzerland", "españa": "spain", "espana": "spain", "italia": "italy", "polska": "poland",
                   "slovensko": "slovakia", "slovak republic": "slovakia", "magyarország": "hungary", "românia": "romania",
                   "uae": "united arab emirates", "macedonia": "north macedonia", "moldova republic of": "moldova",
                   "republic of moldova": "moldova", "ivory coast": "côte d'ivoire", "cote d'ivoire": "côte d'ivoire",
                   "burma": "myanmar", "swaziland": "eswatini", "cape verde": "cabo verde", "hong kong sar": "hong kong"}


def country(name):
    """One spelling per country: 'The Czech Republic', 'Czechia' and 'Česko' all become 'czechia'."""
    name = " ".join(str(name or "").lower().replace(",", " ").replace("the ", " ").split())
    return COUNTRY_ALIASES.get(name, name)


WARM_DEFAULTS = {"pages_per_site": 3, "time_per_page": [8, 20], "scroll": "normal", "random_order": True,
                 "accept_consent": True, "external": False, "headless": False, "images": True}
# Sensitive sections: passwords, cookies, client login, client update and team management.
# Requests to them always require confirmation, even reads.
SENSITIVE = ("password", "cookies", "/auth/", "/update", "/teams")
SAFE_GET = {"/teams/extensions"}  # the team extension list is read without confirmation
# Actions that cannot be undone
IRREVERSIBLE = ("/transfer", "/export", "cookies", "password", "force_stop")  # force stop drops unsynced work
# Proxy and VPN secrets come back in clear text: hide them in the chat, keep them in --out files for the user
SECRET = re.compile(r"login|pass|secret|private|token|conf|key|change_ip", re.I)  # conf: a VPN's settings
# Starting and stopping profiles doesn't change any data, so no confirmation is needed
SAFE_LOCAL = {"/profiles/start", "/profiles/stop"}
# What local client calls cost in Octo's rate limits: the rest of the local API counts 1 like the cloud
LOCAL_COSTS = {"/profiles/active": 0, "/username": 0, "/update": 0, "/auth/login": 0, "/auth/logout": 0,
               "/profiles/one_time/start": 4, "/profiles/force_stop": 2}
# ...but only with these fields and flags: anything else (only_local=false, --remote-debugging-address, profile_data)
# opens the profile to the network or changes it, so it goes through show and confirmation
SAFE_START_KEYS = {"uuid", "headless", "debug_port", "timeout", "flags", "password"}
SAFE_FLAGS = {"--disable-backgrounding-occluded-windows", "--start-maximized", "--blink-settings=imagesEnabled=false",
              "--autoplay-policy=user-gesture-required"}
SESSION_LOCAL = {"/profiles/start", "/profiles/one_time/start", "/profiles/stop"}  # requests that update sessions.json
JOURNAL = DIR / "logs" / "actions.jsonl"
WARM_LOG = DIR / "logs" / "warm.jsonl"  # the farm log: one line per profile per warm-up, kept apart from the journal
INFLIGHT = DIR / "logs" / "inflight.json"  # profiles started by warm or proxycheck and not stopped yet
SITES = DIR / "allowed_sites.json"  # sites browser.py may open: only the user adds them, with the site command
SESSIONS = DIR / "sessions.json"  # addresses of running profiles for browser.py and scripts
REPORTS = DIR / "reports"
PRIVATE = REPORTS / "private"  # --out files with passwords or keys: .claude/settings.json denies Claude reading them
PROXIES_IN = DIR / "proxies_in"  # files with proxies to add: only the executor reads them, Claude is not allowed to
# Official Octo API spec: plan checks only warn and never block
SPEC_URL = "https://raw.githubusercontent.com/octobrowser/documentation/main/openapi.json"
SPEC = DIR / "api_spec.json"
VALUES = DIR / "fingerprint_values.json"  # fingerprint values supported by Octo (OS versions, CPU, RAM), maintained by the team
PREFIX = "/api/v2/automation"
# Fields returned by the profile list: verifying them costs one request per 100 profiles
LIST_FIELDS = ("title", "description", "tags", "folder", "proxy", "start_pages", "storage_options",
               "launch_args", "images_load_limit", "local_cache")
BULK = ("export", "import", "force_stop", "transfer", "set_password")  # /profiles/<this> is an action, not a uuid
SNAPSHOT_FIELDS = {"title", "description", "tags", "folder", "proxy", "start_pages", "storage_options"}  # the list returns them
STARTED, STATS = time.time(), {"requests": 0, "prefetched": {}, "unreadable": set()}
USAGE = ("Commands: read <plan> [--out reports/file.csv] | show <plan> | run <plan> <code> [--out ...] | "
         "delete <plan> <code> | undo [operation number] | audit [--days 30] [--mark tag|folder] | "
         "proxycheck [--profile <titles>] [--proxy <name>] [--limit 20] [--octo [--failed]] | warm <plan> | site [address] | "
         "history [--last 20] [--profile <uuid>] | warmlog [--last 30] [--profile <uuid>] | "
         "export [--add fingerprint,pages,settings,other|all] [--tag a,b] [--any-tag a,b] [--folder f|none] [--search s] "
         "[--proxy p|none] [--password yes|no] [--created-after/--created-before/--active-after/--active-before 2026-10-01|7d] "
         "[--idle 30d] [--never-started] [--sort -created|-active|title] [--limit 20] [--exclude a,b] [--format xlsx|csv] "
         "[--sep ;] [--out reports/x.xlsx] | proxyimport [--geo] [--prefix DE] [--type socks5] [--file name.txt] | "
         "find [the filters of export] | start <title or uuid> [--headless] | stop <title or uuid> | "
         "any command with --help: print this and do nothing")


def norm(path):
    return path.split("?")[0].rstrip("/")


def sensitive(r):
    if r["method"] == "GET" and norm(r["path"]) in SAFE_GET:
        return False
    return any(word in r["path"] for word in SENSITIVE)


def exposed(r):
    """A local start that opens the profile beyond this computer or changes it: needs show and confirmation."""
    if not (r["api"] == "local" and norm(r["path"]) in SAFE_LOCAL):
        return None
    body = r.get("body") or {}
    if not isinstance(body, dict):
        return "the body must be an object"
    extra = set(body) - SAFE_START_KEYS
    if extra:
        return f"fields {', '.join(sorted(extra))}"
    if body.get("only_local") is False:
        return "only_local=false opens the debug port to the network"
    flags = body.get("flags") or []
    odd = [f for f in flags if not isinstance(f, str) or not (f in SAFE_FLAGS or f.startswith("--disk-cache-dir="))]
    return f"flags {', '.join(map(str, odd))}" if odd else None


def is_change(r):
    if r["method"] == "GET":
        return False
    return not (r["api"] == "local" and norm(r["path"]) in SAFE_LOCAL and not exposed(r))


def fill(value, n):
    """Replaces {n} with the number in every string of a repeated request."""
    if isinstance(value, str):
        return value.replace("{n}", str(n))
    if isinstance(value, list):
        return [fill(v, n) for v in value]
    if isinstance(value, dict):
        return {k: fill(v, n) for k, v in value.items()}
    return value


def expand(reqs):
    """Expands {"repeat": {"from": 1, "to": 100}, ...} into one request per number."""
    out = []
    for r in reqs:
        repeat = r.get("repeat")
        if not isinstance(repeat, dict):
            out.append(r)
            continue
        first, last = int(repeat.get("from", 1)), int(repeat.get("to", 0))
        if last - first + 1 > MAX_REQUESTS:
            sys.exit(f"Too many repeats: {last - first + 1}. At most {MAX_REQUESTS} per plan, split the task into parts.")
        out += [fill({k: v for k, v in r.items() if k != "repeat"}, n) for n in range(first, last + 1)]
    return out


def load(path):
    """Reads a plan and checks it against the limits."""
    try:
        raw = Path(path).read_bytes()
        STATS["code"] = hashlib.sha256(raw).hexdigest()[:6]  # the code covers exactly the bytes that will be sent
        plan = json.loads(raw.decode("utf-8-sig"))
    except (OSError, ValueError) as e:
        sys.exit(f"Can't read the plan {path}: {e}")
    if not isinstance(plan, dict) or not isinstance(plan.get("requests"), list) or not plan["requests"]:
        sys.exit('The plan must be {"summary": "...", "requests": [...]} with at least one request.')
    for n, r in enumerate(plan["requests"], 1):  # one spelling per field: the safety checks below compare exact values
        if not isinstance(r, dict):
            sys.exit(f"Request {n} must be an object.")
        r["method"] = str(r.get("method", "")).upper()
        r["api"] = str(r.get("api", "public")).lower()
        if r["method"] not in METHODS or r["api"] not in BASES:
            sys.exit(f"Request {n}: method must be one of {', '.join(METHODS)} and api public or local.")
        if not isinstance(r.get("path"), str) or not r["path"].startswith("/") or ".." in r["path"]:
            sys.exit(f"Request {n}: path must start with / and must not contain '..'.")
    reqs = expand(plan["requests"])
    if len(reqs) > MAX_REQUESTS:
        sys.exit(f"Too many requests: {len(reqs)}. At most {MAX_REQUESTS} per plan, split the task into parts.")
    for r in reqs:
        r.setdefault("api", "public")
        page_len = (r.get("params") or {}).get("page_len")
        if page_len is not None and str(page_len) not in PAGE_LENS:  # a 400 would cost 6 RPM and 1 RPH
            sys.exit(f"page_len {page_len} is not allowed: use {', '.join(PAGE_LENS)}. "
                     "To count profiles, read total_count from one page with page_len 10.")
        if r["method"] == "DELETE" and norm(r["path"]) == "/profiles":
            r.setdefault("body", {})["skip_trash_bin"] = False  # deleted profiles go to the trash
    deleting = sum(len((r.get("body") or {}).get("uuids", [])) for r in reqs
                   if r["method"] == "DELETE" and norm(r["path"]) == "/profiles")
    if deleting > MAX_DELETE:
        sys.exit(f"Too many profiles to delete: {deleting}. At most {MAX_DELETE} per plan, "
                 "so that a mass deletion is confirmed in parts.")
    return plan, reqs


class LimitStop(SystemExit):
    """Raised instead of sending a request when the hourly API budget is almost used up."""


class Interrupted(SystemExit):
    """The process was told to stop (Claude Code timeout, Ctrl+Break, closed terminal)."""


def call(req):
    """Sends one request and respects Octo rate limits."""
    stop = req["api"] == "local" and norm(req["path"]) == "/profiles/stop"  # a stop is always sent: it syncs the profile
    if STATS.get("hour_low") and not stop:  # sending more would extend the restriction, Octo warns about stricter limits
        raise LimitStop("The hourly API limit is almost used up. Wait for the next hour, then run the remaining part.")
    url = BASES[req["api"]] + req["path"]
    headers = {"X-Octo-Api-Token": token()} if req["api"] == "public" else {}
    cost = LOCAL_COSTS.get(norm(req["path"]), 1) if req["api"] == "local" else 1
    for attempt in range(6):  # gives up after several 429 answers in a row
        STATS["requests"] += cost
        r = requests.request(req["method"], url, headers=headers, params=req.get("params"),
                             json=req.get("body"), timeout=120)
        if r.status_code != 429 or attempt == 5:
            break
        retry = r.headers.get("Retry-After", "")
        wait = max(1, int(retry) if retry.isdigit() else 5)
        if wait > 60:  # a long ban: don't hang silently and don't knock again, stop and keep the rest of the plan
            STATS["hour_low"] = True
            break
        time.sleep(wait)
    hour_left = r.headers.get("X-Ratelimit-Remaining-Hour", "")
    if hour_left.isdigit() and int(hour_left) < 3:
        STATS["hour_low"] = True
    remaining = r.headers.get("X-Ratelimit-Remaining", "")
    if remaining.isdigit() and hour_left.isdigit():  # what is left, for the last line of the command
        STATS["left"] = (remaining, r.headers.get("X-Ratelimit-Limit", ""), hour_left, r.headers.get("X-Ratelimit-Limit-Hour", ""))
    if remaining.isdigit() and int(remaining) < 5:  # pause until the window resets, before the limit runs out
        reset = r.headers.get("X-Ratelimit-Reset", "")
        wait = int(reset) - time.time() + 1 if reset.isdigit() else 5
        time.sleep(min(max(wait, 1), 60))
    try:
        body = r.json()
    except ValueError:
        body = r.text
    return {"request": f"{req['method']} {req['path']}", "status": r.status_code, "response": body}


def snapshot(req):
    """Saves the previous field values of a profile so the change can be undone."""
    keys = set(req.get("body") or {})
    data = STATS["prefetched"].get(norm(req["path"]).split("/")[2])
    if not (isinstance(data, dict) and keys <= SNAPSHOT_FIELDS and keys <= set(data)):  # read the profile itself
        got = call({"api": "public", "method": "GET", "path": req["path"]})
        data = got["response"].get("data") if isinstance(got["response"], dict) else None
        if got["status"] != 200 or not isinstance(data, dict):
            return None
    before = {}
    for key, value in (req.get("body") or {}).items():
        old = data.get(key)
        before[key] = {k: (old or {}).get(k) for k in value} if isinstance(value, dict) and isinstance(old, dict) else old
    return {"api": "public", "method": "PATCH", "path": req["path"], "body": before}


def load_spec():
    """Downloads the spec once a week and keeps a copy next to the script."""
    try:
        if not SPEC.exists() or time.time() - SPEC.stat().st_mtime > 7 * 86400:
            r = requests.get(SPEC_URL, timeout=20)
            r.raise_for_status()
            SPEC.write_text(r.text, encoding="utf-8")
    except Exception:
        pass  # no network: use the saved copy if there is one
    try:
        return json.loads(SPEC.read_text(encoding="utf-8"))
    except Exception:
        return None


def schema(spec, s):
    while isinstance(s, dict) and "$ref" in s:
        s = spec["components"]["schemas"][s["$ref"].split("/")[-1]]
    return s or {}


def check_fields(spec, sch, body, where, notes):
    sch = schema(spec, sch)
    props = sch.get("properties")
    if not isinstance(body, dict) or not props:
        return
    for key in sch.get("required", []):
        if key not in body:
            notes.append(f"{where}: missing required field '{key}'")
    for key, value in body.items():
        if key not in props:
            notes.append(f"{where}: field '{key}' is not in the spec (a typo or a new field?)")
            continue
        sub = schema(spec, props[key])
        allowed = sub.get("enum") or schema(spec, sub.get("items", {})).get("enum")
        for v in (value if isinstance(value, list) else [value]):
            if allowed and not isinstance(v, (dict, list)) and v not in allowed:
                notes.append(f"{where}: value '{v}' of field '{key}' is not allowed ({', '.join(map(str, allowed))})")
        if isinstance(value, dict):
            check_fields(spec, sub, value, f"{where}.{key}", notes)


def spec_notes(reqs):
    """Checks plan requests against the Octo spec and returns warnings."""
    spec = load_spec()
    if not spec:
        return ["the Octo spec is unavailable, the plan was not checked"]
    notes = []
    for r in reqs:
        if r["api"] != "public":
            continue  # the spec covers only the cloud API
        where = f"{r['method']} {norm(r['path'])}"
        matches = sorted((k.count("{"), v) for k, v in spec["paths"].items()
                         if re.fullmatch(re.sub(r"\\{[^/]+?\\}", "[^/]+", re.escape(k)), PREFIX + norm(r["path"])))
        op = next((v[r["method"].lower()] for _, v in matches if r["method"].lower() in v), None)
        if not op:
            notes.append(f"{where}: method is not in the spec (wrong path?)")
            continue
        known = {p.get("name"): p for p in op.get("parameters", [])}
        for name, value in (r.get("params") or {}).items():
            if name not in known:
                notes.append(f"{where}: parameter '{name}' is not in the spec")
                continue
            sub = schema(spec, known[name].get("schema", {}))
            for part in sub.get("allOf", []):  # page_len and others refer to a shared list of values
                sub = {**sub, **schema(spec, part)}
            allowed = sub.get("enum")
            if allowed and str(value) not in map(str, allowed):
                notes.append(f"{where}: value '{value}' of parameter '{name}' is not allowed ({', '.join(map(str, allowed))})")
        body_schema = op.get("requestBody", {}).get("content", {}).get("application/json", {}).get("schema")
        if body_schema and r.get("body") is not None:
            found, covered = [], covered_fields(r)
            check_fields(spec, body_schema, r["body"], where, found)
            notes += [n for n in found  # fields from fingerprint_values.json are checked there instead
                      if not (n.startswith(f"{where}.fingerprint") and any(f"'{f}'" in n for f in covered))]
    return list(dict.fromkeys(notes))


def load_values():
    try:
        return json.loads(VALUES.read_text(encoding="utf-8"))
    except Exception:
        return {}


def kind_of(fp):
    """Device type for Android (phone, tablet), architecture otherwise (arm, x86)."""
    return fp.get("device_type") if fp.get("os") == "android" else fp.get("os_arch")


def covered_fields(r):
    """Fingerprint fields checked against fingerprint_values.json: the client lists override the spec."""
    fp = (r.get("body") or {}).get("fingerprint")
    values = load_values()
    if not isinstance(fp, dict) or fp.get("os") not in values:
        return set()
    kind = kind_of(fp)
    covered = {field for field, allowed in values[fp["os"]].items()
               if not isinstance(allowed, dict) or not kind or kind in allowed}
    extra = {"device_type", "device_model"} if fp["os"] == "android" else set()
    return covered | {"os"} | extra


def fingerprint_notes(reqs):
    """Checks fingerprint values (OS versions, CPU, RAM) against fingerprint_values.json."""
    values = load_values()
    notes = []
    for r in reqs:
        fp = (r.get("body") or {}).get("fingerprint")
        if r["method"] == "PATCH" and isinstance(fp, dict) and ({"os", "os_version"} & set(fp)):
            notes.append(f"PATCH {norm(r['path'])}: os and os_version can't be changed after the profile is created "
                         "(the API answers 400, which costs 6 RPM)")
            continue
        if not isinstance(fp, dict) or fp.get("os") not in values:
            continue
        where = f"{r['method']} {norm(r['path'])}"
        if fp["os"] == "android" and r["method"] == "POST" and not fp.get("device_model"):
            notes.append(f"{where}: android profiles need device_model from GET /fingerprint/device_models")
        kind = kind_of(fp)
        for field, allowed in values[fp["os"]].items():
            if field not in fp:
                continue
            if isinstance(allowed, dict):
                if kind and kind not in allowed:
                    if field == "os_version":  # os_version lists every kind, so an unknown kind is an error
                        notes.append(f"{where}: '{kind}' is not supported for {fp['os']} "
                                     f"(fingerprint_values.json: {', '.join(allowed)})")
                    continue  # no list for this kind: nothing to check
                allowed = allowed.get(kind) or sorted({str(v) for vs in allowed.values() for v in vs}, key=int)
            if str(fp[field]) not in map(str, allowed):
                target = f"{fp['os']} {kind}" if kind else fp["os"]
                notes.append(f"{where}: {field} '{fp[field]}' is not supported for {target} "
                             f"(fingerprint_values.json: {', '.join(map(str, allowed))})")
    return notes


def names(value):
    """Tags, folder or proxy may come as strings or objects: normalize them to names and uuids."""
    items = value if isinstance(value, list) else [value]
    return {str(x.get("name") or x.get("uuid")) if isinstance(x, dict) else str(x) for x in items if x is not None}


def differences(expected, actual):
    """Compares what was requested with what the API returns. Skips fields that can't be compared reliably."""
    diffs = []
    for key, want in expected.items():
        have = actual.get(key)
        if key in ("tags", "start_pages", "launch_args"):
            missing = names(want) - names(have)
            if missing:
                diffs.append(f"{key}: missing {sorted(missing)}")
        elif key == "folder":
            if names(want) != names(have):
                diffs.append(f"folder: expected '{want}', got '{have}'")
        elif key == "proxy" and isinstance(want, dict) and want.get("uuid"):
            if not isinstance(have, dict) or have.get("uuid") != want["uuid"]:
                diffs.append("proxy: a different proxy is assigned")
        elif key in ("fingerprint", "storage_options") and isinstance(want, dict):
            for sub, value in want.items():
                if isinstance(have, dict) and sub in have and have[sub] != value:
                    diffs.append(f"{key}.{sub}: expected {value!r}, got {have[sub]!r}")
        elif key == "bookmarks" and isinstance(want, list):
            urls = {b.get("url") for b in have or [] if isinstance(b, dict)}
            if {b.get("url") for b in want if isinstance(b, dict)} - urls:
                diffs.append("bookmarks: some bookmarks are missing")
        elif key in ("title", "description", "images_load_limit", "local_cache") and have != want:
            diffs.append(f"{key}: expected {want!r}, got {have!r}")
    return diffs


def fetch(uuids, full, ordering=None):
    """Re-reads profiles: pages through the list when there are many, otherwise one by one."""
    found, scanned_all, spent, page = {}, False, 0, 0
    if not full and len(uuids) > 3:
        while True:
            params = {"page_len": 100, "page": page, "fields": ",".join(LIST_FIELDS)}
            if ordering:  # for new profiles "-created" puts them on the first page
                params["ordering"] = ordering
            res = call({"api": "public", "method": "GET", "path": "/profiles", "params": params})
            spent += 1
            if res["status"] != 200:  # the list failed: read the rest one by one below
                break
            resp = res["response"] if isinstance(res["response"], dict) else {}
            data = resp.get("data") or []
            found.update({p["uuid"]: p for p in data if p.get("uuid") in uuids})
            page += 1
            if not data or page * 100 >= (resp.get("total_count") or 0):
                scanned_all = True
                break
            if len(found) == len(uuids) or page >= len(uuids) - len(found):
                break
    for uuid in uuids:
        if uuid not in found and not scanned_all:
            res = call({"api": "public", "method": "GET", "path": f"/profiles/{uuid}"})
            spent += 1
            data = res["response"].get("data") if isinstance(res["response"], dict) else None
            if res["status"] == 200 and isinstance(data, dict):
                found[uuid] = data
            elif res["status"] != 404:  # not "no such profile" but "no answer": the protection check can't trust it
                STATS["unreadable"].add(uuid)
    return found, spent


def verify(reqs, results):
    """Checks that the changes were actually applied."""
    expect, absent, issues, unchecked = {}, set(), [], 0
    for req, res in zip(reqs, results):
        path = norm(req["path"])
        ok = 200 <= res["status"] < 300
        data = res["response"].get("data") if isinstance(res["response"], dict) else None
        body = req.get("body") or {}
        if not is_change(req):
            continue
        if not ok:
            issues.append(f"{req['method']} {req['path']}: response {res['status']}")
        elif req["method"] == "POST" and path == "/profiles" and isinstance(data, dict) and data.get("uuid"):
            expect[data["uuid"]] = {k: v for k, v in body.items() if k in LIST_FIELDS}
        elif req["method"] == "PATCH" and path.startswith("/profiles/") and path.count("/") == 2:
            expect.setdefault(path.split("/")[2], {}).update(body)
        elif req["method"] == "DELETE" and path == "/profiles":
            for uuid in (data or {}).get("active_uuids", []) if isinstance(data, dict) else []:
                issues.append(f"{uuid}: not deleted because the profile is running; stop it and retry")
        elif req["method"] == "POST" and path == "/profiles/transfer":
            absent.update(body.get("uuids", []))
        else:
            unchecked += 1
    full = any(k in ("fingerprint", "bookmarks") for b in expect.values() for k in b)
    created = any(r["method"] == "POST" and norm(r["path"]) == "/profiles" for r in reqs)
    found, spent = (fetch(list(expect) + sorted(absent), full, "-created" if created else None)
                    if expect or absent else ({}, 0))
    ok_count = 0
    for uuid, body in expect.items():
        if uuid not in found:
            issues.append(f"{uuid}: profile not found")
            continue
        diffs = differences(body, found[uuid])
        title = found[uuid].get("title") or uuid
        issues += [f"{title}: {d}" for d in diffs]
        ok_count += not diffs
    for uuid in absent:
        if uuid in found:
            issues.append(f"{found[uuid].get('title') or uuid}: the profile is still in your account, the transfer failed")
        else:
            ok_count += 1
    total = len(expect) + len(absent)
    lines = []
    if total:
        lines.append(f"Verification: {ok_count} of {total} match (requests used: {spent}).")
    elif issues:
        lines.append("Verification:")
    lines += [f"  {i}" for i in issues[:15]]
    if len(issues) > 15:
        lines.append(f"  ...and {len(issues) - 15} more")
    if unchecked:
        lines.append(f"  Not verified automatically: {unchecked} request(s), check the result in the client.")
    return {"ok": ok_count, "total": total, "issues": issues}, "\n".join(lines)


def targets(reqs):
    """Profiles that a plan changes, deletes or transfers."""
    uuids = []
    for r in reqs:
        if r["api"] != "public" or not is_change(r):
            continue
        path, body = norm(r["path"]), r.get("body") or {}
        if isinstance(body, dict) and isinstance(body.get("uuids"), list):  # bulk delete, transfer, export, force stop...
            uuids += [u for u in body["uuids"] if isinstance(u, str)]
        if path.startswith("/profiles/") and path.split("/")[2] not in BULK:
            uuids.append(path.split("/")[2])
    return list(dict.fromkeys(uuids))


def protected(reqs):
    """Profiles of the plan that have the protected tag: the plan must not touch them.
    Fails closed: a profile that could not be read is blocked too, and so is any change of the protected tag itself."""
    uuids = targets(reqs)
    found, _ = fetch(uuids, False) if uuids else ({}, 0)
    STATS["prefetched"] = found  # read just before the run: reused for undo snapshots
    blocked = [(u, found[u].get("title") or u) for u in uuids
               if u in found and PROTECTED_TAG in {t.lower() for t in names(found[u].get("tags"))}]
    blocked += [(u, "could not be read, so its tags can't be checked") for u in uuids if u in STATS["unreadable"]]
    tag_ids = {norm(r["path"]).split("/")[2] for r in reqs
               if r["method"] in ("PATCH", "DELETE") and norm(r["path"]).startswith("/tags/") and norm(r["path"]).count("/") == 2}
    if tag_ids:
        res = call({"api": "public", "method": "GET", "path": "/tags"})
        tags = (res["response"].get("data") if isinstance(res["response"], dict) else None) if res["status"] == 200 else None
        if not isinstance(tags, list):
            blocked += [(t, "tag: the tag list could not be read") for t in sorted(tag_ids)]
        else:
            blocked += [(t["uuid"], f"the tag '{t.get('name')}' itself") for t in tags if isinstance(t, dict)
                        and t.get("uuid") in tag_ids and str(t.get("name", "")).lower() == PROTECTED_TAG]
    return blocked


def remember_session(req, res):
    """Saves ws_endpoint and debug_port of a started profile for browser.py and scripts."""
    try:
        sessions = json.loads(SESSIONS.read_text(encoding="utf-8")) if SESSIONS.exists() else {}
    except ValueError:  # a broken file: start a new one rather than lose the address of a running profile
        sessions = {}
    uuid = (req.get("body") or {}).get("uuid")
    resp = res["response"] if isinstance(res["response"], dict) else {}
    if norm(req["path"]) in ("/profiles/start", "/profiles/one_time/start") and resp.get("ws_endpoint"):
        sessions[resp.get("uuid") or uuid] = {"ws_endpoint": resp["ws_endpoint"], "debug_port": resp.get("debug_port")}
    elif norm(req["path"]) == "/profiles/stop":
        sessions.pop(uuid, None)
    SESSIONS.write_text(json.dumps(sessions, ensure_ascii=False, indent=2), encoding="utf-8")


def alive(pid):
    """Is the process still running. On Windows os.kill(pid, 0) would kill it, so ask the system instead."""
    if not isinstance(pid, int) or pid <= 0:
        return False
    if os.name == "nt":
        import ctypes
        kernel = ctypes.windll.kernel32
        handle = kernel.OpenProcess(0x1000, False, pid)  # PROCESS_QUERY_LIMITED_INFORMATION
        if not handle:
            return False
        status = ctypes.c_ulong()
        ok = kernel.GetExitCodeProcess(handle, ctypes.byref(status))
        kernel.CloseHandle(handle)
        return bool(ok) and status.value == 259  # STILL_ACTIVE
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def inflight(uuid, resp=None, by=""):
    """Remembers a profile started by warm or proxycheck until it is stopped (resp=None forgets it)."""
    try:
        data = json.loads(INFLIGHT.read_text(encoding="utf-8")) if INFLIGHT.exists() else {}
    except ValueError:
        data = {}
    if resp is None:
        data.pop(uuid, None)
    else:
        data[uuid] = {"start_time": resp.get("start_time"), "pid": os.getpid(), "by": by}
    INFLIGHT.parent.mkdir(exist_ok=True)
    INFLIGHT.write_text(json.dumps(data, indent=2), encoding="utf-8")


def recover():
    """Stops profiles that a killed warm-up or proxy check left running: only the same session it started,
    never a profile the user has restarted since, and never while that run is still alive."""
    try:
        data = json.loads(INFLIGHT.read_text(encoding="utf-8")) if INFLIGHT.exists() else {}
    except ValueError:
        data = {}
    left = {u: v for u, v in data.items() if isinstance(v, dict) and not alive(v.get("pid"))}
    if not left:
        return
    try:
        active = call({"api": "local", "method": "GET", "path": "/profiles/active"})["response"]  # free request
    except (requests.RequestException, OSError):
        return  # the client is closed: its profiles are not running anyway, check again next time
    running = {x.get("uuid"): x for x in (active if isinstance(active, list) else []) if isinstance(x, dict)}
    for uuid, info in left.items():
        if uuid in running and running[uuid].get("start_time") == info.get("start_time"):
            stop = {"api": "local", "method": "POST", "path": "/profiles/stop", "body": {"uuid": uuid}}
            res = call(stop)
            if not 200 <= res["status"] < 300:
                print(f"A {info.get('by')} run was interrupted and left profile {uuid} running: it did not stop, "
                      "close it in the client.")
                continue
            remember_session(stop, res)
            print(f"A {info.get('by')} run was interrupted and left profile {uuid} running: stopped it normally, "
                  "so its data is synced.")
        inflight(uuid)


def execute(reqs, with_undo):
    """Runs the requests and collects reverse actions for undo."""
    results, undo, manual, created, stopped = [], [], [], [], None
    for i, req in enumerate(reqs, 1):
        path = norm(req["path"])
        before = None
        try:
            if with_undo and req["method"] == "PATCH" and req["api"] == "public" and path.startswith("/profiles/"):
                before = snapshot(req)
            res = call(req)
        except (LimitStop, Interrupted) as e:  # hourly budget or a kill: stop and keep what is done
            stopped = {"at": i, "reason": str(e)}
            break
        except (requests.RequestException, OSError) as e:  # no connection: stop and keep what is done
            hint = (f" Is the Octo client running? If its local port is not {LOCAL_PORT}, set LOCAL_PORT in octo_api.py."
                    if req["api"] == "local" else "")
            stopped = {"at": i, "reason": f"{type(e).__name__}: {e}"[:200] + hint}
            break
        results.append(res)
        if res["status"] >= 500:  # server error: stop instead of sending the rest into a failing server
            stopped = {"at": i, "reason": f"server error {res['status']}"}
            break
        ok = 200 <= res["status"] < 300
        if req["api"] == "local" and path in SESSION_LOCAL and ok:
            remember_session(req, res)
        if not (with_undo and ok and is_change(req)):
            continue
        if before:
            undo.insert(0, before)
        elif req["method"] == "POST" and path == "/profiles":
            data = res["response"].get("data") if isinstance(res["response"], dict) else None
            if isinstance(data, dict) and data.get("uuid"):
                created.append(data["uuid"])
        elif req["method"] == "DELETE" and path == "/profiles":
            manual.append("Deleted profiles are in the trash for 72 hours. Only the master account can restore them, in the client (no API "
                          "method): free if the profile was deleted only once in the last 7 days, otherwise 1 token.")
        elif req["method"] == "DELETE" or any(word in req["path"] for word in IRREVERSIBLE):
            manual.append(f"Cannot be undone: {req['method']} {req['path']}")
        else:
            manual.append(f"Undo manually: {req['method']} {req['path']}")
    if created:
        undo.append({"api": "public", "method": "DELETE", "path": "/profiles", "body": {"uuids": created}})
    return results, undo, manual, stopped


def journal(cmd, path, plan, results, undo, manual, check=None, stopped=None):
    """Writes the executed plan to the journal logs/actions.jsonl."""
    JOURNAL.parent.mkdir(exist_ok=True)
    lines = JOURNAL.read_text(encoding="utf-8").splitlines() if JOURNAL.exists() else []
    entry = {"id": len(lines) + 1, "time": datetime.now().isoformat(timespec="seconds"), "command": cmd,
             "plan": str(path), "summary": plan.get("summary", ""),
             "requests": [{"request": r["request"], "status": r["status"]} for r in results],
             "undo": undo, "manual": manual, "check": check, "stopped": stopped}
    with JOURNAL.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return entry["id"]


def redact(value):
    """Replaces proxy and VPN logins, passwords, keys and IP-change links with ***."""
    if isinstance(value, dict):
        return {k: "***" if SECRET.search(str(k)) and isinstance(v, (str, dict, list)) and v else redact(v)
                for k, v in value.items()}  # flags such as has_user_password stay visible
    if isinstance(value, list):
        return [redact(v) for v in value]
    if isinstance(value, str) and "privatekey" in value.lower().replace(" ", ""):  # a WireGuard config in any field
        return "***"
    return value


def save(out, results):
    """Saves the result to reports/: CSV for tables or the full JSON."""
    target = (DIR / out).resolve()
    if REPORTS.resolve() not in target.parents:
        sys.exit("The output file must be inside the reports folder.")
    note = ""
    if json.dumps(redact(results), ensure_ascii=False) != json.dumps(results, ensure_ascii=False) \
            and PRIVATE.resolve() not in target.parents:  # passwords or keys: a folder Claude is not allowed to read
        target = PRIVATE.resolve() / target.relative_to(REPORTS.resolve())
        out = target.relative_to(DIR.resolve()).as_posix()
        note = " (it has passwords or keys, so it is in reports/private: open it yourself, Claude can't)"
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.suffix.lower() != ".csv":
        target.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
        return f"Result saved to {out}{note}"
    rows = []
    for res in results:
        data = res["response"].get("data") if isinstance(res["response"], dict) else None
        items = data if isinstance(data, list) else [data] if isinstance(data, dict) else []
        rows += [x if isinstance(x, dict) else {"value": x} for x in items]
    columns = list(dict.fromkeys(k for row in rows for k in row))
    with target.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
        writer = csv.DictWriter(f, columns)
        writer.writeheader()
        for row in rows:
            writer.writerow({k: json.dumps(v, ensure_ascii=False) if isinstance(v, (dict, list)) else v
                             for k, v in row.items()})
    return f"Rows saved: {len(rows)} to {out}{note}"


def undo_plan(number):
    """Builds an undo plan for an operation from the journal."""
    entries = [json.loads(line) for line in JOURNAL.read_text(encoding="utf-8").splitlines()] if JOURNAL.exists() else []
    entries = [e for e in entries if e["command"] in ("run", "delete")]
    entry = next((e for e in entries if e["id"] == int(number)), None) if number else (entries[-1] if entries else None)
    if not entry:
        sys.exit("No such operation in the journal.")
    for note in entry["manual"]:
        print(note)
    if not entry["undo"]:
        sys.exit(f"Operation {entry['id']} has no automatic undo.")
    reqs = []
    for r in entry["undo"]:  # a big deletion is split into requests of at most MAX_DELETE profiles
        uuids = (r.get("body") or {}).get("uuids")
        if r["method"] == "DELETE" and uuids and len(uuids) > MAX_DELETE:
            reqs += [{**r, "body": {**r["body"], "uuids": uuids[i:i + MAX_DELETE]}} for i in range(0, len(uuids), MAX_DELETE)]
        else:
            reqs.append(r)
    parts, part = [], []
    for r in reqs:
        if part and (len(part) == MAX_REQUESTS or r["method"] == "DELETE" and any(x["method"] == "DELETE" for x in part)):
            parts.append(part)
            part = []
        part.append(r)
    parts.append(part)
    for i, part in enumerate(parts, 1):
        name = f"undo_{entry['id']}.json" if i == 1 else f"undo_{entry['id']}_{i}.json"
        suffix = f" (part {i} of {len(parts)})" if len(parts) > 1 else ""
        (DIR / "plans" / name).write_text(json.dumps({"summary": f"Undo operation {entry['id']}: {entry['summary']}{suffix}",
                                                     "requests": part}, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"Undo plan saved: plans/{name}{suffix}. Show it with the show command.")


def since(value):
    """Seconds since a last_active or created_at value, or None if it is empty or unreadable."""
    if isinstance(value, (int, float)) and value > 0:
        return time.time() - value / (1000 if value > 1e12 else 1)
    if isinstance(value, str) and value:
        try:
            return time.time() - datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
        except ValueError:
            return None
    return None


def audit(days, mark):
    """Read-only audit: finds profiles that need a look and prepares a plan that only marks them."""
    def get(path, params=None):
        res = call({"api": "public", "method": "GET", "path": path, "params": params})
        if res["status"] != 200:
            sys.exit(f"Audit stopped: GET {path} answered {res['status']}. Nothing was changed.")
        return res["response"] if isinstance(res["response"], dict) else {}

    profiles, page = [], 0
    while True:
        resp = get("/profiles", {"page_len": 100, "page": page, "fields": "title,tags,folder,proxy,last_active,created_at"})
        data = resp.get("data") or []
        profiles += data
        page += 1
        if not data or len(profiles) >= (resp.get("total_count") or 0):
            break
    proxies = get("/proxies").get("data") or []
    tags = get("/tags").get("data") or []
    folders = get("/folders").get("data") or []

    issues = {p["uuid"]: [] for p in profiles}
    by_proxy, by_title = {}, {}
    for p in profiles:
        proxy = p.get("proxy")
        if not proxy:
            issues[p["uuid"]].append("no proxy")
        else:
            key = proxy.get("uuid") or f"{proxy.get('host')}:{proxy.get('port')}" if isinstance(proxy, dict) else str(proxy)
            by_proxy.setdefault(key, []).append(p["uuid"])
        idle = since(p.get("last_active"))
        if idle is None and not p.get("last_active"):
            age = since(p.get("created_at"))
            if age is None or age > days * 86400:
                issues[p["uuid"]].append("never started")
        elif idle is not None and idle > days * 86400:
            issues[p["uuid"]].append(f"not started for {int(idle // 86400)} days")
        by_title.setdefault(p.get("title"), []).append(p["uuid"])
    for group in by_proxy.values():
        for u in group if len(group) > 1 else []:
            issues[u].append(f"proxy shared with {len(group) - 1} other profile(s)")
    for group in by_title.values():
        for u in group if len(group) > 1 else []:
            issues[u].append("duplicate title")

    info = {p["uuid"]: p for p in profiles}
    is_protected = lambda u: PROTECTED_TAG in {t.lower() for t in names(info[u].get("tags"))}  # noqa: E731
    flagged = [u for u, found in issues.items() if found]
    to_mark = [u for u in flagged if not is_protected(u)
               and not (mark == "tag" and AUDIT_TAG.lower() in {t.lower() for t in names(info[u].get("tags"))})
               and not (mark == "folder" and AUDIT_FOLDER in names(info[u].get("folder")))]
    used_proxies = {k for k in by_proxy}
    unused_proxies = [x for x in proxies if isinstance(x, dict) and x.get("uuid") not in used_proxies]
    same_connection = {}  # one proxy saved several times: compared here, the credentials never reach the chat
    for x in proxies:
        if isinstance(x, dict):
            key = (str(x.get("type", "")).lower(), str(x.get("host", "")).lower(), str(x.get("port", "")),
                   x.get("login") or "", x.get("password") or "", x.get("data") or x.get("config") or x.get("conf") or "")
            same_connection.setdefault(key, []).append(str(x.get("title") or x.get("uuid")))
    duplicate_proxies = [titles for titles in same_connection.values() if len(titles) > 1]
    used_tags = {t for p in profiles for t in names(p.get("tags"))}
    unused_tags = [t for t in tags if isinstance(t, dict) and t.get("name") not in used_tags]
    used_folders = {f for p in profiles for f in names(p.get("folder"))}
    empty_folders = [f for f in folders if isinstance(f, dict) and not ({f.get("name"), f.get("uuid")} & used_folders)]

    stamp = datetime.now().strftime("%Y-%m-%d_%H-%M")
    REPORTS.mkdir(exist_ok=True)
    report = REPORTS / f"audit_{stamp}.csv"
    with report.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
        writer = csv.writer(f)
        writer.writerow(["uuid", "title", "tags", "folder", "issues", "protected"])
        for u in flagged:
            p = info[u]
            writer.writerow([u, p.get("title"), ", ".join(sorted(names(p.get("tags")))), ", ".join(names(p.get("folder"))),
                             "; ".join(issues[u]), "yes" if is_protected(u) else ""])

    count = lambda word: sum(any(word in i for i in issues[u]) for u in flagged)  # noqa: E731
    print(f"Audit (read only, nothing was changed): {len(profiles)} profiles, {len(flagged)} need a look.")
    for label, word in (("no proxy", "no proxy"), ("proxy shared with other profiles", "proxy shared"),
                        (f"not started for more than {days} days", "not started"), ("never started", "never started"),
                        ("duplicate titles", "duplicate title")):
        print(f"  {label}: {count(word)}")
    print(f"Also: {len(unused_proxies)} unused proxies, {len(duplicate_proxies)} proxies saved more than once, "
          f"{len(unused_tags)} unused tags, {len(empty_folders)} empty folders.")
    for titles in duplicate_proxies:
        print(f"  Same proxy saved {len(titles)} times: {', '.join(titles)}")
    if any(is_protected(u) for u in flagged):
        print(f"Profiles with the tag '{PROTECTED_TAG}' are listed but will not be marked.")
    print(f"Report: reports/{report.name}")
    if not to_mark:
        print("Nothing to mark.")
        report_time()
        return
    reqs = []
    # tag names are case-insensitive in Octo: "audit-check" and "Audit-check" are the same tag
    tag_name = next((t.get("name") for t in tags if isinstance(t, dict) and str(t.get("name", "")).lower() == AUDIT_TAG.lower()), None)
    if mark == "tag" and not tag_name:
        reqs.append({"api": "public", "method": "POST", "path": "/tags", "body": {"name": AUDIT_TAG, "color": "orange"}})
    tag_name = tag_name or AUDIT_TAG
    if mark == "folder" and AUDIT_FOLDER not in {f.get("name") for f in folders if isinstance(f, dict)}:
        reqs.append({"api": "public", "method": "POST", "path": "/folders", "body": {"folders": [{"name": AUDIT_FOLDER}]}})
    for u in to_mark:
        body = {"tags": sorted(names(info[u].get("tags"))) + [tag_name]} if mark == "tag" else {"folder": AUDIT_FOLDER}
        reqs.append({"api": "public", "method": "PATCH", "path": f"/profiles/{u}", "body": body})
    what = f"add the tag '{AUDIT_TAG}'" if mark == "tag" else f"move to the folder '{AUDIT_FOLDER}'"
    for i in range(0, len(reqs), MAX_REQUESTS):
        name = "audit_mark.json" if i == 0 else f"audit_mark_{i // MAX_REQUESTS + 1}.json"
        (DIR / "plans" / name).write_text(json.dumps({"summary": f"Audit: {what} for profiles that need a look",
                                                     "requests": reqs[i:i + MAX_REQUESTS]}, ensure_ascii=False, indent=2),
                                          encoding="utf-8")
        print(f"Mark plan: plans/{name} ({what}). It only marks profiles, nothing is deleted. Show it with the show command.")
    report_time()


def site(address):
    """Approves a site for browser.py. Runs only after the user confirms it in Claude Code."""
    try:
        sites = json.loads(SITES.read_text(encoding="utf-8")).get("sites", []) if SITES.exists() else []
    except (ValueError, AttributeError):
        sites = []
    if not address:
        print("Approved sites: " + (", ".join(sites) if sites else "none yet") + " (example.com and iplocation.io always)")
        return
    host = (urlparse(address if "://" in address else "https://" + address).hostname or "").lower().removeprefix("www.")
    if not re.fullmatch(r"(?:[a-z0-9-]+\.)+(?:[a-z]{2,}|xn--[a-z0-9-]+)", host):
        sys.exit(f"'{address}' is not a site address, for example: python octo_api.py site example.com")
    if host not in sites:
        SITES.write_text(json.dumps({"sites": sorted([*sites, host])}, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"Approved: {host} and its subdomains. browser.py can open them now.")


def history(last, profile=None):
    """What the assistant did, newest last: plans, proxy checks and every browser.py call. Reads only.
    Warm-ups are kept apart, in the farm log (warmlog)."""
    entries = []
    for line in JOURNAL.read_text(encoding="utf-8").splitlines() if JOURNAL.exists() else []:
        try:
            entries.append(json.loads(line))
        except ValueError:
            continue
    if profile:  # one profile: its browser calls and the plans that touched it
        entries = [e for e in entries if e.get("profile") == profile or profile in json.dumps(e.get("requests"))
                   or profile in json.dumps(e.get("undo"))]
    if not entries:
        print("The journal is empty" + (f" for profile {profile}." if profile else "."))
        return
    shown = entries[-last:]
    print(f"Last {len(shown)} of {len(entries)} operation(s) in logs/actions.jsonl"
          + (f" for profile {profile}" if profile else "") + ":")
    for e in shown:
        when = str(e.get("time", ""))[5:16].replace("T", " ")
        command = e.get("command", "")
        if command == "browser":
            steps = [f"{s.get('action')} {s.get('detail') or ''}".strip()
                     + ("" if s.get("result") == "ok" else f" [{s.get('result')}]") for s in e.get("steps") or []]
            print(f"#{e.get('id')} {when}  browser    profile {str(e.get('profile'))[:8]}: {' -> '.join(steps)}")
            pages = list(dict.fromkeys(s["page"] for s in e.get("steps") or [] if s.get("page")))
            if pages:
                print(f"      pages: {', '.join(pages[:5])}")
        else:
            reqs = e.get("requests") or []
            done = sum(1 for r in reqs if 200 <= (r.get("status") or 0) < 300)
            counts = f" ({done} of {len(reqs)} requests ok)" if reqs else ""
            print(f"#{e.get('id')} {when}  {command:<10} {e.get('summary', '')}{counts}")
            check = e.get("check") or {}
            if check.get("total"):
                print(f"      verification: {check.get('ok')} of {check.get('total')} match")
        if e.get("stopped"):
            reason = e["stopped"].get("reason") if isinstance(e["stopped"], dict) else e["stopped"]
            print(f"      stopped: {reason}")


def warmlog(last, profile=None):
    """The farm log: which profiles were warmed, when, through which IP, how many cookies were added. Reads only."""
    lines = []
    for line in WARM_LOG.read_text(encoding="utf-8").splitlines() if WARM_LOG.exists() else []:
        try:
            lines.append(json.loads(line))
        except ValueError:
            continue
    if profile:
        lines = [x for x in lines if x.get("profile") == profile]
    if not lines:
        print("The farm log is empty" + (f" for profile {profile}." if profile else ": no warm-ups yet."))
        return
    if profile:  # every warm-up of one profile, newest last
        print(f"Profile {lines[-1].get('title')} ({profile}): {len(lines)} warm-up(s)")
        for x in lines[-last:]:
            print(f"  {str(x.get('time'))[:16].replace('T', ' ')}  IP {x.get('ip') or '-'} {x.get('country') or ''}  "
                  f"sites {x.get('sites_ok')}/{x.get('sites_total')}, pages {x.get('pages')}, cookies +{x.get('cookies_added')}  "
                  f"{x.get('status')}")
            bad = [f"{r.get('site')} ({r.get('status') if r.get('status') != 'ok' else r.get('consent')})"
                   for r in x.get("sites") or [] if r.get("status") != "ok" or r.get("consent") == "banner not recognized"]
            if bad:
                print(f"      needs a look: {', '.join(bad[:8])}")
        return
    by = {}
    for x in lines:  # one row per profile, across all warm-ups
        row = by.setdefault(x.get("profile"), {"title": x.get("title"), "runs": 0, "ok": 0, "pages": 0, "cookies": 0})
        row["runs"] += 1
        row["ok"] += x.get("sites_ok") or 0
        row["pages"] += x.get("pages") or 0
        row["cookies"] += x.get("cookies_added") or 0
        row.update(last=x.get("time"), status=x.get("status"), country=x.get("country") or "")
    rows = sorted(by.items(), key=lambda kv: str(kv[1]["last"]))[-last:]
    print(f"Farm log: {len(lines)} warm-up(s) of {len(by)} profile(s), {len(rows)} most recent shown.")
    print(f"{'Profile':26} {'Warm-ups':9} {'Last':17} {'Country':12} {'Sites ok':9} {'Pages':6} {'Cookies':8} Last status")
    for _, r in rows:
        print(f"{str(r['title'])[:26]:26} {r['runs']:<9} {str(r['last'])[:16].replace('T', ' '):17} {r['country'][:12]:12} "
              f"{r['ok']:<9} {r['pages']:<6} {'+' + str(r['cookies']):8} {r['status']}")
    print("One profile in detail: python octo_api.py warmlog --profile <uuid>")


# Export groups: main and proxy always (1 request per 100 profiles), the others with --add when the user wants them.
# A name ending in a dot takes every column under it, so new fields from Octo show up by themselves.
EXPORT_GROUPS = {
    "main": ("title", "description", "tags", "pinned_tag", "folder", "last_active", "created_at", "updated_at",
             "has_user_password"),
    "proxy": ("proxy.title", "proxy.type", "proxy.host", "proxy.port", "proxy.login", "proxy.password",
              "proxy.change_ip_url"),  # the profile's own proxy, not facts about the proxy in general
    "pages": ("start_pages", "bookmarks"),
    "settings": ("storage_options.", "launch_args", "local_cache", "images_load_limit", "extensions"),
    "other": ("extra_info", "extra_info.", "image", "password_set_at"),
    "fingerprint": ("fingerprint.",)}
EXPORT_LIST_FIELDS = ("title", "description", "proxy", "start_pages", "tags", "last_active", "storage_options",
                      "created_at", "updated_at", "has_user_password", "pinned_tag", "folder", "launch_args",
                      "images_load_limit", "local_cache", "extra_info")  # the list gives these for 100 profiles per request
DETAIL_ONLY = ("fingerprint", "bookmarks", "extensions", "image", "password_set_at")  # ...these only each profile's card
EXPORT_CACHE = PRIVATE / "export_cache.json"  # cards already read, with updated_at: the next export reads only changed ones
DATES = ("last_active", "created_at", "updated_at", "password_set_at")


def readable(value):
    """A date from the API as 2026-10-23 01:25:51."""
    if isinstance(value, (int, float)) and value > 0:
        return datetime.fromtimestamp(value / (1000 if value > 1e12 else 1)).strftime("%Y-%m-%d %H:%M:%S")
    return str(value or "").replace("T", " ")[:19]


def flat(value, key, row):
    """Nested API data as columns: proxy.login, fingerprint.noise.webgl. Lists of plain values become 'a, b'."""
    if isinstance(value, dict) and value:
        for sub, item in value.items():
            flat(item, f"{key}.{sub}", row)
    elif isinstance(value, list) and all(isinstance(x, dict) and set(x) <= {"name", "url"} for x in value) and value:
        row[key] = "\n".join(f"{x.get('name') or ''} - {x.get('url') or ''}" for x in value)  # bookmarks
    elif isinstance(value, list):
        row[key] = ", ".join(map(str, value)) if all(not isinstance(x, (dict, list)) for x in value) \
            else json.dumps(value, ensure_ascii=False)
    else:
        row[key] = "" if value is None or value == {} else value


def list_separator():
    """The list separator of the Windows region settings: ';' for Russian and most European Excel, ',' for English."""
    try:
        import winreg
        with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Control Panel\International") as key:
            return (winreg.QueryValueEx(key, "sList")[0] or ",")[:1]
    except Exception:  # not Windows, or no such setting
        return ","


def value(name):
    """The text after an option such as --tag, or None."""
    return args[args.index(name) + 1] if name in args and args.index(name) + 1 < len(args) else None


def filters_from():
    """The filters shared by find and export: --tag, --any-tag, --folder, --proxy, --password, dates, --sort, --limit."""
    if value("--limit") is not None and not value("--limit").isdigit():
        sys.exit("--limit takes a number, for example --limit 20")
    f = {k: value("--" + k.replace("_", "-")) for k in ("tag", "folder", "any_tag", "proxy", "password", "created_after",
                                                       "created_before", "active_after", "active_before", "idle", "sort", "limit")}
    f["never_started"] = "--never-started" in args
    return f, value("--search")


def all_profiles(fields, search=None, tag=None):
    """Every profile of the account, or those whose title starts with search or that have the tags: 100 per request."""
    params = {"page_len": 100, "fields": fields, **({"search": search} if search else {}), **({"search_tags": tag} if tag else {})}
    found, page = [], 0
    while True:
        res = call({"api": "public", "method": "GET", "path": "/profiles", "params": {**params, "page": page}})
        if res["status"] != 200:
            sys.exit(f"GET /profiles answered {res['status']}. Nothing was changed.")
        resp = res["response"] if isinstance(res["response"], dict) else {}
        data = [p for p in resp.get("data") or [] if isinstance(p, dict)]
        found, page = found + data, page + 1
        if not data or len(found) >= (resp.get("total_count") or 0):
            return found


def find(f, search=None):
    """Profiles that match the filters, as a short table with their uuids: no plan file needed. Reads only."""
    found = pick(all_profiles("title,tags,folder,proxy,last_active,created_at,has_user_password", search, f.get("tag")), f)
    if not found:
        sys.exit("No profiles match.")
    print(f"{len(found)} profile(s):\n{'uuid':32}  {'Title':28} {'Tags':24} {'Folder':16} {'Proxy':20} Last start")
    for p in found[:300]:
        px = p.get("proxy") if isinstance(p.get("proxy"), dict) else {}
        proxy = px.get("title") or (f"{px.get('host')}:{px.get('port')}" if px.get("host") else "")
        print(f"{p.get('uuid', ''):32}  {str(p.get('title') or '')[:28]:28} {', '.join(sorted(names(p.get('tags'))))[:24]:24} "
              f"{', '.join(sorted(names(p.get('folder'))))[:16]:16} {proxy[:20]:20} {readable(p.get('last_active')) or '-'}")
    if len(found) > 300:
        print(f"...and {len(found) - 300} more: narrow the filters, or save them all with export.")
    report_time()


def start_stop(cmd, name, headless=False):
    """Starts or stops one profile by its title or uuid, with no plan file. A normal stop syncs the profile's data."""
    if re.fullmatch(r"[0-9a-f]{32}", name):
        uid, title = name, name
    else:  # the exact title, any case: the API search narrows the list to titles starting with it
        same = [p for p in all_profiles("title", name) if str(p.get("title", "")).lower() == name.lower()]
        if len(same) != 1:
            sys.exit(f"No profile is titled '{name}'." if not same else
                     f"{len(same)} profiles are titled '{name}': use the uuid, one of {', '.join(p['uuid'] for p in same)}.")
        uid, title = same[0]["uuid"], same[0].get("title")
    body = {"uuid": uid} if cmd == "stop" else {"uuid": uid, "headless": headless, "debug_port": True,
                                                 "flags": ["--disable-backgrounding-occluded-windows"]}
    req = {"api": "local", "method": "POST", "path": f"/profiles/{cmd}", "body": body}
    try:
        res = call(req)
    except (requests.RequestException, OSError):
        sys.exit(f"The Octo client is not running or its local port is not {LOCAL_PORT}. Nothing was done.")
    resp = res["response"] if isinstance(res["response"], dict) else {}
    if not 200 <= res["status"] < 300:
        sys.exit(f"Could not {cmd} '{title}': {resp.get('code') or resp.get('error') or 'HTTP ' + str(res['status'])}.")
    remember_session(req, res)
    journal(cmd, "-", {"summary": f"{cmd.capitalize()} of the profile {title}"}, [res], [], [])
    conn = resp.get("connection_data") or {}
    print(f"Started '{title}' ({uid}){' without a window' if headless else ''}: IP {conn.get('ip') or '?'} "
          f"{conn.get('country') or ''}. Its address is in sessions.json for browser.py and scripts." if cmd == "start"
          else f"Stopped '{title}' ({uid}) normally: its data is synced.")
    report_time()


def moment(text, name):
    """A date for a filter: 2026-10-01, or 7d for seven days ago. Returns a Unix time."""
    text = str(text).strip().lower()
    if re.fullmatch(r"\d+d", text):
        return time.time() - int(text[:-1]) * 86400
    try:
        return datetime.strptime(text, "%Y-%m-%d").timestamp()
    except ValueError:
        sys.exit(f"{name} takes a date like 2026-10-01 or a number of days like 7d")


def when(value):
    """The Unix time of an API date (a timestamp or an ISO string), or None if it is empty."""
    ago = since(value)
    return None if ago is None else time.time() - ago


def geo_row(results):
    """The checkers' answers as report columns."""
    row = {}
    for name, *_ in GEO_SOURCES:
        r = results.get(name) if isinstance(results.get(name), dict) else {}
        row.update({f"{name} ip": r.get("ip", ""), f"{name} country": r.get("country", ""), f"{name} city": r.get("city", "")})
    return row


def geo_line(results, octo=None):
    """One line for the chat: the IP and every checker's country code, Octo first when it was asked."""
    parts = [f"Octo {octo.get('country') or '?'}"] if octo else []
    parts += [f"{name} {results[name]['cc'] or '?'}" if isinstance(results.get(name), dict) else f"{name} -" for name, *_ in GEO_SOURCES]
    ips = [r["ip"] for r in results.values() if isinstance(r, dict)]
    return (max(set(ips), key=ips.count) if ips else (octo or {}).get("ip", "")), " | ".join(parts)


def write_xlsx(path, columns, rows):
    """A real Excel file with no extra libraries: a frozen bold header, filters, column widths, numbers as numbers."""
    import zipfile
    from xml.sax.saxutils import escape
    bad = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")  # characters XML can't hold

    def letter(n):
        name = ""
        n += 1
        while n:
            n, rest = divmod(n - 1, 26)
            name = chr(65 + rest) + name
        return name

    def cell(ref, value, style):
        if isinstance(value, (int, float)) and not isinstance(value, bool):
            return f'<c r="{ref}"><v>{value}</v></c>'
        text = escape(bad.sub("", str(value))[:32767])
        return f'<c r="{ref}" t="inlineStr" s="{style}"><is><t xml:space="preserve">{text}</t></is></c>'

    widths = [min(max([len(str(c))] + [len(str(r.get(c, ""))) for r in rows[:500]]) + 2, 60) for c in columns]
    lines = ['<row r="1">' + "".join(cell(f"{letter(i)}1", c, 1) for i, c in enumerate(columns)) + "</row>"]
    for n, r in enumerate(rows, 2):
        lines.append(f'<row r="{n}">' + "".join(cell(f"{letter(i)}{n}", r[c], 0)
                                               for i, c in enumerate(columns) if r.get(c, "") != "") + "</row>")
    last = f"{letter(len(columns) - 1)}{len(rows) + 1}"
    sheet = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
             '<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
             '<sheetViews><sheetView workbookViewId="0"><pane ySplit="1" topLeftCell="A2" activePane="bottomLeft" '
             'state="frozen"/></sheetView></sheetViews><cols>'
             + "".join(f'<col min="{i + 1}" max="{i + 1}" width="{w}" customWidth="1"/>' for i, w in enumerate(widths))
             + "</cols><sheetData>" + "".join(lines) + f'</sheetData><autoFilter ref="A1:{last}"/></worksheet>')
    parts = {
        "[Content_Types].xml": '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"><Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/><Default Extension="xml" ContentType="application/xml"/><Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/><Override PartName="/xl/worksheets/sheet1.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/><Override PartName="/xl/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.styles+xml"/></Types>',
        "_rels/.rels": '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/></Relationships>',
        "xl/workbook.xml": '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets><sheet name="Profiles" sheetId="1" r:id="rId1"/></sheets>'
                           f'<definedNames><definedName name="_xlnm._FilterDatabase" localSheetId="0" hidden="1">Profiles!$A$1:${letter(len(columns) - 1)}${len(rows) + 1}</definedName></definedNames></workbook>',
        "xl/_rels/workbook.xml.rels": '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet1.xml"/><Relationship Id="rId2" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/></Relationships>',
        "xl/styles.xml": '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><fonts count="2"><font><sz val="11"/><name val="Calibri"/></font><font><b/><sz val="11"/><name val="Calibri"/></font></fonts><fills count="2"><fill><patternFill patternType="none"/></fill><fill><patternFill patternType="gray125"/></fill></fills><borders count="1"><border/></borders><cellStyleXfs count="1"><xf numFmtId="0" fontId="0" fillId="0" borderId="0"/></cellStyleXfs><cellXfs count="2"><xf numFmtId="0" fontId="0" fillId="0" borderId="0" xfId="0"/><xf numFmtId="0" fontId="1" fillId="0" borderId="0" xfId="0" applyFont="1"/></cellXfs><cellStyles count="1"><cellStyle name="Normal" xfId="0" builtinId="0"/></cellStyles></styleSheet>',
        "xl/worksheets/sheet1.xml": sheet}
    with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as z:
        for name, text in parts.items():
            z.writestr(name, text)


def pick(profiles, f):
    """Applies the export filters. Every filter must match; lists inside one filter mean any of them."""
    split = lambda v: [x.strip().lower() for x in str(v).split(",") if x.strip()]  # noqa: E731
    out = []
    for p in profiles:
        tags = {t.lower() for t in names(p.get("tags"))}
        folders = {n.lower() for n in names(p.get("folder"))}
        proxy = p.get("proxy") if isinstance(p.get("proxy"), dict) else {}
        made, active = when(p.get("created_at")), when(p.get("last_active"))
        checks = [
            not f.get("tag") or set(split(f["tag"])) <= tags,
            not f.get("any_tag") or bool(set(split(f["any_tag"])) & tags),
            not f.get("folder") or any((x == "none" and not folders) or x in folders for x in split(f["folder"])),
            not f.get("proxy") or any((x == "none" and not proxy) or (proxy and x in f"{proxy.get('title') or ''} "
                                      f"{proxy.get('host') or ''}".lower()) for x in split(f["proxy"])),
            not f.get("password") or bool(p.get("has_user_password")) == (str(f["password"]).lower() in ("yes", "true", "1")),
            not f.get("created_after") or (made is not None and made >= moment(f["created_after"], "--created-after")),
            not f.get("created_before") or (made is not None and made < moment(f["created_before"], "--created-before")),
            not f.get("active_after") or (active is not None and active >= moment(f["active_after"], "--active-after")),
            not f.get("active_before") or (active is not None and active < moment(f["active_before"], "--active-before")),
            not f.get("idle") or active is None or active < moment(f["idle"], "--idle"),
            not f.get("never_started") or active is None]
        if all(checks):
            out.append(p)
    order = (f.get("sort") or "").lower()
    if order:
        key = order.lstrip("-")
        if key not in ("created", "active", "title"):
            sys.exit("--sort takes created, active or title, with - for newest or Z-A first: --sort -created")
        value = {"created": lambda p: when(p.get("created_at")) or 0, "active": lambda p: when(p.get("last_active")) or 0,
                 "title": lambda p: str(p.get("title") or "").lower()}[key]
        out.sort(key=value, reverse=order.startswith("-"))
    if f.get("limit"):
        out = out[:int(f["limit"])]
    return out


def export(tag=None, folder=None, search=None, add="", exclude="", fmt="xlsx", sep=None, target=None, filters=None):
    """Exports profiles: main data and the profile's proxy with login and password, more groups with --add. Reads only."""
    extra = [g.strip() for g in add.split(",") if g.strip()]
    extra = [g for g in EXPORT_GROUPS if g not in ("main", "proxy")] if "all" in extra else extra
    unknown = [g for g in extra if g not in EXPORT_GROUPS]
    if unknown:
        sys.exit(f"Unknown group(s) {', '.join(unknown)}: --add takes {', '.join(g for g in EXPORT_GROUPS if g not in ('main', 'proxy'))} or all")
    groups = ["main", "proxy", *[g for g in EXPORT_GROUPS if g in extra and g not in ("main", "proxy")]]
    patterns = [c for g in groups for c in EXPORT_GROUPS[g]]
    excluded = {e.strip() for e in exclude.split(",") if e.strip()}
    dropped = lambda column: (not any(column == c or (c.endswith(".") and column.startswith(c)) for c in patterns)
                              and column != "uuid") or any(column == e or column.startswith(e + ".") for e in excluded)  # noqa: E731
    params = {"page_len": 100, "fields": ",".join(EXPORT_LIST_FIELDS)}  # every field costs the same: 1 request per 100
    if tag:
        params["search_tags"] = tag
    if search:
        params["search"] = search
    profiles, page = [], 0
    while True:
        res = call({"api": "public", "method": "GET", "path": "/profiles", "params": {**params, "page": page}})
        if res["status"] != 200:
            sys.exit(f"Export stopped: GET /profiles answered {res['status']}. Nothing was saved.")
        resp = res["response"] if isinstance(res["response"], dict) else {}
        data = [p for p in resp.get("data") or [] if isinstance(p, dict)]
        profiles += data
        page += 1
        if not data or len(profiles) >= (resp.get("total_count") or 0):
            break
    total = len(profiles)
    filters = {**(filters or {}), "tag": tag, "folder": folder}
    profiles = pick(profiles, filters)
    used = {k: v for k, v in filters.items() if v}
    if used:
        print(f"Filters: {', '.join(f'{k}={v}' for k, v in used.items())}: {len(profiles)} of {total} profile(s).")
    if not profiles:
        sys.exit("No profiles match, nothing was saved.")
    wanted = [f for f in DETAIL_ONLY if any(f == c.rstrip(".") for c in patterns) and f not in excluded]
    try:
        cache = json.loads(EXPORT_CACHE.read_text(encoding="utf-8")) if wanted and EXPORT_CACHE.exists() else {}
    except ValueError:
        cache = {}
    todo = [p for p in profiles if wanted and not (p.get("updated_at") and cache.get(p["uuid"], {}).get("updated_at") == p["updated_at"])]
    print(f"Exporting {len(profiles)} profile(s): {page} list request(s)"
          + (f", {len(todo)} profile card(s) to read for {', '.join(wanted)}, {len(profiles) - len(todo)} from the saved copy"
             if wanted else "") + ".")
    stopped = None
    try:
        for n, p in enumerate(todo, 1):
            try:
                res = call({"api": "public", "method": "GET", "path": f"/profiles/{p['uuid']}"})
            except LimitStop as e:
                stopped = f"{e} {len(todo) - n + 1} card(s) were not read: run the same export then, it continues from there."
                break
            card = res["response"].get("data") if isinstance(res["response"], dict) else None
            if res["status"] == 200 and isinstance(card, dict):
                cache[p["uuid"]] = {"updated_at": p.get("updated_at"), "data": {k: card.get(k) for k in DETAIL_ONLY}}
            if n % 50 == 0:
                print(f"  {n} of {len(todo)} cards read")
    finally:  # even a stopped export keeps the cards it read, so the next run continues
        if wanted and not used and not search:  # the whole account was listed: forget cards of deleted profiles
            alive_ids = {p["uuid"] for p in profiles}
            cache = {u: v for u, v in cache.items() if u in alive_ids}
        if wanted:
            PRIVATE.mkdir(parents=True, exist_ok=True)
            EXPORT_CACHE.write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8")
    rows = []
    for p in profiles:
        row = {}
        for key, value in {**p, **((cache.get(p["uuid"]) or {}).get("data") or {})}.items():
            if key in ("tags", "pinned_tag", "folder"):  # names, as in the client
                row[key] = ", ".join(sorted(names(value)))
            elif key in DATES:
                row[key] = readable(value)
            else:
                flat(value, key, row)
        rows.append({k: v for k, v in row.items() if not dropped(k)})
    fixed = [c for g in ("main", "proxy") for c in EXPORT_GROUPS[g] if not dropped(c)]  # the same columns every time
    seen = list(dict.fromkeys(fixed + [k for r in rows for k in r if r[k] != ""]))  # other columns only when filled
    order = [c for g in groups for c in EXPORT_GROUPS[g]] + ["uuid"]  # group by group, the uuid last
    place = lambda c: next(i for i, p in enumerate(order) if c == p or (p.endswith(".") and c.startswith(p)))  # noqa: E731
    columns = sorted(seen, key=lambda c: (place(c), seen.index(c)))
    stamp = datetime.now().strftime("%Y-%m-%d_%H-%M")
    path = (DIR / (target or f"reports/profiles_{stamp}.{fmt}")).resolve()
    if REPORTS.resolve() not in path.parents:
        sys.exit("The output file must be inside the reports folder.")
    secret = json.dumps(redact(rows), ensure_ascii=False) != json.dumps(rows, ensure_ascii=False)
    if secret and PRIVATE.resolve() not in path.parents:  # logins and passwords: a folder Claude is not allowed to read
        path = PRIVATE.resolve() / path.relative_to(REPORTS.resolve())
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.suffix.lower() == ".csv":
        with path.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
            writer = csv.DictWriter(f, columns, delimiter=sep or list_separator(), restval="")
            writer.writeheader()
            writer.writerows(rows)
    else:
        write_xlsx(path, columns, rows)
    shown = path.relative_to(DIR.resolve()).as_posix()
    print(f"Saved {len(rows)} profile(s), {len(columns)} column(s): {shown}"
          + (". It has proxy logins or passwords, so it is in reports/private: open it yourself, Claude can't." if secret else "."))
    print(f"Groups: {', '.join(groups)}." + ("" if len(groups) == len(EXPORT_GROUPS) else
          " More with --add: " + ", ".join(g for g in EXPORT_GROUPS if g not in groups) + "."))
    if excluded:
        print(f"Left out: {', '.join(sorted(excluded))}.")
    if stopped:
        print(f"Not complete: {stopped}")
    report_time()


PROXY_TYPES = ("http", "https", "socks", "socks5", "ssh")
PROXY_COLUMNS = {"type": "type", "protocol": "type", "host": "host", "ip": "host", "port": "port", "login": "login",
                 "user": "login", "username": "login", "password": "password", "pass": "password", "title": "title",
                 "name": "title", "change_ip_url": "change_ip_url", "proxy_url": "url", "url": "url", "proxy": "url"}


def proxy_line(line, default_type):
    """One proxy in a common format: protocol;host;port;login;password;title;change_ip_url (Octo's own),
    host:port:login:password, login:password@host:port, socks5://login:password@host:port or host:port.
    Returns (fields, None) or (None, reason); the reason never repeats the line, which may hold a password."""
    if ";" in line:
        keys = ("type", "host", "port", "login", "password", "title", "change_ip_url")
        fields = {k: v.strip() for k, v in zip(keys, line.split(";")) if v.strip()}
    elif "@" in line or "://" in line:
        scheme, _, rest = line.rpartition("://")
        creds, _, address = rest.rpartition("@")
        host, _, port = address.rpartition(":")
        login, _, password = creds.partition(":")
        fields = {k: v for k, v in (("type", scheme), ("host", host), ("port", port), ("login", login),
                                    ("password", password)) if v}
    else:
        parts = line.split(":")
        if len(parts) not in (2, 4) and len(parts) < 5:  # host:port or host:port:login:password (a ':' may be in it)
            return None, "unknown format"
        fields = {"host": parts[0], "port": parts[1]}
        if len(parts) >= 4:
            fields.update(login=parts[2], password=":".join(parts[3:]))
    return proxy_fields(fields, default_type)


def proxy_fields(fields, default_type):
    """Checks one proxy the way POST /proxies does, so a bad line costs no request."""
    fields = {k: str(v).strip() for k, v in fields.items() if str(v or "").strip()}
    fields["type"] = {"socks4": "socks"}.get(fields.get("type", default_type).lower(), fields.get("type", default_type).lower())
    if fields["type"] not in PROXY_TYPES:
        return None, f"type must be one of {', '.join(PROXY_TYPES)}"
    if not fields.get("host") or len(fields["host"]) > 253:
        return None, "no host"
    if not fields.get("port", "").isdigit() or not 0 < int(fields["port"]) < 65536:
        return None, "the port must be a number from 1 to 65535"
    fields["port"] = int(fields["port"])
    if any(len(fields.get(k, "")) > 255 for k in ("login", "password", "title")):
        return None, "login, password or title longer than 255 characters"
    return fields, None


# IP checkers asked through the proxy itself, all free and without a key: the sources iplocation.io shows, plus MaxMind
# through GeoJS, which serves MaxMind's GeoLite data. Each answers "who am I" for the address it sees.
# The second address asks about a given IP: used when a shared proxy's IP has used up a checker's free limit.
GEO_SOURCES = (
    ("IP2Location", "https://api.ip2location.io/", "https://api.ip2location.io/?ip={ip}",
     lambda j: (j.get("ip"), j.get("country_code"), j.get("country_name"), j.get("city_name"))),
    ("IPinfo", "https://ipinfo.io/json", "https://ipinfo.io/{ip}/json", lambda j: (j.get("ip"), j.get("country"), "", j.get("city"))),
    ("DB-IP", "https://api.db-ip.com/v2/free/self", "https://api.db-ip.com/v2/free/{ip}",
     lambda j: (j.get("ipAddress"), j.get("countryCode"), j.get("countryName"), j.get("city"))),
    ("MaxMind", "https://get.geojs.io/v1/ip/geo.json", "https://get.geojs.io/v1/ip/geo/{ip}.json",
     lambda j: (j.get("ip"), j.get("country_code"), j.get("country"), j.get("city"))))
GEO_CREDITS = ("IP2Location.io, IPinfo, IP Geolocation by DB-IP (db-ip.com), "
               "GeoLite data created by MaxMind (maxmind.com) through GeoJS")
CURL_SCHEMES = {"http": "http", "https": "https", "socks": "socks4a", "socks5": "socks5h"}
CURL_ERRORS = {5: "the proxy host was not found", 6: "the address of the IP service was not found through the proxy",
               7: "the proxy refused the connection", 28: "timeout: no answer in 20 s", 35: "TLS error through the proxy",
               52: "empty answer through the proxy", 56: "the proxy closed the connection",
               97: "the proxy refused the connection or the login and password"}


def through(px, url):
    """One request through a proxy (or straight from this computer when px is None) with curl, which is built into
    Windows 10 and 11. The proxy login and password reach curl on its standard input, never on the command line.
    Returns (HTTP status, seconds, answer) or (None, None, reason)."""
    q = lambda v: '"' + str(v).replace("\\", "\\\\").replace('"', '\\"') + '"'  # noqa: E731
    lines = [f"url = {q(url)}", "max-time = 20", "silent", "show-error", 'write-out = "\\n%{http_code} %{time_total}"']
    if px:
        lines.insert(0, f"proxy = {q(CURL_SCHEMES[px['type']] + '://' + str(px['host']) + ':' + str(px['port']))}")
    if px and px.get("login"):
        lines.append(f"proxy-user = {q(str(px['login']) + ':' + str(px.get('password') or ''))}")
    curl = os.environ.get("OCTO_CURL") or ("curl.exe" if os.name == "nt" else "curl")
    try:
        r = subprocess.run([*([sys.executable] if curl.endswith(".py") else []), curl, "-K", "-"], input="\n".join(lines),
                           capture_output=True, encoding="utf-8", errors="replace", timeout=40)
    except FileNotFoundError:
        return None, None, "curl was not found: it comes with Windows 10 and 11, update Windows or install curl"
    except subprocess.TimeoutExpired:
        return None, None, "no answer in 40 s"
    if r.returncode:  # the reason, then curl's own words without its "curl: (97)" prefix; they never hold the password
        err = (r.stderr or "").strip()
        detail = re.sub(r"^curl: \(\d+\)\s*", "", err.splitlines()[-1])[:100] if err else ""
        refused = "407" in err or "auth" in err.lower() or "rejected" in err.lower()
        reason = "the proxy refused the login or password" if refused else CURL_ERRORS.get(r.returncode, f"curl error {r.returncode}")
        return None, None, reason + (f" ({detail})" if detail else "")
    answer, _, tail = r.stdout.rpartition("\n")
    code, _, seconds = tail.partition(" ")
    return (int(code) if code.isdigit() else 0), float(seconds or 0), answer


def ask(px, source, ip=None):
    """One checker through the proxy, or about a given IP from this computer: ({ip, cc, country, city}, seconds, None),
    (None, seconds, why) when the checker gave no data, or (None, None, why) when the proxy itself did not work."""
    name, url, by_ip, read = source
    code, seconds, answer = through(px, by_ip.format(ip=ip) if ip else url)
    if code is None:
        return None, None, answer
    try:
        j = json.loads(answer)
        ip, cc, country_name, city = read(j)
    except (ValueError, AttributeError, TypeError):
        j, ip = {}, None
    if code != 200 or not ip:  # the checker's own reason, such as DB-IP's errorCode, which comes with HTTP 200
        err = (j.get("errorCode") or j.get("error")) if isinstance(j, dict) else None
        err = (err.get("error_message") or err.get("title") or err.get("message")) if isinstance(err, dict) else err
        return None, seconds, f"{name} gave no data: " + str(err or f"HTTP {code}")[:80] + (
            ", its free limit may be used up" if code in (403, 429) else "")
    return {"ip": ip, "cc": str(cc or "").upper(), "country": country_name or cc or "", "city": city or ""}, seconds, None


def geo_check(px):
    """Asks every checker through the proxy: GeoJS first, which has no limit, so a dead proxy fails once, then the other
    three at once. A checker that gives no data through the proxy (a shared IP may have used up its free limit) is asked
    about the same IP from this computer. Returns (results by checker, milliseconds, None) or ({}, "", why)."""
    from concurrent.futures import ThreadPoolExecutor
    full = {**{k: px.get(k) for k in ("type", "host", "port", "login", "password")}, **(px.get("inline") or {})}
    if full.get("type") not in CURL_SCHEMES:
        return {}, "", ("not checked: an SSH proxy, check it with --octo" if full.get("type") == "ssh"
                        else f"not checked: unknown proxy type {full.get('type')!r}")
    first = next(s for s in GEO_SOURCES if s[0] == "MaxMind")
    found, seconds, why = ask(full, first)
    if seconds is None:
        return {}, "", f"does not work: {why}"
    results = {first[0]: found or why}
    others = [s for s in GEO_SOURCES if s is not first]
    with ThreadPoolExecutor(max_workers=len(others)) as pool:
        for source, (got, _, why) in zip(others, pool.map(lambda s: ask(full, s), others)):
            results[source[0]] = got or why
    ips = [r["ip"] for r in results.values() if isinstance(r, dict)]
    for source in GEO_SOURCES if ips else ():
        if isinstance(results[source[0]], str):
            got, _, _ = ask(None, source, max(set(ips), key=ips.count))
            if got:
                results[source[0]] = {**got, "note": f"{source[0]} was asked by IP from this computer ({results[source[0]]})"}
    return {name: results[name] for name, *_ in GEO_SOURCES}, round(seconds * 1000), None


def verdict(results, octo=None):
    """'ok' when Octo and every checker agree, otherwise what differs: the country, the IP, or a checker that gave no data."""
    seen = {n: r for n, r in results.items() if isinstance(r, dict)}
    if octo and (octo.get("ip") or octo.get("country")):
        names = {country(r["country"]): r["cc"] for r in seen.values() if r["country"] and r["cc"]}
        seen = {"Octo": {**octo, "cc": octo.get("cc") or names.get(country(octo.get("country")), str(octo.get("country")))}, **seen}
    notes = []
    codes = [r["cc"] for r in seen.values() if r.get("cc")]
    if codes:
        common = max(set(codes), key=codes.count)
        odd = [f"{n} says {r['cc']}" for n, r in seen.items() if r.get("cc") and r["cc"] != common]
        if odd:
            notes.append("the country differs: " + ", ".join(odd))
    if len({r["ip"] for r in seen.values() if r.get("ip")}) > 1:
        notes.append("the IP is not the same everywhere: a rotating proxy changes it between requests")
    notes += [why for why in results.values() if isinstance(why, str)]
    notes += [r["note"] for r in results.values() if isinstance(r, dict) and r.get("note")]
    return "ok" + ("; " + "; ".join(notes) if notes else "")


def proxycheck(only, limit, octo=False, chosen=None, failed=False):
    """Fast proxy check: IP2Location, IPinfo, DB-IP and MaxMind are asked through every proxy, ten proxies at a time.
    No profile is started and nothing in the account changes. --octo adds what Octo itself shows, more slowly."""
    if octo:
        return proxycheck_octo(only, limit, chosen, failed)
    from concurrent.futures import ThreadPoolExecutor
    proxies, users = proxy_list(only, limit, chosen)
    print(f"Checking {len(proxies)} proxies through IP2Location, IPinfo, DB-IP and MaxMind, ten at a time. "
          "No profile is started and nothing in the account changes.")

    def one(px):
        results, ms, why = geo_check(px)
        return {"proxy": px.get("title") or "", "type": px.get("type") or (px.get("inline") or {}).get("type") or "",
                "used_by": ", ".join(users.get(px["uuid"], [])), **geo_row(results), "ms": ms,
                "status": why or verdict(results)}, results

    with ThreadPoolExecutor(max_workers=10) as pool:
        done = list(pool.map(one, proxies))
    stamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")  # seconds: checks in the same minute don't overwrite each other
    REPORTS.mkdir(exist_ok=True)
    report = REPORTS / f"proxycheck_{stamp}.csv"
    with report.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
        writer = csv.DictWriter(f, list(done[0][0]))
        writer.writeheader()
        writer.writerows(row for row, _ in done)
    print(f"{'Proxy':28} {'IP':16} {'Countries':50} {'ms':>5}  Status")
    for row, results in done:
        ip, line = geo_line(results)
        print(f"{row['proxy']:28} {ip[:16]:16} {line:50} {str(row['ms']):>5}  {row['status']}"
              + (f"  [{row['used_by'][:90]}]" if row["used_by"] else ""))
    working = sum(1 for row, _ in done if row["status"].startswith("ok"))
    doubtful = [proxy_mark(px) for px, (row, _) in zip(proxies, done) if not row["status"].startswith("ok")]
    (REPORTS / "proxycheck_failed.json").write_text(json.dumps(doubtful), encoding="utf-8")
    if doubtful:
        print(f"{len(doubtful)} proxies did not answer or can't be checked this way. Before calling them dead, check them "
              "through Octo: python octo_api.py proxycheck --octo --failed")
    print(f"Working: {working} of {len(done)} proxies, used by {sum(len(users.get(px['uuid'], [])) for px in proxies)} "
          f"profile(s). Report: reports/{report.name}. Data: {GEO_CREDITS}.")
    journal("proxycheck", "-", {"summary": f"Fast proxy check of {len(done)} proxies: {working} working"}, [], [], [])
    report_time()


def proxyimport(prefix=None, default_type="http", only=None, geo=False):
    """Adds proxies from the files in proxies_in. Only this command reads them: the logins and passwords
    go straight to the Octo API and never reach the chat."""
    files = sorted(f for f in PROXIES_IN.glob("*") if f.suffix.lower() in (".txt", ".csv")) if PROXIES_IN.exists() else []
    if only:  # one file of the folder, by its name only
        files = [f for f in files if f.name.lower() == only.lower()]
    if not files and only:
        sys.exit(f"No file {only} in the proxies_in folder. Nothing was added.")
    if not files:
        sys.exit("No .txt or .csv file in the proxies_in folder: put a file with proxies there first. Nothing was added.")
    found, skipped = [], []
    for f in files:
        lines = f.read_bytes().decode("utf-8-sig", errors="replace").splitlines()
        head = next((ln for ln in lines if ln.strip() and not ln.lstrip().startswith("#")), "")
        sep = max((";", ",", "\t"), key=head.count)
        names = [PROXY_COLUMNS.get(c.strip().strip('"').lower()) for c in head.split(sep)]
        if {"host", "url"} & set(names):  # a table with a header row, such as the Octo client's proxy export
            for n, row in enumerate(csv.reader(lines, delimiter=sep), 1):
                if n <= lines.index(head) + 1 or not any(c.strip() for c in row):
                    continue
                cells = {k: v for k, v in zip(names, row) if k}
                if cells.get("url"):  # socks5://login:password@host:port in one column
                    parsed, reason = proxy_line(cells.pop("url").strip(), default_type)
                    cells = {**(parsed or {}), **{k: v for k, v in cells.items() if v}} if parsed else cells
                    if not parsed:
                        skipped.append(f"{f.name} line {n}: {reason}")
                        continue
                fields, reason = proxy_fields(cells, default_type)
                if fields:
                    found.append((f.name, n, fields))
                else:
                    skipped.append(f"{f.name} line {n}: {reason}")
            continue
        for n, line in enumerate(lines, 1):
            if not line.strip() or line.lstrip().startswith("#"):
                continue
            fields, reason = proxy_line(line.strip(), default_type)
            if fields:
                found.append((f.name, n, fields))
            else:
                skipped.append(f"{f.name} line {n}: {reason}")
    res = call({"api": "public", "method": "GET", "path": "/proxies"})
    existing = res["response"].get("data") if isinstance(res["response"], dict) and res["status"] == 200 else None
    if not isinstance(existing, list):
        sys.exit(f"Could not read the saved proxies: HTTP {res['status']}. Nothing was added.")
    key = lambda x: (str(x.get("type", "")).lower(), str(x.get("host", "")).lower(), str(x.get("port", "")),  # noqa: E731
                     x.get("login") or "", x.get("password") or "")
    known = {key(x) for x in existing if isinstance(x, dict)}
    used = {str(x.get("title")) for x in existing if isinstance(x, dict)}
    number, new, repeats = 0, [], 0
    fresh = []
    for name, n, fields in found:
        if key(fields) in known:
            repeats += 1
            continue
        known.add(key(fields))
        fresh.append(fields)
    countries = {}
    if geo and fresh:  # the four checkers through every new proxy, ten at a time, before anything is added
        from concurrent.futures import ThreadPoolExecutor
        print(f"Checking the country of {len(fresh)} new proxies through IP2Location, IPinfo, DB-IP and MaxMind...")
        with ThreadPoolExecutor(max_workers=10) as pool:
            for fields, (results, _, why) in zip(fresh, pool.map(geo_check, fresh)):
                codes = [r["cc"] for r in results.values() if isinstance(r, dict) and r.get("cc")]
                countries[id(fields)] = ((max(set(codes), key=codes.count) if codes else "DEAD" if why and why.startswith("does not work")
                                          else "CHECK"), why or verdict(results))
    for fields in fresh:
        if geo:  # DE-1, US-2...: the numbering goes on after the titles already in the client
            label = countries[id(fields)][0]
            number = 1
            while f"{label}-{number}" in used:
                number += 1
            fields["title"] = f"{label}-{number}"
        elif not fields.get("title"):
            if prefix:
                number += 1
                while f"{prefix}-{number}" in used:
                    number += 1
                fields["title"] = f"{prefix}-{number}"
            else:
                fields["title"] = f"{fields['host']}:{fields['port']}"
        used.add(fields["title"])
        new.append(fields)
    if geo:
        for fields in new:
            label, note = countries[id(fields)]
            print(f"  {fields['title']}: {fields['type']} {fields['host']}:{fields['port']} - {note}")
    print(f"Found {len(found)} proxies in {len(files)} file(s) of proxies_in: {len(new)} new, {repeats} already saved or repeated, "
          f"{len(skipped)} line(s) not understood.")
    for line in skipped[:20]:
        print(f"  skipped {line}")
    results, undo = [], []
    for fields in new:
        try:
            res = call({"api": "public", "method": "POST", "path": "/proxies", "body": fields})
        except LimitStop as e:
            print(f"Stopped: {e} The rest is still in proxies_in: run the import again later, added proxies are skipped.")
            break
        results.append(res)
        made = res["response"].get("data") if isinstance(res["response"], dict) else None
        where = f"{fields['type']} {fields['host']}:{fields['port']}"
        if 200 <= res["status"] < 300 and isinstance(made, dict) and made.get("uuid"):
            undo.append({"api": "public", "method": "DELETE", "path": f"/proxies/{made['uuid']}"})
            print(f"  added {fields['title']}: {where}")
        else:
            code = res["response"].get("code") if isinstance(res["response"], dict) else ""
            print(f"  not added {fields['title']}: {where}, HTTP {res['status']} {code or ''}".rstrip())
    number = journal("run", "proxyimport", {"summary": f"Added {len(undo)} proxies from proxies_in"}, results, undo,
                     ["Deleting proxies can't be undone: they have no trash."])
    print(f"Added {len(undo)} proxies. Operation {number} saved to the journal" + (f", undo: python octo_api.py undo {number}." if undo else "."))
    report_time()


def report_time():
    left = STATS.get("left")  # from the last cloud answer: local calls carry no limit headers
    print(f"Done in {time.time() - STARTED:.1f} s, {STATS['requests']} request(s) counted by Octo's rate limits."
          + (f" Left: {left[0]}{' of ' + left[1] if left[1] else ''} this minute, "
             f"{left[2]}{' of ' + left[3] if left[3] else ''} this hour." if left else ""))


def proxy_mark(px):
    """A proxy's mark for proxycheck_failed.json: a hash, so neither its address nor its password is kept."""
    return hashlib.sha256(str(px["uuid"]).encode()).hexdigest()[:16]


def proxy_list(only, limit, chosen=None, failed=False):
    """Every proxy to check: the Proxies list and the ones set right in profiles, and which profiles use each.
    chosen ("Title 1,Title 2" or uuids) keeps only the proxies of those profiles, found by a title search each."""
    def listed(res):
        return res["response"].get("data") if isinstance(res["response"], dict) else None

    proxies = [x for x in listed(call({"api": "public", "method": "GET", "path": "/proxies"})) or [] if isinstance(x, dict)]
    users, temporary, missing = {}, {}, {}
    names = [n.strip() for n in (chosen or "").split(",") if n.strip()]
    hit = lambda p, n: n == p.get("uuid") or n.lower() == str(p.get("title", "")).lower()  # noqa: E731
    rows = [p for n in names for p in all_profiles("title,proxy", None if re.fullmatch(r"[0-9a-f]{32}", n) else n)
            if hit(p, n)] if names else all_profiles("title,proxy")
    unknown = [n for n in names if not any(hit(p, n) for p in rows)]
    if unknown:
        sys.exit(f"No profile titled {', '.join(unknown)}. Nothing was checked.")
    bare = []  # chosen profiles without a proxy
    for p in rows:
        px = p.get("proxy")
        if not isinstance(px, dict) or not (px.get("uuid") or px.get("host")) or p.get("title") == CHECK_PROFILE:
            bare += [p.get("title") or p["uuid"]] if names else []
            continue
        if px.get("uuid"):  # a saved proxy: one the Proxies list did not return is checked from the profile's data
            users.setdefault(px["uuid"], []).append(p.get("title") or p["uuid"])
            missing.setdefault(px["uuid"], {**px, "title": px.get("title") or f"{px.get('type')} {px.get('host')}:{px.get('port')}",
                                            "auto": not px.get("title"), "inline": {  # copied, not referenced
                k: px.get(k) for k in ("type", "host", "port", "login", "password")} if px.get("host") else None})
            continue
        # a temporary proxy set right in the profile: the same connection in many profiles is checked once
        key = "temporary " + "|".join(str(px.get(k) or "") for k in ("type", "host", "port", "login", "password"))
        temporary.setdefault(key, {"uuid": key, "title": f"{px.get('type') or 'proxy'} {px.get('host')}:{px.get('port')}",
                                   "host": px.get("host"), "auto": True, "inline": {k: px.get(k) for k in ("type", "host", "port",
                                                                                              "login", "password")}})
        users.setdefault(key, []).append(p.get("title") or p["uuid"])
    for key, x in [*missing.items(), *temporary.items()]:  # no title of its own: named after the profiles that use it
        who = users.get(key, [])
        if x.get("auto") and who:
            x["title"] += f" ({who[0]}" + (f" +{len(who) - 1})" if len(who) > 1 else ")")
    extra = [x for u, x in missing.items() if u not in {y.get("uuid") for y in proxies}]
    if extra:
        print(f"{len(extra)} proxies used by profiles are not in the Proxies list, for example ones bought in Octo's "
              "Proxy Shop, which has no API: they are checked with the data from the profiles.")
    proxies += extra + list(temporary.values())  # saved proxies and the ones set right in profiles, both checked the same way
    if failed:  # only the ones the last fast check could not confirm
        try:
            marks = set(json.loads((REPORTS / "proxycheck_failed.json").read_text(encoding="utf-8")))
        except (OSError, ValueError):
            sys.exit("Run the fast check first: python octo_api.py proxycheck")
        proxies = [x for x in proxies if proxy_mark(x) in marks]
    if names:  # only the chosen profiles' proxies, each once
        proxies = [x for x in proxies if x["uuid"] in users]
        if bare:
            print(f"No proxy: {', '.join(bare)}.")
    if only:
        proxies = [x for x in proxies if only.lower() in f"{x.get('title') or ''} {x.get('host') or ''}".lower()
                   or x.get("uuid") == only]
    proxies = proxies[:limit]
    if not proxies:
        sys.exit("The last fast check confirmed every proxy: nothing to re-check." if failed else
                 "The chosen profiles have no proxy. Nothing was checked." if names else
                 "No proxies to check: neither in the Proxies list nor set right in profiles.")
    return proxies, users


def proxycheck_octo(only, limit, chosen=None, failed=False):
    """The slow check through Octo itself, in the dedicated empty profile check-proxy. Working profiles are never started."""
    def ok(res):
        return 200 <= res["status"] < 300

    def data(res):
        return res["response"].get("data") if isinstance(res["response"], dict) else None

    try:  # the Octo client must be open: checked before anything is changed (this request is free)
        active = call({"api": "local", "method": "GET", "path": "/profiles/active"})["response"]
    except (requests.RequestException, OSError):
        sys.exit(f"The Octo client is not running or its local port is not {LOCAL_PORT}. "
                 "Open the client and run the check again. Nothing was changed.")
    running = {x.get("uuid") for x in (active if isinstance(active, list) else []) if isinstance(x, dict)}
    found = call({"api": "public", "method": "GET", "path": "/profiles",
                  "params": {"page_len": 100, "page": 0, "search": CHECK_PROFILE, "fields": "title,tags"}})
    check = next((p for p in data(found) or [] if isinstance(p, dict) and p.get("title") == CHECK_PROFILE), None)
    if check and PROTECTED_TAG in {t.lower() for t in names(check.get("tags"))}:
        sys.exit(f"The profile '{CHECK_PROFILE}' has the tag '{PROTECTED_TAG}': it can't be used for proxy checks.")
    if not check:
        off = dict.fromkeys(("cookies", "passwords", "extensions", "localstorage", "history", "bookmarks", "serviceworkers"), False)
        res = call({"api": "public", "method": "POST", "path": "/profiles", "body": {
            "title": CHECK_PROFILE, "fingerprint": {"os": "win"}, "storage_options": off,
            "description": "Used only to test proxies by the assistant. Keep it empty."}})
        if not ok(res) or not isinstance(data(res), dict):
            sys.exit(f"Could not create the profile '{CHECK_PROFILE}': HTTP {res['status']}.")
        check = data(res)
        print(f"Created the empty profile '{CHECK_PROFILE}': it is used only to test proxies.")
    uid = check["uuid"]
    if uid in running:  # left open by an earlier run: every start would fail and look like a dead proxy
        stop = {"api": "local", "method": "POST", "path": "/profiles/stop", "body": {"uuid": uid}}
        if not ok(call(stop)):
            sys.exit(f"The profile '{CHECK_PROFILE}' is running and did not stop: close it in the client and run again.")
        remember_session(stop, {"status": 200, "response": {}})
        print(f"The profile '{CHECK_PROFILE}' was still running from an earlier check: stopped it first.")

    proxies, users = proxy_list(only, limit, chosen, failed)
    print(f"Checking {len(proxies)} proxies in the profile '{CHECK_PROFILE}' (about {3 * len(proxies)} requests). "
          "Working profiles are not started.")

    stamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    table, lines = [], []
    for px in proxies:
        row = {"proxy": px.get("title") or "", "proxy_uuid": "set in the profile" if px.get("inline") else px["uuid"],
               "used_by": ", ".join(users.get(px["uuid"], [])), "octo_ip": "", "octo_country": "", **geo_row({}), "status": ""}
        try:  # a profile's own proxy is copied into check-proxy as its temporary proxy; nothing of it reaches the chat
            res = call({"api": "public", "method": "PATCH", "path": f"/profiles/{uid}",
                        "body": {"proxy": px.get("inline") or {"uuid": px["uuid"]}}})
        except LimitStop as e:  # the report of the proxies already checked is still written below
            row["status"] = f"not checked: {e}"
            table.append(row)
            break
        if not ok(res):
            row["status"] = f"could not assign the proxy: HTTP {res['status']}"
            table.append(row)
            continue
        start = {"api": "local", "method": "POST", "path": "/profiles/start", "body": {"uuid": uid, "headless": True, "debug_port": True}}
        try:
            res = call(start)
        except LimitStop as e:
            row["status"] = f"not checked: {e}"
            table.append(row)
            break
        except (requests.RequestException, OSError):
            row["status"] = "the Octo client stopped answering: the check was stopped"
            table.append(row)
            break
        resp = res["response"] if isinstance(res["response"], dict) else {}
        if not ok(res) or not resp.get("ws_endpoint"):
            reason = resp.get("code") or resp.get("error") or "HTTP " + str(res["status"])
            proxy_fault = "proxy" in str(reason).lower() or resp.get("error_code") in (4, 5)  # Octo: proxy data errors
            row["status"] = f"proxy does not work: {reason}" if proxy_fault else f"could not start the check profile: {reason}"
            table.append(row)
            if proxy_fault:
                continue
            break  # not the proxy's fault: the same error would mark every next proxy as dead
        remember_session(start, res)
        inflight(uid, resp, "proxy check")
        conn = resp.get("connection_data") or {}
        row["octo_ip"], row["octo_country"] = conn.get("ip", ""), conn.get("country", "")
        try:  # the same four checkers through the same proxy, next to what Octo itself shows
            octo = {"ip": row["octo_ip"], "country": row["octo_country"],
                    "cc": str(conn.get("countryCode") or conn.get("country_code") or "").upper()}
            results, _, why = geo_check(px)
            row.update(geo_row(results))
            if not why:
                row["status"] = verdict(results, octo)
            elif why.startswith("not checked"):
                row["status"] = "ok in Octo; the checkers can't use this proxy type"
            else:
                row["status"] = "ok in Octo, but curl could not use it: " + why.removeprefix("does not work: ")
            lines.append((row, results, octo))
        finally:
            stop = {"api": "local", "method": "POST", "path": "/profiles/stop", "body": {"uuid": uid}}
            try:
                res = call(stop)
                remember_session(stop, res)
            except (requests.RequestException, OSError):
                res = {"status": 0, "response": None}
            if ok(res):
                inflight(uid)
            else:
                row["status"] += f"; the profile '{CHECK_PROFILE}' did not stop, close it in the client"
        table.append(row)

    REPORTS.mkdir(exist_ok=True)
    report = REPORTS / f"proxycheck_octo_{stamp}.csv"
    with report.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
        writer = csv.DictWriter(f, list(table[0]))
        writer.writeheader()
        writer.writerows(table)
    print(f"{'Proxy':28} {'IP (Octo)':16} {'Countries':65} Status")
    checked = {id(row): (results, octo) for row, results, octo in lines}
    for row in table:
        results, octo = checked.get(id(row), ({}, None))
        line = geo_line(results, octo)[1] if octo else "-"
        print(f"{row['proxy']:28} {row['octo_ip'][:16]:16} {line:65} {row['status']}" + (f"  [{row['used_by'][:90]}]" if row["used_by"] else ""))
    print(f"Report: reports/{report.name}. Data: Octo Browser, {GEO_CREDITS}.")
    if any(px.get("inline") for px in proxies):  # don't keep a copy of someone's proxy login in check-proxy
        try:
            call({"api": "public", "method": "PATCH", "path": f"/profiles/{uid}", "body": {"proxy": None}})
        except LimitStop:
            pass
    journal("proxycheck", "-", {"summary": f"Proxy check of {len(table)} proxies in the profile {CHECK_PROFILE}"},
            [], [], [f"The profile '{CHECK_PROFILE}' keeps the last checked saved proxy; a copied profile proxy is removed."])
    report_time()


def warm(path):
    """Warms cookies: every profile is started, walked through the approved sites by warmer.py and stopped normally."""
    try:  # utf-8-sig: a plan saved by Notepad starts with a BOM
        plan = json.loads(Path(path).read_bytes().decode("utf-8-sig"))
    except (OSError, ValueError) as e:
        sys.exit(f"Can't read the warm plan {path}: {e}")
    if not isinstance(plan, dict):
        sys.exit('The warm plan must be {"profiles": [...], "sites": [...], "options": {...}}.')
    listed = lambda v: [v] if isinstance(v, str) else [x for x in v if isinstance(x, str)] if isinstance(v, list) else []  # noqa: E731
    uuids, sites = list(dict.fromkeys(listed(plan.get("profiles")))), list(dict.fromkeys(listed(plan.get("sites"))))
    if not isinstance(plan.get("options") or {}, dict):
        sys.exit("options in the warm plan must be an object.")
    opts = {**WARM_DEFAULTS, **(plan.get("options") or {})}
    span = opts["time_per_page"]
    problems = ([] if uuids and sites else ["the plan needs profiles (uuids) and sites"]) \
        + ([] if str(opts["pages_per_site"]).isdigit() and 1 <= int(opts["pages_per_site"]) <= 10 else ["pages_per_site is 1-10"]) \
        + ([] if isinstance(span, list) and len(span) == 2 and all(isinstance(x, (int, float)) for x in span)
           and 3 <= span[0] <= span[1] <= 120 else ["time_per_page is [min, max] seconds within 3-120"]) \
        + ([] if opts["scroll"] in ("less", "normal", "more", "deep") else ["scroll is less, normal, more or deep"]) \
        + (["country is the full name as Octo shows it, for example Germany, not a code"]
           if len(str(plan.get("country") or "").strip()) == 2 and country(plan.get("country")) == str(plan.get("country")).strip().lower()
           else [])
    unknown = set(plan.get("options") or {}) - set(WARM_DEFAULTS) - {"debug"}
    problems += [f"unknown option(s) {', '.join(sorted(unknown))}: allowed are {', '.join(sorted([*WARM_DEFAULTS, 'debug']))}"] \
        if unknown else []
    problems += [f"{k} is true or false" for k in ("random_order", "accept_consent", "external", "headless", "images", "debug")
                 if k in opts and not isinstance(opts[k], bool)]
    if problems:
        sys.exit("Warm plan refused: " + "; ".join(problems) + ". Nothing was started.")
    try:  # the Octo client must be open: checked before anything is started (this request is free)
        res = call({"api": "local", "method": "GET", "path": "/profiles/active"})
    except (requests.RequestException, OSError):
        sys.exit(f"The Octo client is not running or its local port is not {LOCAL_PORT}. "
                 "Open the client and run the warm-up again. Nothing was started.")
    active = res["response"].get("data") if isinstance(res["response"], dict) else res["response"]
    running = {x.get("uuid") for x in active or [] if isinstance(x, dict)}
    info, _ = fetch(uuids, False)
    flags = ["--disable-backgrounding-occluded-windows"] + ([] if opts["images"] else
                                                             ["--blink-settings=imagesEnabled=false", "--autoplay-policy=user-gesture-required"])
    pages, span_max = int(opts["pages_per_site"]), span[1]
    print(f"Warming {len(uuids)} profile(s) on {len(sites)} site(s), up to {pages} page(s) per site. "
          "Profiles are stopped normally afterwards, so the cookies are synced.")
    rows, summary = [], []
    run_id, summary_text = datetime.now().strftime("%Y%m%d-%H%M%S"), plan.get("summary") or f"Cookie warm-up of {len(uuids)} profile(s)"

    def record(item):
        """Adds (base, status, site rows) to the summary and writes the profile to the farm log at once:
        a long run killed halfway keeps every profile it finished."""
        summary.append(item)
        base, status, done = item
        keep = ("site", "started", "finished", "pages", "seconds", "consent", "cookies_added", "cookies_total", "status")  # never values
        sites_done = [{k: r.get(k) for k in keep} for r in done]
        line = {"run": run_id, "time": datetime.now().isoformat(timespec="seconds"), "plan": str(path),
                "summary": summary_text, "profile": base["uuid"], "title": base["profile"], "ip": base["ip"],
                "country": base["country"], "status": status, "sites_total": len(sites),
                "sites_ok": sum(1 for r in done if r.get("status") == "ok"), "pages": sum(r.get("pages") or 0 for r in done),
                "cookies_added": sum(r.get("cookies_added") or 0 for r in done), "sites": sites_done}
        WARM_LOG.parent.mkdir(exist_ok=True)
        with WARM_LOG.open("a", encoding="utf-8") as f:
            f.write(json.dumps(line, ensure_ascii=False) + "\n")

    for uid in uuids:
        title = (info.get(uid) or {}).get("title") or uid
        base = {"profile": title, "uuid": uid, "ip": "", "country": ""}
        skip = ("profile not found" if uid not in info
                else f"has the tag '{PROTECTED_TAG}'" if PROTECTED_TAG in {t.lower() for t in names(info[uid].get("tags"))}
                else "already running: it may be in use" if uid in running else None)
        if skip:
            rows.append({**base, "status": f"skipped: {skip}"})
            record((base, f"skipped: {skip}", []))
            continue
        start = {"api": "local", "method": "POST", "path": "/profiles/start",
                 "body": {"uuid": uid, "headless": bool(opts["headless"]), "debug_port": True, "flags": flags}}
        try:
            res = call(start)
        except LimitStop as e:  # the report of the profiles already warmed is still written below
            rows.append({**base, "status": f"not started: {e}"})
            record((base, rows[-1]["status"], []))
            break
        except (requests.RequestException, OSError):
            rows.append({**base, "status": "the Octo client stopped answering: the warm-up was stopped"})
            record((base, rows[-1]["status"], []))
            break
        resp = res["response"] if isinstance(res["response"], dict) else {}
        if not (200 <= res["status"] < 300 and resp.get("ws_endpoint")):
            rows.append({**base, "status": f"could not start: {resp.get('code') or resp.get('error') or 'HTTP ' + str(res['status'])}"})
            record((base, rows[-1]["status"], []))
            continue
        remember_session(start, res)
        inflight(uid, resp, "warm-up")
        conn = resp.get("connection_data") or {}
        base["ip"], base["country"] = conn.get("ip", ""), conn.get("country", "")
        site_rows, status = [], "ok"
        try:
            want = country(plan.get("country"))
            have = {country(conn.get(k)) for k in ("country", "countryCode", "country_code", "countryName")} - {""}
            if want and want not in have:  # a German persona is not warmed through an American IP
                status = f"skipped: the IP is in {base['country'] or 'an unknown country'}, the plan expects {plan['country']}"
                continue
            job = json.dumps({"ws_endpoint": resp["ws_endpoint"], "sites": sites, "options": opts})
            print(f"{title}: IP {base['ip'] or '?'} {base['country']}, warming {len(sites)} site(s)...", flush=True)
            import threading
            proc = subprocess.Popen([sys.executable, str(DIR / "warmer.py")], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                    stderr=subprocess.STDOUT, encoding="utf-8", errors="replace",
                                    env={**os.environ, "PYTHONIOENCODING": "utf-8"})
            late = threading.Event()  # set when warmer.py runs past its time and is stopped
            timer = threading.Timer(len(sites) * pages * (span_max + 15) + 120, lambda: (late.set(), proc.kill()))
            timer.start()
            err = []
            try:
                proc.stdin.write(job)
                proc.stdin.close()
                for line in proc.stdout:  # one line per finished site: shown at once, so a long or headless run can be watched
                    try:
                        item = json.loads(line)
                    except ValueError:
                        err += [line.strip()] if line.strip() else []
                        continue
                    if isinstance(item, dict) and "site" in item:
                        site_rows.append(item)
                        print(f"  {title[:28]}: site {len(site_rows)}/{len(sites)} {item.get('site')}: {item.get('pages', 0)} page(s), "
                              f"+{item.get('cookies_added', 0)} cookies, consent {item.get('consent') or '-'}, {item.get('status')}",
                              flush=True)
                proc.wait()
            finally:
                timer.cancel()
            err += ["warmer.py did not finish in time"] if late.is_set() else []
            if not site_rows:
                status = "warmer.py failed: " + (err[-1][:150] if err else "no output")
            elif len(site_rows) < len(sites):
                status = f"stopped after {len(site_rows)} of {len(sites)} sites"
        finally:
            stop = {"api": "local", "method": "POST", "path": "/profiles/stop", "body": {"uuid": uid}}
            try:
                res = call(stop)
                remember_session(stop, res)
                stopped_ok = 200 <= res["status"] < 300
            except (requests.RequestException, OSError):
                stopped_ok = False
            if stopped_ok:
                inflight(uid)
            else:
                status += "; the profile did not stop, close it in the client"
            rows += [{**base, **row, **clock(row)} for row in site_rows] or [{**base, "status": status}]
            if site_rows:  # the profile's own line: its first start, last end and every sum
                rows.append(total_row(base, site_rows, len(sites)))
            record((base, status, site_rows))
    totals = [r for r in rows if r.get("site") == "TOTAL"]
    if len(totals) > 1:  # several profiles in one warm-up: one line for all of them
        rows.append({"profile": f"{len(totals)} profiles", "site": "ALL",
                     "started": min((r["started"] for r in totals if r["started"]), default=""),
                     "finished": max((r["finished"] for r in totals if r["finished"]), default=""),
                     **{k: sum(r.get(k) or 0 for r in totals) for k in ("pages", "seconds", "cookies_added", "cookies_total")},
                     "status": f"{len(totals)} profiles warmed"})
    REPORTS.mkdir(exist_ok=True)
    report = REPORTS / f"warm_{datetime.now():%Y-%m-%d_%H-%M-%S}.csv"  # seconds: two warm-ups never overwrite each other
    columns = ["profile", "uuid", "ip", "country", "site", "started", "finished", "pages", "seconds", "consent",
               "cookies_added", "cookies_total", "status"]
    with report.open("w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: Excel shows non-Latin text correctly
        writer = csv.DictWriter(f, columns, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)
    print(f"{'Profile':28} {'IP (Octo)':16} {'Country':14} {'Sites':7} {'Pages':6} {'Cookies':8} Status")
    for base, status, done in summary:
        ok_sites = sum(1 for r in done if r.get("status") == "ok")
        print(f"{base['profile'][:28]:28} {base['ip']:16} {base['country'][:14]:14} {f'{ok_sites}/{len(sites)}':7} "
              f"{sum(r.get('pages', 0) for r in done):<6} {'+' + str(sum(r.get('cookies_added', 0) for r in done)):8} {status}")
    print(f"Report by site: reports/{report.name}. Cookie values are never shown, only their number.")
    print(f"Farm log: logs/warm.jsonl (python octo_api.py warmlog). Run {run_id}.")
    report_time()


def clock(row):
    """A site's start and end in local time, for the report."""
    return {k: datetime.fromtimestamp(row[k]).strftime("%Y-%m-%d %H:%M:%S") if isinstance(row.get(k), (int, float)) else ""
            for k in ("started", "finished")}


def total_row(base, done, planned):
    """The TOTAL line of one profile: first start, last end, sums, consents counted and the cookies it has now."""
    seen = {}
    for r in done:
        seen[r.get("consent") or "-"] = seen.get(r.get("consent") or "-", 0) + 1
    ok = sum(1 for r in done if r.get("status") == "ok")
    return {**base, "site": "TOTAL", **clock({"started": done[0].get("started"), "finished": done[-1].get("finished")}),
            "pages": sum(r.get("pages") or 0 for r in done), "seconds": sum(r.get("seconds") or 0 for r in done),
            "consent": " / ".join(f"{n} {c}" for c, n in seen.items()), "cookies_added": sum(r.get("cookies_added") or 0 for r in done),
            "cookies_total": done[-1].get("cookies_total", ""), "status": f"{ok} of {planned} sites ok"}


def terminated(signum, frame):
    """Turns a kill into a normal exit: finally blocks still stop profiles, a plan journals what is done."""
    raise Interrupted(f"Interrupted by signal {signum} (a timeout or a closed window)")


for name in ("SIGTERM", "SIGBREAK", "SIGHUP"):  # SIGBREAK: Ctrl+Break on Windows, SIGHUP: a closed terminal
    if hasattr(signal, name):
        signal.signal(getattr(signal, name), terminated)
def no_connection(kind, value, tb):
    """A lost connection outside a plan (audit, proxy check, warm-up): a clear message instead of a traceback."""
    if issubclass(kind, requests.RequestException):
        sys.stderr.write(f"No connection to Octo Browser ({kind.__name__}): check the internet and that the Octo client "
                         f"is running (local port {LOCAL_PORT}). Nothing more was sent.\n")
    else:
        sys.__excepthook__(kind, value, tb)


sys.excepthook = no_connection


def option(name, default):
    """A number after --days, --limit and the like."""
    if name not in args:
        return default
    i = args.index(name)
    if i + 1 >= len(args) or not args[i + 1].isdigit():
        sys.exit(f"{name} needs a whole number, for example {name} {default}")
    return int(args[i + 1])


args = sys.argv[1:]
out = None
if "--out" in args:
    i = args.index("--out")
    out = args[i + 1] if i + 1 < len(args) else sys.exit("Specify a file after --out, for example reports/profiles.csv")
    del args[i:i + 2]
if not args:
    sys.exit(USAGE)
if {"--help", "-h", "help", "/?"} & set(args):  # help only prints: it never runs a command
    print(USAGE)
    sys.exit(0)
ALLOWED = {"proxyimport": {"--prefix", "--type", "--file", "--geo"}, "proxycheck": {"--proxy", "--limit", "--octo", "--profile", "--failed"}}
if args[0] in ALLOWED:
    odd = [a for a in args[1:] if a.startswith("-") and a not in ALLOWED[args[0]]]
    if odd:
        sys.exit(f"Unknown option(s) {', '.join(odd)} for {args[0]}: allowed are {', '.join(sorted(ALLOWED[args[0]]))}. Nothing was done.")
if args[0] == "site":
    site(args[1] if len(args) > 1 else None)
    sys.exit(0)
if args[0] == "warmlog":
    warmlog(option("--last", 30), value("--profile"))
    sys.exit(0)
if args[0] == "export":
    fmt = (value("--format") or "xlsx").lower()
    if fmt not in ("xlsx", "csv"):
        sys.exit("--format is xlsx or csv")
    f, search = filters_from()
    export(f["tag"], f["folder"], search, value("--add") or "", value("--exclude") or "", fmt, value("--sep"), out, f)
    sys.exit(0)
if args[0] == "find":
    find(*filters_from())
    sys.exit(0)
if args[0] == "history":
    history(option("--last", 20), value("--profile"))
    sys.exit(0)
if args[0] != "undo":
    recover()
if args[0] == "undo":
    undo_plan(args[1] if len(args) > 1 else None)
    sys.exit(0)
if args[0] in ("start", "stop"):
    start_stop(args[0], args[1], "--headless" in args) if len(args) > 1 and not args[1].startswith("-") else \
        sys.exit(f'Name the profile: python octo_api.py {args[0]} "<title or uuid>"')
    sys.exit(0)
if args[0] == "proxyimport":
    prefix, kind = value("--prefix"), value("--type") or "http"
    if prefix is not None and not re.fullmatch(r"[\w-]{1,40}", prefix):
        sys.exit("--prefix takes letters, digits, _ or -, for example --prefix DE")
    only = value("--file")
    if only is not None and not re.fullmatch(r"[^/\\:*?\"<>|]{1,200}\.(txt|csv)", only, re.I):
        sys.exit("--file takes the name of a .txt or .csv file in proxies_in, for example --file de.txt")
    proxyimport(prefix, kind, only, "--geo" in args)
    sys.exit(0)
if args[0] == "warm":
    warm(args[1]) if len(args) > 1 else sys.exit("Specify the plan: python octo_api.py warm plans/<file>.json")
    sys.exit(0)
if args[0] == "proxycheck":
    proxycheck(value("--proxy"), option("--limit", 20), "--octo" in args, value("--profile"), "--failed" in args)
    sys.exit(0)
if args[0] == "audit":
    days = option("--days", 30)
    mark = args[args.index("--mark") + 1] if "--mark" in args else "tag"
    audit(days, mark if mark in ("tag", "folder") else "tag")
    sys.exit(0)
if args[0] not in ("read", "show", "run", "delete"):  # every other command was handled above
    sys.exit(f"Unknown command: {args[0]}\n{USAGE}")
if len(args) < 2:
    sys.exit(USAGE)
cmd, path, given = args[0], args[1], (args[2] if len(args) > 2 else None)
plan, reqs = load(path)
has_changes = any(is_change(r) for r in reqs)
has_delete = any(r["method"] == "DELETE" for r in reqs)
risky = [r for r in reqs if sensitive(r)]
final = [r for r in reqs if (any(word in r["path"] for word in IRREVERSIBLE) and is_change(r))
         or (r["method"] == "DELETE" and norm(r["path"]) != "/profiles")]  # tags, folders, proxies have no trash

notes = spec_notes(reqs) + fingerprint_notes(reqs) if cmd in ("read", "show") else []
blocked = protected(reqs) if cmd in ("show", "run", "delete") else []
if blocked:
    for uuid, title in blocked:
        print(f"BLOCKED: {uuid} ({title}): the assistant can't change, delete or transfer profiles with the tag "
              f"'{PROTECTED_TAG}', nor the tag itself.")
    sys.exit("Remove these profiles from the plan. Only the user can remove the protected tag, in the Octo client.")
if not ALLOW_DELETE and has_delete and cmd in ("show", "run", "delete"):
    sys.exit("BLOCKED: deleting is turned off (ALLOW_DELETE = False in octo_api.py). Delete by hand in the Octo client, "
             "or the user sets ALLOW_DELETE = True in octo_api.py.")
if cmd == "read":
    if has_changes or risky:
        sys.exit("The plan has changes or sensitive sections: show it with the show command first.")
    for note in notes:
        print(f"CHECK: {note}")
elif cmd == "show":
    print(f"Plan: {plan.get('summary', '')}")
    for r in reqs[:10]:
        print(f"  {r['method']} {r['path']} {json.dumps(redact(r.get('body') or {}), ensure_ascii=False)[:150]}")
    if len(reqs) > 10:
        print(f"  ...and {len(reqs) - 10} more")
    if risky:
        print("WARNING: the plan touches sensitive sections (passwords, cookies, client login, update, team).")
    for r in reqs:
        if exposed(r):
            print(f"WARNING: {r['method']} {r['path']} starts the profile with {exposed(r)}.")
    for r in final:
        receiver = (r.get("body") or {}).get("receiver_email")
        note = (f" -> recipient {receiver}" if receiver else " (deleted for good, there is no trash)" if r["method"] == "DELETE"
                else " (work done since the profile was started is not synced and is lost)" if "force_stop" in r["path"] else "")
        print(f"IRREVERSIBLE: {r['method']} {r['path']}{note}")
    for note in notes:
        print(f"CHECK: {note}")
    if any(r["method"] == "DELETE" and norm(r["path"]) == "/profiles" for r in reqs):
        print("Deleted profiles will go to the trash for 72 hours. Only the master account can restore them, in the client (no API "
              "method): free if the profile was deleted only once in the last 7 days, otherwise 1 token per profile.")
    print(f"To run: python octo_api.py {'delete' if has_delete else 'run'} {path} {STATS['code']}")
    sys.exit(0)
elif cmd in ("run", "delete"):
    if given != STATS["code"]:
        sys.exit("Code mismatch: the plan changed after it was shown. Show it again with the show command.")
    if has_delete != (cmd == "delete"):
        sys.exit("Use delete for plans with deletions and run for all other plans.")
else:
    sys.exit(USAGE)

results, undo, manual, stopped = execute(reqs, with_undo=cmd != "read")
try:
    check, report = verify(reqs, results) if cmd != "read" else (None, "")
except (requests.RequestException, OSError, LimitStop) as e:  # no requests possible: keep the journal anyway
    check, report = None, f"Verification skipped: {str(e)[:150]}"
number = (journal(cmd, path, plan, results, undo, manual, check, stopped)
          if cmd != "read" or any(r["method"] != "GET" for r in reqs) else None)
if out:
    print(save(out, results))  # full data stays on the user's computer
else:
    text = json.dumps(redact(results), ensure_ascii=False)
    print(text if len(text) < 20000 else text[:20000] + " ...(response truncated, save it to a file with --out)")
if report:
    print(report)
if stopped:
    done = stopped["at"] - 1
    print(f"Stopped at request {stopped['at']} of {len(reqs)}: {stopped['reason']}. Done: {done}.")
    rest = reqs[done:]
    if cmd != "read" and rest:
        rest_path = Path(path).with_name(Path(path).stem + "_rest.json")
        rest_path.write_text(json.dumps({"summary": plan.get("summary", "") + " (remaining part)", "requests": rest},
                                        ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"The remaining {len(rest)} request(s) are saved to {rest_path.as_posix()}: "
              "fix the problem, then show and run it.")
if number:
    print(f"Operation {number} saved to the journal." + (" Undo: python octo_api.py undo " + str(number) if undo else ""))
report_time()
Browser script: browser.py
The browser.py script performs a single action in the browser of a running profile: opening a website, reading a page, clicking a button, entering text, or taking a screenshot. You can perform several actions faster in a single call using then—for example, open, followed by wait and text. Create the file by running notepad browser.py, paste in the code, and save it:
import json
import os
import re
import sys
import time
from datetime import datetime
from pathlib import Path
from urllib.parse import parse_qsl, urlparse
from playwright.sync_api import TimeoutError as PlaywrightTimeout, sync_playwright

DIR = Path(__file__).parent
# True: browser.py opens only sites the user approved (python octo_api.py site <address>, confirmed in Claude Code),
# so text on a page can't make it send data to an address of its choice. False: any site opens without a question.
APPROVE_SITES = False
BUILT_IN = ("example.com", "iplocation.io")  # always open: the guide's checks use them
JOURNAL = DIR / "logs" / "actions.jsonl"  # the executor's journal: every browser.py call is added to it
SECRET_PARAM = re.compile(r"token|key|pass|secret|auth|code|sig|session|otp", re.I)  # hidden in logged addresses


def address(text):
    return text if text.lower().startswith(("http://", "https://")) else "https://" + text


def approved(url):
    host = (urlparse(url).hostname or "").lower().removeprefix("www.")
    try:
        sites = json.loads((DIR / "allowed_sites.json").read_text(encoding="utf-8")).get("sites", [])
    except (OSError, ValueError, AttributeError):
        sites = []
    return bool(host) and any(host == s or host.endswith("." + s) for s in (*BUILT_IN, *map(str.lower, sites)))


def masked(url):
    """An address as the journal keeps it: no login:password@, no fragment, secret parameters as ***."""
    u = urlparse(url)
    host = (u.hostname or "") + (f":{u.port}" if u.port else "")
    query = "&".join(f"{k}=***" if SECRET_PARAM.search(k) else f"{k}={v}" for k, v in parse_qsl(u.query, keep_blank_values=True))
    return u._replace(netloc=host, query=query, fragment="").geturl()


def shown(action, args):
    """What the journal keeps of a step: values typed with fill are never written."""
    if action == "fill":
        return f"{args[0]} ***" if args else ""
    if action == "open" and args:
        return masked(address(args[0]))
    return " ".join(args)


def journal(records, stopped):
    """Adds this call to the journal. The journal must never break a browser action, so errors are ignored."""
    if os.environ.get("OCTO_NO_BROWSER_JOURNAL"):  # the proxy check journals its own run
        return
    try:
        JOURNAL.parent.mkdir(exist_ok=True)
        lines = JOURNAL.read_text(encoding="utf-8").splitlines() if JOURNAL.exists() else []
        entry = {"id": len(lines) + 1, "time": datetime.now().isoformat(timespec="seconds"), "command": "browser",
                 "plan": "-", "profile": uuid, "summary": " -> ".join(f"{r['action']} {r['detail']}".strip() for r in records),
                 "steps": records, "requests": [], "undo": [], "manual": [], "check": None, "stopped": stopped}
        with JOURNAL.open("a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except OSError:
        pass


USAGE = ("Usage: browser.py <profile uuid> <action> [arguments] [then <action> [arguments] ...]\n"
         "Actions: open <url>, read, text, click <text>, fill <field> <text>, press <key>, scroll, wait <seconds>, "
         "screenshot [file name]")
if len(sys.argv) < 3:
    sys.exit(USAGE)
uuid, rest = sys.argv[1], sys.argv[2:]
steps, step = [], []
for word in rest:  # several actions in one call: open example.com then wait 2 then text
    if word == "then":
        steps.append(step)
        step = []
    else:
        step.append(word)
steps.append(step)
if any(not s for s in steps):
    sys.exit(USAGE)
NEEDS = {"open": 1, "click": 1, "fill": 2, "press": 1}  # actions that need arguments
for action, *args in steps:
    if len(args) < NEEDS.get(action, 0):
        sys.exit(f"'{action}' needs {NEEDS[action]} argument(s).\n{USAGE}")
    if action == "wait" and args and not args[0].replace(".", "", 1).isdigit():
        sys.exit(f"wait takes seconds, for example: wait 2\n{USAGE}")
    if APPROVE_SITES and action == "open" and not approved(address(args[0])):  # checked before anything is opened
        host = urlparse(address(args[0])).hostname or args[0]
        sys.exit(f"The site {host} is not approved yet, nothing was opened. If the user asked for this site, run:\n"
                 f"python octo_api.py site {host}\nClaude Code will ask the user to confirm it.")
records = [{"action": action, "detail": shown(action, args), "result": "not run"} for action, *args in steps]
current, stopped = None, None
try:
    sessions = DIR / "sessions.json"
    session = json.loads(sessions.read_text(encoding="utf-8")).get(uuid) if sessions.exists() else None
    ws = session.get("ws_endpoint") if isinstance(session, dict) else session
    if not ws:
        sys.exit(f"Profile {uuid} is not running. Start it first via octo_api.py (POST /profiles/start).")

    with sync_playwright() as p:
        try:
            browser = p.chromium.connect_over_cdp(ws)
        except Exception as e:  # the window was closed by hand: sessions.json still has the old address
            sys.exit(f"Can't connect to profile {uuid}: it was probably closed. Start it again via octo_api.py. ({str(e)[:120]})")
        context = browser.contexts[0]
        page = context.pages[-1] if context.pages else context.new_page()
        page.set_default_timeout(15000)

        for current, (action, *args) in enumerate(steps):
            if action == "open":
                page.goto(address(args[0]), wait_until="domcontentloaded")
            elif action == "read":
                print(page.locator("body").aria_snapshot()[:6000])
            elif action == "text":
                print(page.inner_text("body")[:20000])
            elif action == "click":
                text = " ".join(args)
                found = None
                for exact in (True, False):  # "Delete" must not press "Delete account" when a plain "Delete" exists
                    for target in (page.get_by_role("button", name=text, exact=exact),
                                   page.get_by_role("link", name=text, exact=exact),
                                   page.get_by_text(text, exact=exact)):
                        if target.count():
                            found = target
                            break
                    if found is not None:
                        break
                if found is None:
                    sys.exit(f"Not found on the page: '{text}'")
                found.first.click()
            elif action == "fill":
                field, value = args[0], " ".join(args[1:])
                page.get_by_label(field).or_(page.get_by_placeholder(field)).first.fill(value)
            elif action == "press":
                page.keyboard.press(args[0])
            elif action == "scroll":
                page.mouse.wheel(0, 800)
            elif action == "wait":
                time.sleep(min(float(args[0]) if args else 2, 30))
            elif action == "screenshot":
                name = args[0] if args else "screenshot.png"
                (DIR / "reports").mkdir(exist_ok=True)
                page.bring_to_front()  # a minimized or covered window can't be captured
                try:
                    page.screenshot(path=str(DIR / "reports" / Path(name).name), timeout=30000)
                except PlaywrightTimeout:
                    sys.exit("Screenshot timed out: the profile window is probably minimized or covered. "
                             "Restore it and try again.")
                print(f"Screenshot saved: reports/{Path(name).name}")
            else:
                sys.exit(f"Unknown action: {action}\n{USAGE}")
            page.wait_for_load_state("domcontentloaded")
            records[current].update(result="ok", page=masked(page.url))

        print(f"Page: {page.title()} - {page.url}")
except BaseException as e:  # the journal shows where and why the call stopped, then the error goes on as before
    reason = str(e.code) if isinstance(e, SystemExit) else "interrupted" if isinstance(e, KeyboardInterrupt) \
        else f"{type(e).__name__}: {(str(e).splitlines() or [''])[0][:150]}