Python实战：如何利用海外代理IP采集加拿大机票酒店价格

def get_qg_proxies():
    """
    从青果网络API获取海外代理IP列表
    """
    try:
        response = requests.get(QG_PROXY_API, timeout=10)
        if response.status_code == 200:
            data = response.json()
            proxies = []
            for item in data['data']:
                proxy_url =  "https://overseas.proxy.qg.net/get?key=yourkey&num=1&area=&isp=&format=txt&seq=\r\n&distinct=false" # 青果网络海外代理IP API地址
                proxies.append(proxy_url)
            return proxies
        else:
            print("获取代理IP失败，使用直连")
            return []
    except Exception as e:
        print(f"获取代理IP异常: {e}")
        return []

def test_proxy(proxy_url, test_url="https://httpbin.org/ip"):
    """
    测试代理IP是否可用
    """
    try:
        proxies = {
            "http": proxy_url,
            "https": proxy_url
        }
        response = requests.get(test_url, proxies=proxies, timeout=5)
        if response.status_code == 200:
            return True
    except:
        return False
    return False

import requests
from bs4 import BeautifulSoup
import time
from playwright.sync_api import sync_playwright
import pandas as pd  # Added for DataFrame in analysis module

# 青果网络API配置（替换为你的实际API密钥）
QG_API_KEY = 'your_actual_key_here'  # Replace with your real API key
QG_PROXY_BASE_URL = 'https://overseas.proxy.qg.net/get?key={key}&num=1&area={area}&isp=&format=txt&seq=\\r\\n&distinct=false'

# 获取代理IP函数：从青果API获取代理，返回代理字典
def get_proxy(country='US'):
    """
    从青果网络API获取指定国家的代理IP。
    :param country: 国家代码，如 'US', 'GB', 'JP'
    :return: 代理字典或None
    """
    try:
        url = QG_PROXY_BASE_URL.format(key=QG_API_KEY, area=country)
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            proxy_text = response.text.strip()
            if ':' in proxy_text:
                ip, port = proxy_text.split(':', 1)
                proxy_url = f"http://{ip}:{port}"
                return {
                    'http': proxy_url,
                    'https': proxy_url
                }
            else:
                raise ValueError("Invalid proxy format returned")
        else:
            raise Exception(f"Failed to get proxy for {country}: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"获取代理IP异常 ({country}): {e}")
        return None

# 测试代理IP是否可用
def test_proxy(proxy, test_url="https://httpbin.org/ip"):
    """
    测试代理IP是否可用。
    :param proxy: 代理字典
    :param test_url: 测试URL
    :return: True如果可用，否则False
    """
    if not proxy:
        return False
    try:
        response = requests.get(test_url, proxies=proxy, timeout=5)
        return response.status_code == 200
    except Exception:
        return False

# 获取Google Flights机票价格（优化：使用Playwright直接提取元素，避免BeautifulSoup如果可能）
def scrape_flight_price(proxy, origin='NYC', destination='YOW', date='2025-10-01'):
    """
    使用Playwright刮取Google Flights机票价格。
    :param proxy: 代理字典
    :param origin: 出发地
    :param destination: 目的地
    :param date: 日期
    :return: 价格字符串或'N/A'
    """
    if not proxy or not test_proxy(proxy):
        print("无效代理，使用直连（可能不准确）")
        proxy = None  # Fallback to no proxy

    with sync_playwright() as p:
        launch_args = {}
        if proxy:
            launch_args['proxy'] = {'server': proxy['http']}
        browser = p.chromium.launch(**launch_args)
        page = browser.new_page()
        url = f'https://www.google.com/flights?hl=en#flt={origin}.{destination}.{date}'
        try:
            page.goto(url, wait_until='networkidle', timeout=30000)
            page.wait_for_selector('.gws-flights-results__price', timeout=10000)  # 增加超时
            price = page.query_selector('.gws-flights-results__price').inner_text().strip()
        except Exception as e:
            print(f"刮取机票价格失败: {e}")
            price = 'N/A'
        finally:
            browser.close()
    return price

# 获取Booking.com酒店价格（类似优化）
def scrape_hotel_price(proxy, hotel='fairmont-chateau-laurier', city='Ottawa', checkin='2025-10-01', checkout='2025-10-02'):
    """
    使用Playwright刮取Booking.com酒店价格。
    :param proxy: 代理字典
    :param hotel: 酒店slug
    :param city: 城市（未使用，但保留）
    :param checkin: 入住日期
    :param checkout: 退房日期
    :return: 价格字符串或'N/A'
    """
    if not proxy or not test_proxy(proxy):
        print("无效代理，使用直连（可能不准确）")
        proxy = None

    with sync_playwright() as p:
        launch_args = {}
        if proxy:
            launch_args['proxy'] = {'server': proxy['http']}
        browser = p.chromium.launch(**launch_args)
        page = browser.new_page()
        url = f'https://www.booking.com/hotel/ca/{hotel}.en-gb.html?checkin={checkin}&checkout={checkout}'
        try:
            page.goto(url, wait_until='networkidle', timeout=30000)
            page.wait_for_selector('[data-testid="price-and-discounted-price"]', timeout=10000)
            price = page.query_selector('[data-testid="price-and-discounted-price"]').inner_text().strip()
        except Exception as e:
            print(f"刮取酒店价格失败: {e}")
            price = 'N/A'
        finally:
            browser.close()
    return price

# 主函数：监控不同市场
def main_scrape():
    markets = {'US': 'US', 'EU': 'GB', 'Asia': 'JP'}  # 国家代码
    results = {}
    for market, country in markets.items():
        try:
            proxy = get_proxy(country)
            if proxy:
                flight_price = scrape_flight_price(proxy)
                hotel_price = scrape_hotel_price(proxy)
                results[market] = {'Flight': flight_price, 'Hotel': hotel_price}
                print(f"{market} 市场数据获取成功: 机票 {flight_price}, 酒店 {hotel_price}")
            else:
                results[market] = {'Flight': 'N/A', 'Hotel': 'N/A'}
            time.sleep(10)  # 避免高频请求
        except Exception as e:
            print(f"{market} 市场获取失败: {e}")
            results[market] = {'Flight': 'N/A', 'Hotel': 'N/A'}
    return results

# 数据存储与分析模块（假设CanadaHotelMonitor是自定义类，这里简化；原代码中未定义，所以添加占位符）
class CanadaHotelMonitor:
    def monitor_all_hotels(self):
        # 占位符：假设返回当前价格列表（基于main_scrape结果扩展）
        # 在实际中，实现为采集多个酒店
        return [
            {'hotel': 'Fairmont Château Laurier', 'price': 'CAD 250', 'currency': 'CAD'},
            # 添加更多...
        ]

class PriceAnalyzer:
    def __init__(self):
        self.history_data = []
    
    def find_optimal_offers(self, current_data):
        """
        找出最优报价。
        :param current_data: 当前数据列表
        :return: 前3个最优报价
        """
        if not current_data:
            return None
        
        optimized_offers = []
        for offer in current_data:
            try:
                # 提取数值价格（改进：处理更多格式，如'CAD 250' -> 250.0）
                price_str = offer['price'].split()[-1]  # 假设最后是数字
                price_value = float(''.join(filter(lambda c: c.isdigit() or c == '.', price_str)))
                offer['price_value'] = price_value
                optimized_offers.append(offer)
            except ValueError:
                continue
                
        # 按价格排序
        optimized_offers.sort(key=lambda x: x['price_value'])
        return optimized_offers[:3]

def main_monitoring_loop():
    """
    主监控循环。
    """
    monitor = CanadaHotelMonitor()
    analyzer = PriceAnalyzer()
    
    while True:
        print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} 开始新一轮价格监控...")
        
        # 采集数据（集成main_scrape或扩展）
        current_prices = monitor.monitor_all_hotels()  # 替换为实际采集
        
        # 分析最优报价
        best_offers = analyzer.find_optimal_offers(current_prices)
        
        # 保存结果
        if best_offers:
            df = pd.DataFrame(best_offers)
            df.to_csv(f"ottawa_hotels_best_offers_{time.strftime('%Y%m%d')}.csv", index=False)
            print("最优报价已保存:")
            for offer in best_offers:
                print(f"{offer['hotel']}: {offer['price']} {offer['currency']}")
        
        # 间隔1小时
        print("等待下一轮监控...")
        time.sleep(3600)

if __name__ == "__main__":
    # 示例：运行一次刮取
    results = main_scrape()
    print("最终结果:", results)
    # 或运行循环：main_monitoring_loop()