Des proxys pour outils d'IA et des appels d'API fluides

#!/usr/bin/env python3
"""Rotating-proxy client for OpenAI-compatible chat APIs (Advanced.name proxies).

Sends each prompt from a file to a chat API through a different proxy, with
retries, and writes the answers to a CSV. Use it only with an API key and a
service you are authorised for. Standard library only. Run: python3 ai_calls.py
"""

# ================================ CONFIG ====================================
CSV_URL = "https://get.advanced.name/YOUR_LIST_ID.csv"  # CSV export link from your dashboard
COUNTRY = ""            # ISO code where the service is available, or "" for the whole pool
MIN_SPEED = 3
API_URL = "https://api.openai.com/v1/chat/completions"  # OpenAI-compatible endpoint you may use
API_KEY = "YOUR_API_KEY"    # your own key; it stays in this file, never in the proxy
MODEL = "gpt-4o-mini"       # model name the endpoint expects
PROMPTS_FILE = "prompts.txt"     # one user prompt per line
OUT_FILE = "answers.csv"         # prompt;answer;proxy is written here
THREADS = 20
TIMEOUT = 30
RETRIES = 3
# ============================================================================

import csv
import io
import itertools
import json
import sys
import threading
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

COL_IP, COL_PORT, COL_COUNTRY, COL_SPEED = 0, 1, 2, 3


def load_proxies():
    raw = urllib.request.urlopen(CSV_URL, timeout=30).read().decode("utf-8", "replace")
    pool = []
    for row in csv.reader(io.StringIO(raw), delimiter=";"):
        if len(row) <= COL_SPEED or not row[COL_PORT].strip().isdigit():
            continue
        if COUNTRY and row[COL_COUNTRY].strip().upper() != COUNTRY.upper():
            continue
        if int(row[COL_SPEED] or 0) < MIN_SPEED:
            continue
        pool.append(row[COL_IP].strip() + ":" + row[COL_PORT].strip())
    return pool


def ask(prompt, proxy):
    body = json.dumps({"model": MODEL, "messages": [{"role": "user", "content": prompt}]}).encode()
    handler = urllib.request.ProxyHandler({"http": "http://" + proxy, "https": "http://" + proxy})
    req = urllib.request.Request(API_URL, data=body, method="POST", headers={
        "Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"})
    data = json.loads(urllib.request.build_opener(handler).open(req, timeout=TIMEOUT).read())
    return data["choices"][0]["message"]["content"].strip()


def main():
    proxies = load_proxies()
    if not proxies:
        sys.exit("No proxies for COUNTRY=%r." % COUNTRY)
    with open(PROMPTS_FILE, encoding="utf-8") as fh:
        prompts = [line.strip() for line in fh if line.strip() and not line.startswith("#")]
    print("proxies: %d (%s), prompts: %d, threads: %d" % (len(proxies), COUNTRY or "any", len(prompts), THREADS))

    rotation = itertools.cycle(proxies)
    lock = threading.Lock()

    def run(prompt):
        for _ in range(RETRIES):
            with lock:
                proxy = next(rotation)
            try:
                return prompt, ask(prompt, proxy), proxy
            except Exception:
                continue
        return prompt, "error: unreachable", ""

    rows = []
    with ThreadPoolExecutor(max_workers=THREADS) as pool:
        for fut in as_completed([pool.submit(run, p) for p in prompts]):
            prompt, answer, proxy = fut.result()
            rows.append((prompt, answer, proxy))
            print("%-40s %s" % (prompt[:40], (answer[:50] + "...") if len(answer) > 50 else answer))

    with open(OUT_FILE, "w", encoding="utf-8", newline="") as out:
        writer = csv.writer(out, delimiter=";")
        writer.writerow(["prompt", "answer", "proxy"])
        writer.writerows(rows)
    print("written %s (%d rows)" % (OUT_FILE, len(rows)))


if __name__ == "__main__":
    main()