tor_army.py

#!/usr/bin/env python3
"""Tor Army v3 -- Async Swarm Scraper

Major upgrades over v2:
  - Async engine (asyncio + curl_cffi AsyncSession) -- eliminates GIL bottleneck
  - HTTP/2 multiplexing -- shared session per Tor instance, multiple workers
    send concurrent streams on one TCP connection (400 FDs instead of 2000)
  - Worker multiplexing -- N async workers per Tor instance (default 5x)
  - Optimized torrc -- faster circuits, no entry guards, shorter timeouts
  - Per-instance rate limiting -- prevents WAF bursts from shared exit IPs
  - Jittered exponential backoff -- smarter WAF/error recovery
  - Expanded browser fingerprints (7 browsers incl. Firefox)
  - Max retry cap -- prevents infinite re-queue loops
  - Scale-ready for 400+ Tor instances (128GB RAM headroom)
  - Live dashboard with rate, WAF/min, per-target breakdown, ETA

  Measured performance:
  v2: 198K pages/hr peak (240 instances x 2 workers, threading)
  v3: 234K pages/hr sustained (400 instances x 5 workers, async + HTTP/2)

Usage:
    python tor_army.py --start-tor --workers 400 --targets npc
    python tor_army.py --workers 400 --multiplier 5 --targets npc,quest
    python tor_army.py --targets npc --reparse
    python tor_army.py --list-targets
    python tor_army.py --smoke 50 --targets npc --workers 10 --start-tor
    python tor_army.py --kill-tor
"""

import argparse
import asyncio
import gzip
import json
import os
import random
import re
import select
import subprocess
import sys
import time
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path

# -- Windows select() FD limit bypass -----------------------------------------
# Windows select() is limited to 512 FDs (FD_SETSIZE). curl_cffi's vendored
# Tornado selector hits this with >240 Tor instances. Monkey-patch select.select
# to batch calls in chunks of 450.
if sys.platform == "win32":
    _orig_select = select.select

    def _batched_select(rlist, wlist, xlist, timeout=None):
        BATCH = 450
        if len(rlist) <= BATCH and len(wlist) <= BATCH and len(xlist) <= BATCH:
            return _orig_select(rlist, wlist, xlist, timeout)
        # For large FD sets, poll in batches with zero timeout,
        # fall back to a short sleep if nothing is ready
        all_r, all_w, all_x = [], [], []
        max_len = max(len(rlist), len(wlist), len(xlist), 1)
        for i in range(0, max_len, BATCH):
            rb = rlist[i:i + BATCH]
            wb = wlist[i:i + BATCH]
            xb = xlist[i:i + BATCH]
            if not rb and not wb and not xb:
                continue
            try:
                r, w, x = _orig_select(rb, wb, xb, 0)
                all_r.extend(r)
                all_w.extend(w)
                all_x.extend(x)
            except (ValueError, OSError):
                pass
        if not all_r and not all_w and not all_x and timeout != 0:
            # Nothing ready — do a short blocking wait on the first batch
            try:
                r, w, x = _orig_select(
                    rlist[:BATCH], wlist[:BATCH], xlist[:BATCH],
                    min(timeout, 0.1) if timeout is not None else 0.1,
                )
                all_r.extend(r)
                all_w.extend(w)
                all_x.extend(x)
            except (ValueError, OSError):
                pass
        return all_r, all_w, all_x

    select.select = _batched_select
# ------------------------------------------------------------------------------

from curl_cffi.const import CurlMOpt
from curl_cffi.requests import AsyncSession
from stem import Signal
from stem.control import Controller

# == Configuration =============================================================

WAGO_DIR = Path(__file__).parent

# Tor Expert Bundle location — override with --tor-dir flag or TOR_DIR env var
# Download from: https://www.torproject.org/download/tor/
_default_tor = Path(os.environ.get("TOR_DIR", "")) if os.environ.get("TOR_DIR") else None
TOR_DIR = _default_tor or (WAGO_DIR / "tor")

# Expanded fingerprint pool -- more diversity = harder for CF to pattern-match
FINGERPRINTS = [
    "chrome120", "chrome124", "chrome131",
    "edge101",
    "safari17_0", "safari17_2_ios",
    "firefox120",
]

MAX_RETRIES = 5  # Per work item, then dropped as permanent error

# Target configurations — all 39 entity types
# IDs loaded from id_lists/{target}.txt (generated by generate_id_lists.py)
def _tcfg(url_path: str, target: str):
    return {
        "ids_file": f"id_lists/{target}.txt",
        "out_dir": f"wowhead_data/{target}/raw",
        "html_dir": f"wowhead_data/{target}/html",
        "out_pattern": f"{target}_{{id}}.json",
        "html_pattern": "{id}.html.gz",
        "url": f"https://www.wowhead.com/{url_path}={{id}}",
    }

TARGET_CONFIGS = {
    # --- Core entities ---
    "quest":              _tcfg("quest", "quest"),
    "npc":                _tcfg("npc", "npc"),
    "trainer":            _tcfg("npc", "trainer"),
    "vendor":             _tcfg("npc", "vendor"),
    "object":             _tcfg("object", "object"),
    "item":               _tcfg("item", "item"),
    "spell":              _tcfg("spell", "spell"),
    "achievement":        _tcfg("achievement", "achievement"),
    "mount":              _tcfg("mount", "mount"),
    "currency":           _tcfg("currency", "currency"),
    "faction":            _tcfg("faction", "faction"),
    "title":              _tcfg("title", "title"),
    "questline":          _tcfg("questline", "questline"),
    "event":              _tcfg("event", "event"),
    "item_set":           _tcfg("item-set", "item_set"),
    # --- Transmog & appearance ---
    "transmog_set":       _tcfg("transmog-set", "transmog_set"),
    "transmog_item":      _tcfg("transmog", "transmog_item"),
    # --- Classes, races, professions ---
    "class":              _tcfg("class", "class"),
    "race":               _tcfg("race", "race"),
    "specialization":     _tcfg("specialization", "specialization"),
    "profession":         _tcfg("profession", "profession"),
    "skill_ability":      _tcfg("spell", "skill_ability"),
    "profession_trait":   _tcfg("spell", "profession_trait"),
    # --- Pet & mount related ---
    "battle_pet":         _tcfg("npc", "battle_pet"),
    "pet_family":         _tcfg("petfamily", "pet_family"),
    # --- Garrison ---
    "garrison_mission":   _tcfg("mission", "garrison_mission"),
    "garrison_building":  _tcfg("building", "garrison_building"),
    "follower":           _tcfg("follower", "follower"),
    # --- Dungeon & encounter ---
    "dungeon":            _tcfg("dungeon", "dungeon"),
    "encounter":          _tcfg("npc", "encounter"),
    # --- Zones & maps ---
    "zone":               _tcfg("zone", "zone"),
    "map":                _tcfg("zone", "map"),
    "ui_map":             _tcfg("zone", "ui_map"),
    "flight_path":        _tcfg("taxinode", "flight_path"),
    # --- Azerite ---
    "azerite_essence":    _tcfg("azerite-essence", "azerite_essence"),
    "azerite_power":      _tcfg("azerite-essence-power", "azerite_power"),
    # --- Icons ---
    "icon":               _tcfg("icon", "icon"),
    # --- Content tuning ---
    "content_tuning":     _tcfg("content-tuning", "content_tuning"),
    # --- Emotes ---
    "emote":              _tcfg("emote", "emote"),
}

# Speed-optimized torrc template
# Key changes vs bare CLI args:
#   UseEntryGuards 0  -- skip persistent guards, more relay diversity
#   CircuitBuildTimeout 10  -- don't wait for slow relays (default 60)
#   NewCircuitPeriod 20  -- pre-build circuits more often (default 30)
#   ExcludeExitNodes  -- avoid slow/censored countries
#   ConnectionPadding 0  -- reduce overhead
TORRC_TEMPLATE = """\
SocksPort {socks_port}
ControlPort {control_port}
CookieAuthentication 0
DataDirectory {data_dir}
GeoIPFile {geoip}
GeoIPv6File {geoip6}

CircuitBuildTimeout 10
LearnCircuitBuildTimeout 0
UseEntryGuards 0
NumEntryGuards 1
NewCircuitPeriod 20
MaxCircuitDirtiness 600
KeepalivePeriod 60
SocksTimeout 20
ConnectionPadding 0
ReducedConnectionPadding 1

ExcludeExitNodes {{cn}},{{ru}},{{ir}},{{kp}},{{sy}},{{by}},{{ve}}
StrictNodes 0

Log warn stderr
"""


# == Tor Fleet Management =====================================================

class TorFleet:
    """Manages a fleet of Tor instances with optimized torrc configs."""

    def __init__(self, count: int):
        self.count = count
        self.procs: list[subprocess.Popen] = []
        # Cross-platform: tor.exe on Windows, tor on Linux/Mac
        tor_name = "tor.exe" if sys.platform == "win32" else "tor"
        self.tor_exe = TOR_DIR / "tor" / tor_name
        if not self.tor_exe.exists():
            # Fallback: tor binary directly in TOR_DIR (Linux package layout)
            self.tor_exe = TOR_DIR / tor_name
        self.geoip = TOR_DIR / "data" / "geoip"
        self.geoip6 = TOR_DIR / "data" / "geoip6"
        self.torrc_dir = TOR_DIR / "torrc"

    def start(self):
        self.torrc_dir.mkdir(exist_ok=True)
        print(f"  Generating {self.count} optimized torrc configs...")

        for i in range(self.count):
            socks_port = 9050 + i * 2
            control_port = 9051 + i * 2
            data_dir = TOR_DIR / f"data{i + 1 if i > 0 else ''}"
            data_dir.mkdir(exist_ok=True)

            torrc_path = self.torrc_dir / f"torrc_{i}"
            torrc_content = TORRC_TEMPLATE.format(
                socks_port=socks_port,
                control_port=control_port,
                data_dir=str(data_dir).replace("\\", "/"),
                geoip=str(self.geoip).replace("\\", "/"),
                geoip6=str(self.geoip6).replace("\\", "/"),
            )
            torrc_path.write_text(torrc_content)

            proc = subprocess.Popen(
                [str(self.tor_exe), "-f", str(torrc_path)],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            self.procs.append(proc)

            if (i + 1) % 50 == 0:
                print(f"  Started {i + 1}/{self.count} Tor instances...")

        wait = min(15 + self.count // 15, 45)
        print(f"  All {self.count} Tor instances launched. Waiting {wait}s for circuits...")
        time.sleep(wait)

        # Spot-check health
        alive = 0
        sample = random.sample(range(self.count), min(10, self.count))
        for i in sample:
            try:
                with Controller.from_port(port=9051 + i * 2) as c:
                    c.authenticate()
                    alive += 1
            except Exception:
                pass
        print(f"  Health: {alive}/{len(sample)} sampled instances responsive")

    def stop(self):
        for proc in self.procs:
            try:
                proc.terminate()
            except Exception:
                pass
        if self.procs:
            print(f"\n  {len(self.procs)} Tor instances terminated.")

    @staticmethod
    def kill_all():
        if sys.platform == "win32":
            subprocess.run(
                ["taskkill", "/F", "/IM", "tor.exe"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
        else:
            subprocess.run(
                ["pkill", "-f", "tor"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )


# == Tor Instance (session + circuit + rate limiter) ===========================

class TorInstance:
    """Manages one Tor instance: shared HTTP/2 session, circuit rotation,
    and rate limiting.

    HTTP/2 multiplexing: all workers sharing this instance send requests as
    concurrent streams on a SINGLE TCP connection. This reduces FD usage from
    N-per-instance (one per worker) to 1-per-instance, eliminating the
    Windows select() 512 FD bottleneck at scale.

    curl_cffi's CurlMulti handle pools connections internally. When multiple
    .get() calls target the same host through the same proxy, libcurl reuses
    the existing HTTP/2 connection and multiplexes as new streams.
    """

    def __init__(self, instance_id: int, socks_port: int, control_port: int,
                 per_circuit: int, min_interval: float):
        self.id = instance_id
        self.socks_port = socks_port
        self.proxy = f"socks5h://127.0.0.1:{socks_port}"
        self.control_port = control_port
        self.per_circuit = per_circuit
        self.min_interval = min_interval

        self._request_count = 0
        self._rotation_lock = asyncio.Lock()
        self._rate_lock = asyncio.Lock()
        self._last_request = 0.0
        self._fingerprint = random.choice(FINGERPRINTS)

        # Shared HTTP/2 session — all workers multiplex on this
        self._session: AsyncSession | None = None
        self._session_fp: str | None = None
        self._session_lock = asyncio.Lock()

    async def acquire(self) -> AsyncSession:
        """Acquire a request slot and return the shared HTTP/2 session.

        Handles circuit rotation (when budget exhausted) and rate limiting
        (minimum interval between requests from this exit IP). The returned
        session is shared across all workers — concurrent .get() calls
        multiplex as HTTP/2 streams on a single connection.
        """
        # Check circuit rotation
        async with self._rotation_lock:
            if self._request_count >= self.per_circuit:
                await self._rotate()
            self._request_count += 1

        # Rate limit: stagger requests from this exit IP
        async with self._rate_lock:
            now = time.monotonic()
            elapsed = now - self._last_request
            if elapsed < self.min_interval:
                jitter = random.uniform(0, self.min_interval * 0.3)
                await asyncio.sleep(self.min_interval - elapsed + jitter)
            self._last_request = time.monotonic()

        return await self._get_session()

    async def handle_waf(self):
        """Force circuit rotation on WAF hit. Session recreated lazily."""
        async with self._rotation_lock:
            await self._rotate()
        # Don't close session here — other workers may have in-flight
        # requests on it. _get_session() will create a new one when it
        # sees the fingerprint has changed; old session gets GC'd.

    async def _get_session(self) -> AsyncSession:
        """Get or create the shared session. Creates new on fingerprint change."""
        async with self._session_lock:
            if self._session is None or self._session_fp != self._fingerprint:
                # Old session (if any) gets GC'd naturally — don't close it
                # as other workers may have in-flight requests on it
                self._session = AsyncSession(impersonate=self._fingerprint)
                self._session_fp = self._fingerprint
                # Enable HTTP/2 multiplexing on the multi handle
                try:
                    self._session.acurl.setopt(CurlMOpt.PIPELINING, 2)
                    self._session.acurl.setopt(CurlMOpt.MAX_HOST_CONNECTIONS, 1)
                    self._session.acurl.setopt(CurlMOpt.MAX_CONCURRENT_STREAMS, 100)
                except (AttributeError, Exception):
                    pass
            return self._session

    async def _rotate(self):
        self._fingerprint = random.choice(FINGERPRINTS)
        self._request_count = 0
        await asyncio.to_thread(self._sync_newnym)
        await asyncio.sleep(1.0 + random.uniform(0, 0.5))

    def _sync_newnym(self):
        try:
            with Controller.from_port(port=self.control_port) as c:
                c.authenticate()
                c.signal(Signal.NEWNYM)
        except Exception:
            time.sleep(1)

    async def close(self):
        """Close the session. Only call after all workers have finished."""
        async with self._session_lock:
            if self._session:
                await self._session.close()
                self._session = None


# == Page Parsers ==============================================================
# Consolidated from scrape_all_gaps_tor.py + wowhead_scraper_v2.py
# Includes all extractors: listview, g_npcs, g_mapperData, infobox

def _extract_listview_data(html: str, listview_id: str) -> list | None:
    """Extract data array from WH.Listview({id:'<id>', ...data:[...]})."""
    for quote in ["'", '"']:
        idx = html.find(f"id: {quote}{listview_id}{quote}")
        if idx < 0:
            idx = html.find(f"id:{quote}{listview_id}{quote}")
        if idx < 0:
            idx = html.find(f'"id":{quote}{listview_id}{quote}')
        if idx < 0:
            idx = html.find(f'"id": {quote}{listview_id}{quote}')
        if idx >= 0:
            break
    else:
        return None

    data_idx = html.find("data:", idx)
    if data_idx < 0 or data_idx > idx + 2000:
        return None

    bracket_start = html.find("[", data_idx)
    if bracket_start < 0 or bracket_start > data_idx + 50:
        return None

    depth = 0
    in_str = False
    esc = False
    for i in range(bracket_start, min(bracket_start + 500000, len(html))):
        ch = html[i]
        if esc:
            esc = False
            continue
        if ch == "\\":
            esc = True
            continue
        if ch == '"' and not in_str:
            in_str = True
            continue
        if ch == '"' and in_str:
            in_str = False
            continue
        if in_str:
            continue
        if ch == "[":
            depth += 1
        elif ch == "]":
            depth -= 1
            if depth == 0:
                raw = html[bracket_start:i + 1]
                raw = re.sub(r'(?<=[{,\[])\s*(\w+)\s*:', r'"\1":', raw)
                raw = re.sub(r",\s*([}\]])", r"\1", raw)
                try:
                    return json.loads(raw)
                except (json.JSONDecodeError, ValueError):
                    return None
    return None


def _extract_g_data(html: str, var_name: str) -> dict | None:
    """Extract $.extend(g_npcs[ID], {...}) or similar."""
    pattern = re.compile(
        rf'\$\.extend\({var_name}\[\d+\],\s*(\{{.*?\}})\)', re.DOTALL
    )
    m = pattern.search(html)
    if m:
        try:
            raw = m.group(1)
            raw = re.sub(r'(?<=[{,])\s*(\w+)\s*:', r'"\1":', raw)
            raw = re.sub(r",\s*([}\]])", r"\1", raw)
            return json.loads(raw)
        except (json.JSONDecodeError, ValueError):
            pass
    return None


def _extract_mapper_data(html: str) -> list | None:
    """Extract g_mapperData (spawn coordinates)."""
    idx = html.find("g_mapperData")
    if idx < 0:
        return None
    bracket = html.find("[", idx)
    if bracket < 0 or bracket > idx + 100:
        return None
    depth = 0
    for i in range(bracket, min(bracket + 200000, len(html))):
        if html[i] == "[":
            depth += 1
        elif html[i] == "]":
            depth -= 1
            if depth == 0:
                raw = html[bracket:i + 1]
                raw = re.sub(r'(?<=[{,\[])\s*(\w+)\s*:', r'"\1":', raw)
                raw = re.sub(r",\s*([}\]])", r"\1", raw)
                try:
                    return json.loads(raw)
                except (json.JSONDecodeError, ValueError):
                    return None
    return None


def parse_quest_page(html: str, entity_id: int) -> dict:
    result = {}

    for label, key_npc, key_go in [
        ("Start:", "start_npcs", "start_gos"),
        ("End:", "end_npcs", "end_gos"),
    ]:
        idx = html.find(label)
        if idx < 0:
            continue
        block = html[idx:idx + 2000]
        npcs = re.findall(r'\[url=/npc=(\d+)/', block)
        gos = re.findall(r'\[url=/object=(\d+)/', block)
        if not npcs and not gos:
            npcs = re.findall(r'npc=(\d+)', block)
            gos = re.findall(r'object=(\d+)', block)
        if npcs:
            result[key_npc] = [int(x) for x in npcs]
        if gos:
            result[key_go] = [int(x) for x in gos]

    if "start_npcs" not in result and "start_gos" not in result:
        for tag in ["Quest start", "Starts at", "Start"]:
            idx = html.find(tag)
            if idx >= 0:
                block = html[idx:idx + 2000]
                npcs = re.findall(r'npc=(\d+)', block)
                gos = re.findall(r'object=(\d+)', block)
                if npcs:
                    result["start_npcs"] = [int(x) for x in npcs]
                if gos:
                    result["start_gos"] = [int(x) for x in gos]
                break

    for section, key in [("Progress", "request_text"), ("Completion", "reward_text")]:
        pat = re.compile(
            rf'<h[23][^>]*>.*?{section}.*?</h[23]>\s*(.*?)(?=<h[23]|<div\s+class="pad"|$)',
            re.DOTALL | re.IGNORECASE,
        )
        m = pat.search(html)
        if m:
            text = re.sub(r'<[^>]+>', ' ', m.group(1)).strip()
            text = re.sub(r'\s+', ' ', text)
            text = re.sub(r'\[/?[a-z]+[^\]]*\]', '', text)
            if text and len(text) > 3:
                result[key] = text

    g_data = _extract_g_data(html, "g_quests")
    if g_data:
        for key in ("level", "reqlevel", "category", "side", "money", "xp"):
            if key in g_data:
                result[key] = g_data[key]

    return result


def parse_npc_page(html: str, entity_id: int) -> dict:
    result = {}

    g_data = _extract_g_data(html, "g_npcs")
    if g_data:
        for key in ("minlevel", "maxlevel", "react", "type", "classification",
                     "boss", "health", "displayId"):
            if key in g_data:
                result[key] = g_data[key]

    mapper = _extract_mapper_data(html)
    if mapper:
        coords = []
        for entry in mapper:
            if isinstance(entry, dict) and "coords" in entry:
                for coord in entry["coords"]:
                    if isinstance(coord, list) and len(coord) >= 2:
                        coords.append({
                            "x": coord[0], "y": coord[1],
                            "uiMapId": entry.get("uiMapId"),
                        })
        if coords:
            result["coords"] = coords

    sells = _extract_listview_data(html, "sells")
    if sells:
        items = []
        for item in sells:
            if isinstance(item, dict) and "id" in item:
                entry = {"id": item["id"]}
                if "cost" in item:
                    entry["cost"] = item["cost"]
                if "avail" in item:
                    entry["maxcount"] = item["avail"]
                items.append(entry)
        if items:
            result["vendor_items"] = items

    for lv_id in ["teaches-recipe", "teaches"]:
        teaches = _extract_listview_data(html, lv_id)
        if teaches:
            spells = []
            for item in teaches:
                if isinstance(item, dict) and "id" in item:
                    entry = {"id": item["id"]}
                    if "learnedat" in item:
                        entry["learnedat"] = item["learnedat"]
                    if "name" in item:
                        entry["name"] = item["name"]
                    spells.append(entry)
            if spells:
                result["teaches"] = spells
            break

    drops = _extract_listview_data(html, "drops")
    if drops:
        result["drops"] = [
            {"id": d["id"],
             **({"count": d["count"]} if "count" in d else {}),
             **({"drop_chance": d["pctstack"]} if "pctstack" in d else {})}
            for d in drops if isinstance(d, dict) and "id" in d
        ]

    abilities = _extract_listview_data(html, "abilities")
    if abilities:
        result["abilities"] = [
            {"id": a["id"], **({"name": a["name"]} if "name" in a else {})}
            for a in abilities if isinstance(a, dict) and "id" in a
        ]

    skinning = _extract_listview_data(html, "skinning")
    if skinning:
        result["skinning"] = [
            {"id": s["id"]} for s in skinning if isinstance(s, dict) and "id" in s
        ]

    pickpocket = _extract_listview_data(html, "pickpocketing")
    if pickpocket:
        result["pickpocketing"] = [
            {"id": p["id"]} for p in pickpocket if isinstance(p, dict) and "id" in p
        ]

    models = _extract_listview_data(html, "models")
    if models:
        result["models"] = [
            {"displayId": m.get("displayId", m.get("id"))}
            for m in models if isinstance(m, dict)
        ]
    else:
        model_ids = re.findall(r'(?:displayId|data-mv-display-id)["\s:=]+(\d+)', html)
        if model_ids:
            result["modelIds"] = [int(m) for m in model_ids[:5]]

    for lv_id, key in [("starts", "quests_started"), ("ends", "quests_ended")]:
        q_data = _extract_listview_data(html, lv_id)
        if q_data:
            ids = [q["id"] for q in q_data if isinstance(q, dict) and "id" in q]
            if ids:
                result[key] = ids

    sounds = _extract_listview_data(html, "sounds")
    if sounds:
        result["sounds"] = [
            {"id": s["id"], **({"name": s["name"]} if "name" in s else {})}
            for s in sounds if isinstance(s, dict) and "id" in s
        ]

    return result


def parse_item_page(html: str, entity_id: int) -> dict:
    result = {}
    for lv_id, key in [
        ("sold-by", "sold_by"), ("dropped-by", "dropped_by"),
        ("reward-from-q", "reward_from_quests"), ("starts", "starts_quests"),
        ("contained-in-object", "contained_in_objects"),
    ]:
        data = _extract_listview_data(html, lv_id)
        if data:
            result[key] = [e["id"] for e in data if isinstance(e, dict) and "id" in e]
    return result


def parse_spell_page(html: str, entity_id: int) -> dict:
    result = {}
    for lv_id, key in [
        ("taught-by-npc", "taught_by_npcs"), ("used-by-npc", "used_by_npcs"),
    ]:
        data = _extract_listview_data(html, lv_id)
        if data:
            result[key] = [e["id"] for e in data if isinstance(e, dict) and "id" in e]
    return result


def parse_object_page(html: str, entity_id: int) -> dict:
    result = {}

    g_data = _extract_g_data(html, "g_objects")
    if g_data:
        for key in ("type", "displayId"):
            if key in g_data:
                result[key] = g_data[key]

    mapper = _extract_mapper_data(html)
    if mapper:
        coords = []
        for entry in mapper:
            if isinstance(entry, dict) and "coords" in entry:
                for coord in entry["coords"]:
                    if isinstance(coord, list) and len(coord) >= 2:
                        coords.append({
                            "x": coord[0], "y": coord[1],
                            "uiMapId": entry.get("uiMapId"),
                        })
        if coords:
            result["coords"] = coords

    for lv_id, key in [
        ("starts", "quests_started"), ("ends", "quests_ended"),
        ("contains", "contains_items"),
    ]:
        data = _extract_listview_data(html, lv_id)
        if data:
            result[key] = [e["id"] for e in data if isinstance(e, dict) and "id" in e]
    return result


# Use comprehensive parsers from parsers.py (39 entity types, 1245 lines)
# Falls back to inline parsers above for the 7 core types if import fails
try:
    from parsers import PARSERS
except ImportError:
    PARSERS = {
        "quest": parse_quest_page,
        "npc": parse_npc_page,
        "trainer": parse_npc_page,
        "vendor": parse_npc_page,
        "item": parse_item_page,
        "spell": parse_spell_page,
        "object": parse_object_page,
    }


# == Work Items ================================================================

@dataclass(order=True)
class WorkItem:
    priority: int
    seq: int                                    # Tie-breaking (FIFO within priority)
    target: str = field(compare=False)
    item_id: int = field(compare=False)
    retries: int = field(default=0, compare=False)


# == Stats Tracker =============================================================

class StatsTracker:
    def __init__(self, total: int):
        self.total = total
        self.ok = 0
        self.waf = 0
        self.error = 0
        self.dropped = 0
        self.skip = 0
        self.notfound = 0
        self.by_target: dict[str, dict] = defaultdict(
            lambda: {"ok": 0, "waf": 0, "err": 0, "404": 0}
        )
        self.start_time = time.time()
        self._last_print = 0
        self._waf_times: list[float] = []
        self._lock = asyncio.Lock()
        self._seq = 0  # For re-queued items

    def next_seq(self) -> int:
        self._seq += 1
        return self._seq

    async def record(self, event: str, target: str):
        async with self._lock:
            if event == "ok":
                self.ok += 1
                self.by_target[target]["ok"] += 1
            elif event == "waf":
                self.waf += 1
                self.by_target[target]["waf"] += 1
                self._waf_times.append(time.time())
            elif event == "error":
                self.error += 1
                self.by_target[target]["err"] += 1
            elif event == "skip":
                self.skip += 1
            elif event == "notfound":
                self.notfound += 1
                self.by_target[target]["404"] += 1
            elif event == "dropped":
                self.dropped += 1

            if self.ok > 0 and self.ok % 500 == 0 and self.ok != self._last_print:
                self._last_print = self.ok
                self._print_progress()

    def _print_progress(self):
        elapsed = time.time() - self.start_time
        rate = self.ok / elapsed * 3600
        done = self.ok + self.skip + self.notfound + self.dropped
        remaining = self.total - done
        eta = (remaining / (self.ok / elapsed) / 60) if self.ok > 0 else 0

        parts = [f"{t}:{s['ok']}" for t, s in sorted(self.by_target.items())]
        waf_min = len([t for t in self._waf_times if t > time.time() - 60])

        print(f"  [{self.ok:>7,}/{self.total:,}] "
              f"rate={rate:,.0f}/hr waf={self.waf}({waf_min}/min) err={self.error} "
              f"ETA={eta:.1f}m | {' '.join(parts)}")

    def get_adaptive_delay(self, base_delay: float) -> float:
        cutoff = time.time() - 60
        self._waf_times = [t for t in self._waf_times if t > cutoff]
        recent = len(self._waf_times)

        if recent == 0 and self.ok > 500:
            return max(base_delay * 0.3, 0.04)   # Aggressive when clean
        elif recent < 5:
            return max(base_delay * 0.5, 0.06)   # Near-clean — stay fast
        elif recent < 15:
            return base_delay                      # Normal
        elif recent < 30:
            return base_delay * 1.5                # Mild backoff
        else:
            return base_delay * 2.5                # Moderate backoff

    def summary(self) -> str:
        elapsed = time.time() - self.start_time
        rate = self.ok / max(elapsed, 1) * 3600
        lines = [
            f"\n{'=' * 70}",
            f"SCRAPE COMPLETE in {elapsed:.0f}s ({elapsed / 60:.1f} min)",
            f"  Total OK:      {self.ok:>7,}  ({rate:,.0f}/hr)",
            f"  Total WAF:     {self.waf:>7,}",
            f"  Total errors:  {self.error:>7,}",
            f"  Total 404:     {self.notfound:>7,}",
            f"  Dropped:       {self.dropped:>7,}  (exceeded {MAX_RETRIES} retries)",
            f"  Skipped:       {self.skip:>7,}",
            "",
        ]
        for t in sorted(self.by_target.keys()):
            s = self.by_target[t]
            lines.append(f"  {t:10s}: ok={s['ok']:>7,} waf={s['waf']} "
                         f"err={s['err']} 404={s['404']}")
        return "\n".join(lines)


# == Async Worker ==============================================================

async def async_worker(
    worker_id: int,
    tor: TorInstance,
    work_queue: asyncio.PriorityQueue,
    base_delay: float,
    stats: StatsTracker,
    cache_html: bool,
):
    """Async worker coroutine. Multiple workers share one TorInstance.

    The TorInstance provides a shared AsyncSession — all workers on the same
    Tor instance send HTTP/2 streams on a single multiplexed connection.
    """
    consecutive_errors = 0

    while True:
        # Get next work item
        try:
            work_item: WorkItem = work_queue.get_nowait()
        except asyncio.QueueEmpty:
            break

        target = work_item.target
        item_id = work_item.item_id
        cfg = TARGET_CONFIGS[target]

        out_dir = WAGO_DIR / cfg["out_dir"]
        out_file = out_dir / cfg["out_pattern"].format(id=item_id)

        # Already done
        if out_file.exists():
            await stats.record("skip", target)
            continue

        # Max retries exceeded
        if work_item.retries >= MAX_RETRIES:
            await stats.record("dropped", target)
            continue

        # Acquire rate-limited slot + shared HTTP/2 session
        session = await tor.acquire()

        # Adaptive delay with jitter
        delay = stats.get_adaptive_delay(base_delay)
        await asyncio.sleep(delay + random.uniform(0, delay * 0.15))

        url = cfg["url"].format(id=item_id)

        try:
            r = await session.get(
                url, timeout=20,
                proxies={"https": tor.proxy, "http": tor.proxy},
            )
        except Exception:
            await stats.record("error", target)
            consecutive_errors += 1
            # Jittered exponential backoff on consecutive errors
            if consecutive_errors > 3:
                backoff = min(2 ** (consecutive_errors - 3) * 0.5, 30)
                await asyncio.sleep(backoff + random.uniform(0, 1))
            await work_queue.put(WorkItem(
                work_item.priority + 10, stats.next_seq(),
                target, item_id, work_item.retries + 1,
            ))
            continue

        consecutive_errors = 0

        # WAF / Cloudflare challenge
        if r.status_code == 403 or (
            len(r.text) < 5000 and "cf-challenge" in r.text.lower()
        ):
            await stats.record("waf", target)
            await work_queue.put(WorkItem(
                work_item.priority + 5, stats.next_seq(),
                target, item_id, work_item.retries + 1,
            ))
            await tor.handle_waf()
            continue

        if r.status_code == 404:
            await stats.record("notfound", target)
            await asyncio.to_thread(out_file.write_text, "{}", "utf-8")
            continue

        if r.status_code != 200:
            await stats.record("error", target)
            continue

        html_text = r.text

        # Cache raw HTML (gzip, fast compression)
        if cache_html:
            html_dir = WAGO_DIR / cfg["html_dir"]
            html_file = html_dir / cfg["html_pattern"].format(id=item_id)
            if not html_file.exists():
                html_dir.mkdir(parents=True, exist_ok=True)
                await asyncio.to_thread(
                    _write_gzip, html_file, html_text.encode("utf-8")
                )

        # Parse
        parser = PARSERS.get(target, parse_npc_page)
        data = parser(html_text, item_id)

        await asyncio.to_thread(
            out_file.write_text,
            json.dumps(data, ensure_ascii=False, indent=1),
            "utf-8",
        )
        await stats.record("ok", target)


def _write_gzip(path: Path, data: bytes):
    with gzip.open(path, "wb", compresslevel=1) as f:
        f.write(data)


# == Reparse Mode ==============================================================

def reparse(targets: list[str]):
    """Re-extract data from cached HTML files (offline, no network)."""
    print("REPARSE MODE -- extracting from cached HTML\n")

    for target in targets:
        cfg = TARGET_CONFIGS[target]
        html_dir = WAGO_DIR / cfg["html_dir"]
        out_dir = WAGO_DIR / cfg["out_dir"]
        out_dir.mkdir(parents=True, exist_ok=True)
        parser = PARSERS.get(target, parse_npc_page)

        if not html_dir.exists():
            print(f"  {target}: no HTML cache at {html_dir}")
            continue

        html_files = list(html_dir.glob("*.html.gz"))
        ok = 0
        for hf in html_files:
            try:
                item_id = int(hf.stem.split(".")[0])
            except ValueError:
                continue

            out_file = out_dir / cfg["out_pattern"].format(id=item_id)

            with gzip.open(hf, "rb") as f:
                html = f.read().decode("utf-8", errors="replace")

            data = parser(html, item_id)
            out_file.write_text(
                json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8",
            )
            ok += 1

        print(f"  {target}: reparsed {ok}/{len(html_files)} cached pages")


# == ID Loading ================================================================

def load_ids(target: str) -> list[int]:
    cfg = TARGET_CONFIGS.get(target)
    if not cfg:
        return []
    id_file = WAGO_DIR / cfg["ids_file"]
    if not id_file.exists():
        return []
    ids = []
    for line in id_file.read_text().splitlines():
        line = line.strip()
        if line and line.split()[0].isdigit():
            ids.append(int(line.split()[0]))
    return ids


def count_done(target: str) -> int:
    cfg = TARGET_CONFIGS.get(target)
    if not cfg:
        return 0
    out_dir = WAGO_DIR / cfg["out_dir"]
    if not out_dir.exists():
        return 0
    pattern = cfg["out_pattern"].replace("{id}", "*")
    return len(list(out_dir.glob(pattern)))


# == Main ======================================================================

async def async_main(args):
    if args.targets.strip().lower() == "all":
        targets = sorted(TARGET_CONFIGS.keys())
    else:
        targets = [t.strip() for t in args.targets.split(",")]

    for t in targets:
        (WAGO_DIR / TARGET_CONFIGS[t]["out_dir"]).mkdir(parents=True, exist_ok=True)

    # Build work queue
    all_items = []
    seq = 0
    for t in targets:
        ids = load_ids(t)
        if not ids:
            continue
        priority = 0 if len(ids) < 2000 else 1
        for item_id in ids:
            all_items.append(WorkItem(priority, seq, t, item_id))
            seq += 1

    random.shuffle(all_items)
    all_items.sort(key=lambda x: x.priority)

    # Reassign sequence after shuffle for stable ordering
    for i, item in enumerate(all_items):
        item.seq = i

    work_queue: asyncio.PriorityQueue[WorkItem] = asyncio.PriorityQueue()
    for item in all_items:
        await work_queue.put(item)

    # Apply smoke test limit
    if args.smoke and args.smoke > 0:
        trimmed: asyncio.PriorityQueue[WorkItem] = asyncio.PriorityQueue()
        for _ in range(min(args.smoke, work_queue.qsize())):
            await trimmed.put(work_queue.get_nowait())
        work_queue = trimmed

    total = work_queue.qsize()
    eff = args.workers * args.multiplier

    print(f"Tor Army v3 -- Async Swarm (HTTP/2 Multiplexed)")
    print(f"  Tor instances:     {args.workers}")
    print(f"  Workers/instance:  {args.multiplier}")
    print(f"  Effective workers: {eff}")
    print(f"  HTTP/2 connections:{args.workers} (multiplexed, {eff} streams)")
    print(f"  Targets:           {', '.join(targets)}")
    print(f"  Total pages:       {total:,}")
    print(f"  Base delay:        {args.delay}s (adaptive)")
    print(f"  Per-circuit:       {args.per_circuit}")
    print(f"  Cache HTML:        {args.cache_html}")

    for t in targets:
        ids = load_ids(t)
        done = count_done(t)
        print(f"    {t:10s}: {len(ids):>7,} total  {done:>7,} done  "
              f"{len(ids) - done:>7,} remaining")

    # Start Tor fleet
    fleet = None
    if args.start_tor:
        print(f"\nLaunching Tor fleet ({args.workers} instances w/ optimized torrc)...")
        fleet = TorFleet(args.workers)
        fleet.start()

    # Create TorInstances + workers
    stats = StatsTracker(total)
    min_interval = args.delay / args.multiplier
    tasks = []
    tor_instances = []

    print(f"\nLaunching {eff} async workers ({args.workers} HTTP/2 connections)...")

    for tor_id in range(args.workers):
        socks_port = 9050 + tor_id * 2
        control_port = 9051 + tor_id * 2

        tor = TorInstance(
            instance_id=tor_id,
            socks_port=socks_port,
            control_port=control_port,
            per_circuit=args.per_circuit,
            min_interval=min_interval,
        )
        tor_instances.append(tor)

        for mux_id in range(args.multiplier):
            wid = tor_id * args.multiplier + mux_id
            tasks.append(asyncio.create_task(
                async_worker(
                    worker_id=wid,
                    tor=tor,
                    work_queue=work_queue,
                    base_delay=args.delay,
                    stats=stats,
                    cache_html=args.cache_html,
                )
            ))

            # Stagger launches to avoid thundering herd
            if len(tasks) % 50 == 0:
                await asyncio.sleep(0.3)

    await asyncio.gather(*tasks)
    print(stats.summary())

    # Cleanup shared sessions
    for tor in tor_instances:
        await tor.close()

    if fleet:
        fleet.stop()


def main():
    parser = argparse.ArgumentParser(description="Tor Army v3 -- Async Swarm Scraper")

    parser.add_argument("--start-tor", action="store_true",
                        help="Start Tor instances before scraping")
    parser.add_argument("--kill-tor", action="store_true",
                        help="Kill all existing Tor instances and exit")
    parser.add_argument("--tor-dir", type=str, default=None,
                        help="Path to Tor Expert Bundle (or set TOR_DIR env var)")

    parser.add_argument("--workers", type=int, default=400,
                        help="Tor instances to run (default: 400)")
    parser.add_argument("--multiplier", type=int, default=5,
                        help="Async workers per Tor instance (default: 5)")

    parser.add_argument("--targets", type=str, default="npc",
                        help="Comma-separated targets, or 'all' for all 39 types")
    parser.add_argument("--per-circuit", type=int, default=150,
                        help="Requests per circuit before rotation (default: 150)")
    parser.add_argument("--delay", type=float, default=0.15,
                        help="Base delay between requests in seconds (default: 0.15)")

    parser.add_argument("--cache-html", action="store_true", default=True,
                        help="Cache raw HTML as gzip (default: on)")
    parser.add_argument("--no-cache-html", action="store_false", dest="cache_html")

    parser.add_argument("--reparse", action="store_true",
                        help="Re-extract from cached HTML (no network)")
    parser.add_argument("--list-targets", action="store_true",
                        help="Show available targets with counts and exit")
    parser.add_argument("--smoke", type=int, default=0,
                        help="Smoke test: limit to first N items")

    args = parser.parse_args()

    # Override TOR_DIR if --tor-dir provided
    if args.tor_dir:
        global TOR_DIR
        TOR_DIR = Path(args.tor_dir)

    if args.kill_tor:
        TorFleet.kill_all()
        print("All Tor instances killed.")
        return

    if args.list_targets:
        print("Target Status:")
        total_ids = 0
        total_done = 0
        for t in sorted(TARGET_CONFIGS.keys()):
            ids = load_ids(t)
            done = count_done(t)
            remain = len(ids) - done
            total_ids += len(ids)
            total_done += done
            exists = "YES" if (WAGO_DIR / TARGET_CONFIGS[t]["ids_file"]).exists() else " NO"
            print(f"  {t:25s}: {len(ids):>9,} total  {done:>9,} done  "
                  f"{remain:>9,} remain  (file: {exists})")
        print(f"\n  {'TOTAL':25s}: {total_ids:>9,} total  {total_done:>9,} done  "
              f"{total_ids - total_done:>9,} remain")
        return

    if args.reparse:
        if args.targets.strip().lower() == "all":
            targets = sorted(TARGET_CONFIGS.keys())
        else:
            targets = [t.strip() for t in args.targets.split(",")]
        reparse(targets)
        return

    asyncio.run(async_main(args))


if __name__ == "__main__":
    main()