Bright Data: プロキシサービスで堅牢な Web スクレイピング

#%pip install requests python-dotenv -q

BRIGHTDATA_HOST='brd.superproxy.io'
BRIGHTDATA_PORT='your_port'
BRIGHTDATA_USERNAME='brd-customer-hl_xxxxxxxx-zone-your_zone_name'
BRIGHTDATA_PASSWORD='your_zone_password'

import os
import requests
from dotenv import load_dotenv

# Load environment variables from the .env file
load_dotenv()

# Retrieve credentials
host = os.getenv("BRIGHTDATA_HOST")
port = os.getenv("BRIGHTDATA_PORT")
username = os.getenv("BRIGHTDATA_USERNAME")
password = os.getenv("BRIGHTDATA_PASSWORD")

# Check if all credentials are loaded
if not all([host, port, username, password]):
    raise ValueError("Proxy credentials not found in .env file. Please check your configuration.")
    
# Construct the proxy URL for the requests library
proxy_url = f"http://{username}:{password}@{host}:{port}"
proxies = {
    "http": proxy_url,
    "https": proxy_url
}

print("✅ Proxy credentials loaded successfully!")

target_url = 'https://geo.brdtest.com/mygeo.json'

response_local = requests.get(target_url)
response_local.raise_for_status() # Raise an exception for bad status codes
local_ip = response_local.json().get('geo')
local_ip

from IPython.display import display, Markdown

target_url = 'https://geo.brdtest.com/mygeo.json'

# 1. Request with your local IP
try:
    print("Requesting without a proxy...")
    response_local = requests.get(target_url)
    response_local.raise_for_status() # Raise an exception for bad status codes
    local_data = response_local.json().get('geo')
    display(f"🌍 Your Local Geo: {local_data}")
except requests.exceptions.RequestException as e:
    print(f"An error occurred: {e}")

print("-" * 30)

# 2. Request with the Bright Data proxy
try:
    print("Requesting with a Bright Data proxy...")
    # The 'proxies' dict tells requests to route the call through the proxy.
    # `verify=False` is used here like the -k flag in cURL to handle SSL certificates via the proxy.
    response_proxy = requests.get(target_url, proxies=proxies, verify=False)
    response_proxy.raise_for_status()
    proxy_data = response_proxy.json().get('geo')
    display(f"🕵️ Your Proxy Geo: {proxy_data}")
except requests.exceptions.RequestException as e:
    print(f"An error occurred: {e}")

def get_proxied_ip(country_code=None):
    """Fetches the origin IP through a potentially geo-targeted proxy."""
    
    proxy_user_geo = username
    if country_code:
        proxy_user_geo += f"-country-{country_code.lower()}"
    
    geo_proxy_url = f"http://{proxy_user_geo}:{password}@{host}:{port}"
    geo_proxies = {
        'http': geo_proxy_url,
        'https': geo_proxy_url
    }
    
    try:
        print(f"Requesting with a proxy from country: {country_code or 'Any (Rotating)'}...")
        response = requests.get(target_url, proxies=geo_proxies, timeout=10, verify=False)
        response.raise_for_status()
        data = response.json()
        display(f"  -> 🕵️ Proxy Data: {data.get('geo')}")
    except requests.exceptions.RequestException as e:
        print(f"  -> An error occurred: {e}\n")

# Example: Request from Germany (DE) and Canada (CA)
get_proxied_ip(country_code="de")
get_proxied_ip(country_code="ca")

import random
import requests

# Simplified sticky session test using a single IP echo endpoint.
IP_ENDPOINT = "https://api.ipify.org?format=json"

def test_session_ip(session_id: int, attempts: int = 2, timeout: int = 10):
    """Check whether the same proxy IP is kept across multiple requests using a session ID.
    Args:
        session_id: Arbitrary integer/str to pin the session.
        attempts: How many requests to make (default 2 for a simple comparison).
        timeout: Seconds before timing out each request.
    """
    session_username = f"{username}-session-{session_id}"
    session_proxy_url = f"http://{session_username}:{password}@{host}:{port}"
    session_proxies = {
        'http': session_proxy_url,
        'https': session_proxy_url
    }

    display(f"--- Testing Sticky Session (Session ID: {session_id}) ---")
    ips = []
    for i in range(attempts):
        try:
            display(f"  Request #{i+1} -> querying {IP_ENDPOINT}")
            resp = requests.get(IP_ENDPOINT, proxies=session_proxies, timeout=timeout, verify=False)
            resp.raise_for_status()
            ip = resp.json().get('ip')
            display(f"    Returned IP: {ip}")
            ips.append(ip)
        except Exception as e:
            display(f"    Error: {e}")
            ips.append(None)

    # Simple evaluation
    if len(ips) >= 2 and all(ips) and len(set(ips)) == 1:
        display("✅ Sticky success: All requests used the same IP.")
    else:
        display("❌ Not sticky (or undetermined). IPs observed:")
        for idx, ip in enumerate(ips, start=1):
            display(f"    Attempt {idx}: {ip}")
        display("    (Different or missing IPs can mean rotation is enforced or the request failed.)")

# Run the simplified test
random_session_id = random.randint(100000, 999999)
test_session_ip(random_session_id)