from crucible_proxy import Session

session = Session(
    types         = ["http", "socks5"],   # which protocols
    workers       = 300,                   # concurrency
    timeout       = 5,                     # per-request seconds
    verify_twice  = False,                 # second-pass confirmation
    source_timeout= 10,                    # seconds per source URL
    max_sources   = 5,                     # cap sources per type (None = all)

    # filters
    elite         = True,                  # ELITE only
    max_latency   = 1000,                  # drop slow proxies (ms)
    min_score     = 70,                    # quality score 0-100
    countries     = ["US", "DE", "TR"],    # allow-list
    exclude_ports = [8080],                # deny-list

    # output
    save          = True,
    output_format = "json",                # txt | json | csv
    output_dir    = "./proxies",
)

results = session.run()
# results -> {"http": [CheckResult, ...], "socks5": [...]}

for proto, proxies in results.items():
    print(f"{proto}: {len(proxies)} proxies passed the filters")

from crucible_proxy import Session

# From a dict
session = Session.from_dict({
    "types":       ["socks5"],
    "workers":     200,
    "elite":       True,
    "max_latency": 1500,
})

# From a JSON config file
session = Session.from_json("proxy_config.json")

# Save the current config back to JSON
session.save_config("proxy_config.json")

report = session.run_with_stats()
for proto, info in report.items():
    print(f"{proto}: {info['alive']}/{info['total']} alive, "
          f"{info['matched']} matched filters, {info['elapsed_s']}s")
    for s in info["source_stats"]:
        print(f"  {s.domain}: +{s.proxies_found}" if s.success else f"  {s.domain}: FAIL")

session = Session(types=["http", "socks5"], elite=True, max_latency=1000)

flat    = session.live_only()   # [CheckResult, ...] across all types
strings = session.to_list()     # ["1.2.3.4:8080", ...]
top10   = session.best(10)      # top 10 by quality score (0-100)
fast10  = session.fastest(10)   # 10 lowest-latency proxies

def on_alive(r):
    print(f"LIVE {r.proxy}  {r.country}/{r.city}  {r.latency_ms}ms  {r.anonymity.value}")

def on_progress(checked, total):
    print(f"\r{checked}/{total}", end="")

session = Session(
    types       = ["http"],
    on_alive    = on_alive,
    on_progress = on_progress,
)
session.run()

import asyncio
from crucible_proxy import Session

async def main():
    results = await Session(types=["http"]).run_async()
    print(sum(len(v) for v in results.values()), "proxies found")

asyncio.run(main())

session = Session(types=["socks5"], elite=True)
session.run()

session.export("clash",       "proxies-clash.yaml")   # Clash YAML
session.export("v2ray",       "proxies-v2ray.json")   # v2ray outbounds JSON
session.export("shadowsocks", "proxies-ss.txt")       # ss:// list
session.export("txt",         "proxies.txt")          # plain host:port

# Or get the string directly (no path)
yaml_str = session.export("clash")

session = Session(
    types        = ["http"],
    countries    = ["DE", "US"],
    cities       = ["Berlin", "Frankfurt"],   # substring match
    exclude_asn  = ["AS9999"],                # drop specific networks
)

from crucible_proxy import fetch_proxies, check_all, ProxyType

# Fetch from all sources for one protocol
proxies = fetch_proxies(ProxyType.SOCKS5)
print(f"Found {len(proxies)} proxies")

# Validate concurrently; results stream as they complete
for result in check_all(proxies, max_workers=200, timeout=8):
    if result.alive:
        print(
            f"{result.proxy}  "
            f"{result.latency_ms}ms  "
            f"{result.country}  "
            f"{result.anonymity.value}"
        )

from crucible_proxy import load_config, fetch_proxies, check_all, ProxyType

cfg = load_config({
    "workers":       300,
    "timeout":       6,
    "verify_twice":  False,
    "output_format": "json",
    "output_dir":    "./results",
    "log_level":     "INFO",
    "debug_report":  True,
})

proxies = fetch_proxies(ProxyType.HTTP)
results = list(check_all(proxies, max_workers=cfg.workers, timeout=cfg.timeout))

import asyncio
from concurrent.futures import ThreadPoolExecutor
from crucible_proxy import fetch_proxies, check_all, ProxyType

async def get_alive_proxies(proto: ProxyType, workers: int = 200):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=4) as pool:
        proxies = await loop.run_in_executor(pool, fetch_proxies, proto)
        results = await loop.run_in_executor(
            pool,
            lambda: list(check_all(proxies, max_workers=workers))
        )
    return [r for r in results if r.alive]

alive = asyncio.run(get_alive_proxies(ProxyType.SOCKS5))

import itertools
from crucible_proxy import fetch_proxies, check_all, ProxyType

proxies  = fetch_proxies(ProxyType.HTTP)
alive    = [r.proxy for r in check_all(proxies) if r.alive]
pool     = itertools.cycle(alive)

def next_proxy() -> dict:
    """Returns a requests-compatible proxy dict."""
    return next(pool).as_requests_dict()

# Use it:
import requests
for url in my_urls:
    resp = requests.get(url, proxies=next_proxy(), timeout=10)

import requests
from crucible_proxy import fetch_proxies, check_all, ProxyType, Anonymity

# Get elite-only proxies
proxies = fetch_proxies(ProxyType.HTTPS)
elite = [
    r.proxy for r in check_all(proxies)
    if r.alive and r.anonymity == Anonymity.ELITE
]

proxy = elite[0]
resp  = requests.get("https://httpbin.org/ip", proxies=proxy.as_requests_dict(), timeout=10)
print(resp.json())

import asyncio
from playwright.async_api import async_playwright
from crucible_proxy import fetch_proxies, check_all, ProxyType

async def scrape(url: str):
    proxies = fetch_proxies(ProxyType.SOCKS5)
    alive   = [r.proxy for r in check_all(proxies, max_workers=100) if r.alive]
    proxy   = alive[0]

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            proxy={
                "server":   f"socks5://{proxy.host}:{proxy.port}",
            }
        )
        page = await browser.new_page()
        await page.goto(url)
        content = await page.content()
        await browser.close()
    return content

asyncio.run(scrape("https://example.com"))

from crucible_proxy import fetch_proxies, check_all, ProxyType, Anonymity

results = list(check_all(fetch_proxies(ProxyType.SOCKS5)))

# Fast proxies only (< 1 second latency)
fast = [r for r in results if r.alive and r.latency_ms and r.latency_ms < 1000]

# Only US proxies
us = [r for r in results if r.alive and r.country == "US"]

# Elite only (no headers leaked)
elite = [r for r in results if r.alive and r.anonymity == Anonymity.ELITE]

# Elite US proxies sorted by speed
best = sorted(
    [r for r in results if r.alive and r.country == "US" and r.anonymity == Anonymity.ELITE],
    key=lambda r: r.latency_ms or float("inf"),
)
print(f"Best US elite proxy: {best[0].proxy} @ {best[0].latency_ms}ms")

from pathlib import Path
from crucible_proxy import fetch_proxies, check_all, save_results, ProxyType

results = list(check_all(fetch_proxies(ProxyType.SOCKS5)))

# TXT (default)
simple, detailed = save_results(results, ProxyType.SOCKS5)

# JSON
json_path, _ = save_results(results, ProxyType.SOCKS5, fmt="json", output_dir=Path("./out"))

# CSV
csv_path, _  = save_results(results, ProxyType.SOCKS5, fmt="csv",  output_dir=Path("./out"))

import time
from crucible_proxy import fetch_proxies, check_all, write_debug_report, ProxyType
from pathlib import Path

proxies = fetch_proxies(ProxyType.HTTP)
t0      = time.perf_counter()
results = list(check_all(proxies))
elapsed = time.perf_counter() - t0

report_path = write_debug_report(results, ProxyType.HTTP, elapsed, Path("./debug"))
print(f"Debug report: {report_path}")
# JSON contains: meta stats, all alive proxies, all dead proxies + error messages

from crucible_proxy import ProxyCache, fetch_proxies, check_all, ProxyType

cache = ProxyCache()

# First run — checks all proxies over the network
proxies = fetch_proxies(ProxyType.HTTP)
results = list(check_all(proxies, cache=cache))
print(f"Checked {len(proxies)} proxies, cache now has {len(cache)} entries")

# Second run with the same list — cached proxies returned instantly
results2 = list(check_all(proxies, cache=cache))
# No network calls made for already-checked proxies

# Clear the cache when you want fresh results
cache.clear()

from crucible_proxy import fetch_proxies_with_stats, ProxyType

proxies, stats = fetch_proxies_with_stats(ProxyType.HTTP)

for s in stats:
    status = f"+{s.proxies_found}" if s.success else f"FAIL: {s.error[:40]}"
    print(f"{s.domain:<45} {status}  ({s.elapsed_s}s)")

# Filter out dead sources for next run
good_sources = [s.url for s in stats if s.success and s.proxies_found > 0]

from crucible_proxy.constants import ANONYMITY_CHECK_URLS
print(ANONYMITY_CHECK_URLS)
# ['http://httpbin.org/headers', 'https://httpbin.org/headers',
#  'http://httpbingo.org/headers', 'https://httpbingo.org/headers',
#  'http://eu.httpbin.org/headers']