find_ip.py

import os
import time
import requests
import json
import asyncio
import re
import argparse
import subprocess
from typing import Dict, Any, Optional
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Copying logic from main.py to make find_ip.py self-contained
HTTP_PROXY = os.getenv("HTTP_PROXY")
HTTPS_PROXY = os.getenv("HTTPS_PROXY")

def get_request_proxies():
    proxies = {}
    if HTTP_PROXY:
        proxies["http"] = HTTP_PROXY
    if HTTPS_PROXY:
        proxies["https"] = HTTPS_PROXY
    return proxies if proxies else None

def webshare_replace_proxy(
    token: str,
    plan_id: str,
    socks_username: str,
    socks_password: str,
    asn: Optional[int] = None,
    ip_ranges: Optional[list[str]] = None,
    wait_seconds: int = 5
) -> Dict[str, Any] | None:
    headers = {
        "authorization": f"Token {token}",
        "accept": "application/json",
        "content-type": "application/json",
        "user-agent": "Mozilla/5.0"
    }
    current_proxies = get_request_proxies()
    cookies = {"ssotoken": token}

    def get_latest_proxy():
        url = f"https://proxy.webshare.io/api/v2/proxy/list/replaced/?page=1&page_size=1&plan_id={plan_id}"
        try:
            response = requests.get(url, headers=headers, cookies=cookies, proxies=current_proxies, timeout=30)
            if response.status_code == 200:
                results = response.json().get("results")
                if results:
                    return results[0].get("replaced_with"), results[0].get("replaced_with_port")
        except Exception:
            pass
        return None, None

    def replace(ip):
        url = f"https://proxy.webshare.io/api/v2/proxy/replace/?plan_id={plan_id}"

        # 优先级：如果指定了 ip_ranges，则使用 ip_range；否则使用 asn
        if ip_ranges:
            replace_with = [{"type": "ip_range", "ip_ranges": ip_ranges}]
        else:
            replace_with = [{"type": "asn", "asn_numbers": [asn] if asn else [6079]}]

        data = {
            "to_replace": {"type": "ip_address", "ip_addresses": [ip]},
            "replace_with": replace_with,
            "dry_run": False
        }
        try:
            response = requests.post(url, headers=headers, cookies=cookies, json=data, proxies=current_proxies, timeout=30)
            return response.status_code in [200, 201]
        except Exception:
            return False

    current_ip, _ = get_latest_proxy()
    if not current_ip:
        return None

    if not replace(current_ip):
        return None

    time.sleep(wait_seconds)

    new_ip, new_port = None, None
    for _ in range(5):
        new_ip, new_port = get_latest_proxy()
        if new_ip and new_ip != current_ip:
            break
        time.sleep(3)

    if not new_ip or new_ip == current_ip:
        return None

    socks_url = f"socks5://{socks_username}:{socks_password}@{new_ip}:{new_port}"
    socks_cmd = f"bash <(curl -Ls IP.Check.Place) -x {socks_url}"

    return {
        "cmd": socks_cmd,
        "ip": new_ip,
        "port": new_port,
        "socks_url": socks_url
    }

async def replace_ip_and_check_logic(asn: Optional[int] = None, ip_ranges: Optional[list[str]] = None) -> str:
    token = os.getenv("WEBSHARE_TOKEN")
    plan_id = os.getenv("WEBSHARE_PLAN_ID")
    socks_username = os.getenv("WEBSHARE_SOCKS_USERNAME")
    socks_password = os.getenv("WEBSHARE_SOCKS_PASSWORD")

    if not all([token, plan_id, socks_username, socks_password]):
        return json.dumps({"status": "failed", "message": "Missing environment variables"})

    res = webshare_replace_proxy(token, plan_id, socks_username, socks_password, asn=asn, ip_ranges=ip_ranges)
    if not res:
        return json.dumps({"status": "failed", "message": "Failed to replace IP"})

    ip = res["ip"]
    cmd = res["cmd"]

    try:
        process = subprocess.run(cmd, shell=True, capture_output=True, text=True, executable='/bin/bash', timeout=300)
        report_text = process.stdout
    except Exception as e:
        return json.dumps({"status": "failed", "message": str(e)})

    return json.dumps({
        "status": "success",
        "ip": ip,
        "socks_url": res["socks_url"],
        "report": report_text
    }, ensure_ascii=False)

def check_quality(report: str, min_low_risk: int = 6, max_commercial: int = 2) -> tuple[bool, int, int]:
    # 移除 ANSI 转义序列以便正确匹配
    clean_report = re.sub(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])', '', report)

    low_risk_count = len(re.findall(r"低风险", clean_report))
    business_count = len(re.findall(r"商业", clean_report))

    is_good = low_risk_count >= min_low_risk and business_count <= max_commercial
    return is_good, low_risk_count, business_count

def save_report(ip: str, report: str):
    filename = f"{ip}-report.log"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(report)
    print(f"Report saved to {filename}")

async def main():
    parser = argparse.ArgumentParser(description="Find High Quality IP")
    parser.add_argument("--asn", type=int, help="ASN number (default: 6079 if no ip-range)")
    parser.add_argument("--ip-range", type=str, help="IP range to replace with (e.g. 45.0.0.0/8)")
    parser.add_argument("--max-tries", type=int, default=50, help="Maximum number of tries (default: 50)")
    parser.add_argument("--min-low-risk", type=int, default=6, help="Minimum 'low risk' count (default: 6)")
    parser.add_argument("--max-commercial", type=int, default=2, help="Maximum 'commercial' count (default: 2)")
    args = parser.parse_args()

    asn = args.asn
    ip_range = args.ip_range
    ip_ranges = [ip_range] if ip_range else None

    # If neither is provided, default to ASN 6079
    if not asn and not ip_ranges:
        asn = 6079

    max_tries = args.max_tries
    min_low_risk = args.min_low_risk
    max_commercial = args.max_commercial

    # 1. Removed Pre-check test-report.log
    print(f"Starting search for high quality IP. ASN: {asn}, Range: {ip_range}, Max Tries: {max_tries}")
    print(f"Thresholds -> Min Low Risk: {min_low_risk}, Max Commercial: {max_commercial}")

    for i in range(1, max_tries + 1):
        print(f"\n[Attempt {i}/{max_tries}] Replacing IP and checking quality...")

        try:
            result_json = await replace_ip_and_check_logic(asn=asn, ip_ranges=ip_ranges)
            result = json.loads(result_json)
        except Exception as e:
            print(f"Unexpected error: {e}")
            continue

        if result.get("status") != "success":
            print(f"Failed: {result.get('message')}")
            continue

        ip = result.get("ip")
        report = result.get("report", "")
        socks_url = result.get("socks_url")

        # Save report for every attempt
        save_report(ip, report)

        is_good, low_risk, business = check_quality(report, min_low_risk, max_commercial)
        print(f"IP: {ip}")
        print(f"Counts -> 低风险: {low_risk} (Target >= {min_low_risk}), 商业: {business} (Target <= {max_commercial})")

        if is_good:
            print(f"\n✨ SUCCESS! High quality IP found: {ip}")
            print(f"SOCKS5 URL: {socks_url}")
            print(f"Final Counts -> 低风险: {low_risk}, 商业: {business}")

            with open("found_ip.json", "w", encoding="utf-8") as f:
                json.dump(result, f, indent=2, ensure_ascii=False)
            return

        print("Criteria not met. Continuing...")

    print(f"\n❌ Finished {max_tries} attempts without finding a suitable IP.")

if __name__ == "__main__":
    asyncio.run(main())