Oxylabs: AI 対応データ収集のための Web Scraper API

# !pip install requests python-dotenv beautifulsoup4

import os
import json
import requests
from dotenv import load_dotenv

load_dotenv()

OXYLABS_USERNAME = os.getenv("OXYLABS_USERNAME")
OXYLABS_PASSWORD = os.getenv("OXYLABS_PASSWORD")

assert OXYLABS_USERNAME and OXYLABS_PASSWORD, "Please set OXYLABS_USERNAME and OXYLABS_PASSWORD in your .env file"
print("✓ Oxylabs credentials loaded successfully")

# Basic request to scrape any website
payload = {
    "source": "universal",
    "url": "https://sandbox.oxylabs.io/"
}

response = requests.post(
    "https://realtime.oxylabs.io/v1/queries",
    auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
    json=payload
)

print(f"Status Code: {response.status_code}")
data = response.json()
print(f"Response keys: {list(data.keys())}")
print(f"\nContent length: {len(data['results'][0]['content'])} characters")
print(f"\nFirst 500 characters of content:\n{data['results'][0]['content'][:500]}")

# Scrape Google search results with structured parsing
payload = {
    "source": "google_search",
    "query": "artificial intelligence tutorials 2025",
    "geo_location": "United States",
    "parse": True
}

response = requests.post(
    "https://realtime.oxylabs.io/v1/queries",
    auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
    json=payload
)

data = response.json()
results = data["results"][0]["content"]

print("=== Google Search Results (Parsed) ===\n")
print(f"Total organic results: {len(results.get('results', {}).get('organic', []))}\n")

# Display top 5 organic results
for i, result in enumerate(results.get("results", {}).get("organic", [])[:5], 1):
    print(f"{i}. {result.get('title', 'N/A')}")
    print(f"   URL: {result.get('url', 'N/A')}")
    print(f"   Snippet: {result.get('desc', 'N/A')[:100]}...")
    print()

# Compare search results from different locations
locations = ["United States", "United Kingdom", "Germany"]

for location in locations:
    payload = {
        "source": "google_search",
        "query": "best AI tools",
        "geo_location": location,
        "parse": True
    }
    
    response = requests.post(
        "https://realtime.oxylabs.io/v1/queries",
        auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
        json=payload
    )
    
    data = response.json()
    results = data["results"][0]["content"]
    organic = results.get("results", {}).get("organic", [])
    
    print(f"\n{'='*50}")
    print(f"Location: {location}")
    print(f"{'='*50}")
    print(f"Top 3 results:")
    for i, result in enumerate(organic[:3], 1):
        print(f"  {i}. {result.get('title', 'N/A')}")

# Scrape Amazon search results with structured parsing
payload = {
    "source": "amazon_search",
    "query": "mechanical keyboard",
    "geo_location": "90210",
    "parse": True
}

response = requests.post(
    "https://realtime.oxylabs.io/v1/queries",
    auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
    json=payload
)

data = response.json()
results = data["results"][0]["content"]

print("=== Amazon Search Results ===\n")
# Display product results
products = results.get("results", {}).get("organic", [])
print(f"Found {len(products)} products\n")

for i, product in enumerate(products[:5], 1):
    print(f"{i}. {product.get('title', 'N/A')[:80]}")
    print(f"   Price: {product.get('price_upper', product.get('price', 'N/A'))}")
    print(f"   Rating: {product.get('rating', 'N/A')}")
    print(f"   URL: {product.get('url', 'N/A')[:80]}")
    print()

from bs4 import BeautifulSoup

# Scrape a webpage and extract clean text for LLM processing
payload = {
    "source": "universal",
    "url": "https://sandbox.oxylabs.io/"
}

response = requests.post(
    "https://realtime.oxylabs.io/v1/queries",
    auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
    json=payload
)

data = response.json()
html_content = data["results"][0]["content"]

# Parse HTML and extract clean text
soup = BeautifulSoup(html_content, "html.parser")

# Remove script and style elements
for script in soup(["script", "style"]):
    script.decompose()

# Get text content
text = soup.get_text(separator="\n", strip=True)

print("=== Extracted Text (first 1000 chars) ===\n")
print(text[:1000])
print(f"\n\n--- Total extracted text length: {len(text)} characters ---")
print("\nThis clean text is now ready to be used as context for LLM queries,")
print("RAG pipelines, or training data preprocessing.")

def oxylabs_scrape(url=None, source="universal", query=None, parse=False, geo_location=None):
    """
    Reusable helper function for Oxylabs Web Scraper API.
    
    Args:
        url: Target URL to scrape (for universal and URL-based sources)
        source: Scraper source type (e.g., 'universal', 'google_search', 'amazon_search')
        query: Search query (for search-based sources)
        parse: Whether to return structured/parsed data
        geo_location: Geographic location for geo-targeted results
    
    Returns:
        dict: API response data
    """
    payload = {"source": source}
    
    if url:
        payload["url"] = url
    if query:
        payload["query"] = query
    if parse:
        payload["parse"] = True
    if geo_location:
        payload["geo_location"] = geo_location
    
    response = requests.post(
        "https://realtime.oxylabs.io/v1/queries",
        auth=(OXYLABS_USERNAME, OXYLABS_PASSWORD),
        json=payload
    )
    
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error {response.status_code}: {response.text}")
        return None

# Example usage
result = oxylabs_scrape(url="https://sandbox.oxylabs.io/")
if result:
    print(f"✓ Successfully scraped! Content length: {len(result['results'][0]['content'])} chars")