コード統合

import json
import threading
import time
import requests

'''
まず外部IPをホワイトリストに追加してください
'''

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:85.0) Gecko/20100101 Firefox/85.0",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
    "Accept-Encoding": "gzip, deflate, br"
}

# (http) ビジネスロジック 
def execute_http(proxy_addr):
    targetUrl = 'http://ip123.in/ip.json'

    proxies = {
        'http': 'http://{}'.format(proxy_addr),
        'https': 'http://{}'.format(proxy_addr),
    }
    res = ""

    while True:
        try:
            res = requests.get(targetUrl, proxies=proxies, timeout=5)
            # print(res.status_code)
            print(res.status_code, "***", res.text)
            break
        except Exception as e:
            print(e)
            break
    return

# (socks5) ビジネスロジック 
def execute_socks5(proxy_addr):
    targetUrl = 'http://ip123.in/ip.json'

    proxies = {
        'http': 'socks5h://{}'.format(proxy_addr),
        'https': 'socks5h://{}'.format(proxy_addr),
    }
    res = ""

    while True:
        try:
            res = requests.get(targetUrl, proxies=proxies, timeout=5)
            # print(res.status_code)
            print(res.status_code, "***", res.text)
            break
        except Exception as e:
            print(e)
            break
    return

class ThreadFactory(threading.Thread):
    def __init__(self, proxy_addr,proxy_type):
        threading.Thread.__init__(self)
        self.proxy_addr = proxy_addr
        self.proxy_type = proxy_type
    def run(self):
        if self.proxy_type == "socks5":
            execute_socks5(self.proxy_addr)
        else:
            execute_http(self.proxy_addr)

def test_http():

    # 一度に5つ取得 (text形式)
    get_proxy_url = 'http://api2.rola.vip/user_get_ip_list?token=TOKEN&qty=5&country=&state=&city=&time=5&format=txt&protocol=http&filter=1'

    while True:
        try:
            resp = requests.get(url=get_proxy_url, timeout=5)
            if resp.status_code != 200:
                print("IP取得失敗")
                time.sleep(1)
                continue
            ips = resp.text.split("\r\n")
            threads = []
            for ip in ips:
                threads.append(ThreadFactory(ip,"http"))
            for t in threads:  
                t.start()
                time.sleep(0.01)
            for t in threads:  
                t.join()

            # break
            time.sleep(1)
        except Exception as e:
            print(e)

def test_socks5():
     # 一度に5つ取得 (text形式)
    get_proxy_url = 'http://api2.rola.vip/user_get_ip_list?token=TOKEN&qty=5&country=&state=&city=&time=5&format=txt&protocol=http&filter=1'

    while True:
        try:
            resp = requests.get(url=get_proxy_url, timeout=5)
            if resp.status_code != 200:
                print("IP取得失敗")
                time.sleep(1)
                continue
            ips = resp.text.split("\r\n")
            threads = []
            for ip in ips:
                threads.append(ThreadFactory(ip,"socks5"))
            for t in threads:  
                t.start()
                time.sleep(0.01)
            for t in threads:  
                t.join()

            # break
            time.sleep(1)
        except Exception as e:
            print(e)

if __name__ == "__main__":
    # httpテスト
    #test_http()

    # socks5テスト
    test_socks5()

アカウント・パスワード接続

import json
import threading
import time
import requests

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:85.0) Gecko/20100101 Firefox/85.0",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
    "Accept-Encoding": "gzip, deflate, br"
}

# ビジネスロジック (http)
def execute_http(user_name,password):
    targetUrl = 'http://ip123.in/ip.json'

    proxies = {
        'http': 'http://{}:{}@gate.rola.vip:1000'.format(user_name,password),
        'https':'http://{}:{}@gate.rola.vip:1000'.format(user_name,password),
    }
    res = ""

    while True:
        try:
            res = requests.get(targetUrl, proxies=proxies, timeout=5)
            # print(res.status_code)
            print(res.status_code, "***", res.text)
            break
        except Exception as e:
            print(e)
            break
    return

# ビジネスロジック (socks5)
def execute_socks5(user_name,password):
    targetUrl = 'http://ip123.in/ip.json'

    proxies = {
        'http': 'socks5h://{}:{}@gate.rola.vip:2000'.format(user_name,password),
        'https':'socks5h://{}:{}@gate.rola.vip:2000'.format(user_name,password),
    }

    res = ""

    while True:
        try:
            res = requests.get(targetUrl, proxies=proxies, timeout=5)
            # print(res.status_code)
            print(res.status_code, "***", res.text)
            break
        except Exception as e:
            print(e)
            break
    return

class ThreadFactory(threading.Thread):
    def __init__(self, user_name,password,proxy_type):
        threading.Thread.__init__(self)
        self.user_name = user_name
        self.password = password
        self.proxy_type = proxy_type
    def run(self):
        if self.proxy_type == "socks5":
            execute_socks5(self.user_name,self.password)
        else:
            execute_http(self.user_name,self.password)

def test_http():
  # プロキシユーザー名
    user_name = "USERNAME"
    # プロキシパスワード
    passwrod = "PASSWORD"
    # 国コード
    country="us"
    while True:
        try:
            threads = [] 
            # プロキシを1つ生成、形式: USERNAME-country-COUNTRYCODE-sid-SESSION
            proxy_user = "{}-country-{}-sid-{}".format(user_name,country,time.time() * 1000000)

            threads.append(ThreadFactory(proxy_user,passwrod,"http"))    
            for t in threads:  # スレッドを開始
                t.start()
                time.sleep(0.01)
            for t in threads:  # スレッドをブロック
                t.join()
            # break
            time.sleep(1)
        except Exception as e:
            print(e)

def test_socks5():
    # プロキシユーザー名
    user_name = "USERNAME"
    # プロキシパスワード
    passwrod = "PASSWORD"
    # 国コード
    country="us"

    while True:
        try:
            threads = [] 
            # プロキシを1つ生成、形式: USERNAME-country-COUNTRYCODE-sid-SESSION   SESSION は乱数
            proxy_user = "{}-country-{}-sid-{}".format(user_name,country,int(time.time() * 1000000))
            threads.append(ThreadFactory(proxy_user,passwrod,"socks5"))    
            for t in threads: 
                t.start()
                time.sleep(0.01)
            for t in threads:  
                t.join()
            # break
            time.sleep(1)
        except Exception as e:
            print(e)

if __name__ == "__main__":
    # httpテスト
    test_http()

    # socks5テスト
    #test_socks5()