旅游比价代理IP实操：用Socks5.IO配置Python机票与酒店价格采集

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "requests[socks]>=2.32,<3"

n
import argparse
import csv
import ipaddress
import json
import os
import re
import sys
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.parse import quote, urlsplit, urlunsplit
from uuid import uuid4

import requests

EXPECTED = {"search_id": "example-trip-01", "currency": "USD",
            "locale": "en-US", "requested_market": "US"}
FLIGHT = {"itinerary": "DEMO100 JFK-LHR 2026-11-10T19:00-05:00/2026-11-11T07:00+00:00",
          "passengers": "2 adults; 0 children", "cabin": "economy",
          "fare_brand": "Demo Standard", "baggage": "1 checked bag per adult",
          "refund_terms": "non-refundable", "payment_terms": "public card total"}
HOTEL = {"property_key": "DEMO-MAPPED-HOTEL-01", "room": "Standard Double",
         "checkin": "2026-11-11", "checkout": "2026-11-13",
         "occupancy": "2 adults; 0 children; 1 room", "meals": "room only",
         "cancellation": "non-refundable", "payment_terms": "pay now"}
DEMO = {"context": EXPECTED.copy(), "offers": [
    {"source_offer_id": "flight-a", "kind": "flight", "amount": "620.00",
     "basis": "party_total", "tax_status": "included", "eligibility": "public", "facts": FLIGHT},
    {"source_offer_id": "flight-b", "kind": "flight", "amount": "570.00",
     "basis": "party_total", "tax_status": "included", "eligibility": "public",
     "facts": {**FLIGHT, "baggage": "no checked bag"}},
    {"source_offer_id": "hotel-a", "kind": "hotel", "amount": "420.00",
     "basis": "stay_total", "tax_status": "unknown", "eligibility": "public", "facts": HOTEL}]}
FIELDS = {"flight": tuple(FLIGHT), "hotel": tuple(HOTEL)}


def normalize(payload, run_id, retrieved_at, source, source_kind, expected_context=None):
    expected = EXPECTED if expected_context is None else expected_context
    if payload.get("context") != expected:
        raise ValueError("Returned search context differs from EXPECTED")
    offers = payload.get("offers")
    if not isinstance(offers, list) or not offers:
        raise ValueError("No offer list; this does not prove unavailability")
    rows, seen = [], set()
    for offer in offers:
        kind = offer.get("kind")
        if kind not in FIELDS:
            raise ValueError("Implement a separate adapter for this product type")
        identity = offer.get("source_offer_id")
        if not isinstance(identity, str) or not identity.strip() or identity in seen:
            raise ValueError("Missing or duplicate source offer identifier")
        seen.add(identity)
        amount = str(offer.get("amount", ""))
        if not re.fullmatch(r"\d+(?:\.\d+)?", amount) or Decimal(amount) <= 0:
            raise ValueError("Require an explicit positive numeric amount")
        raw_facts = offer.get("facts", {})
        if not isinstance(raw_facts, dict):
            raise ValueError("Expected a facts object")
        facts, flags = {}, []
        for field in FIELDS[kind]:
            value = raw_facts.get(field)
            if not isinstance(value, str) or not value.strip() or value.strip().lower() == "unknown":
                value = "unknown"
                flags.append("unknown_" + field)
            facts[field] = value.strip()
        basis = offer.get("basis", "unknown")
        tax = offer.get("tax_status", "unknown")
        eligibility = offer.get("eligibility", "unknown")
        if basis != {"flight": "party_total", "hotel": "stay_total"}[kind]:
            flags.append("price_basis_requires_review")
        if tax != "included":
            flags.append("taxes_not_confirmed_included")
        if eligibility != "public":
            flags.append("eligibility_requires_review")
        comparison_eligible = not flags
        comparison_key = (json.dumps([kind, facts, expected["currency"], basis,
                                      tax, eligibility], sort_keys=True)
                          if comparison_eligible else "")
        rows.append({"run_id": run_id, "retrieved_at_utc": retrieved_at,
                     "source": source, "source_kind": source_kind, **expected,
                     "source_offer_id": identity, "kind": kind,
                     "amount": str(Decimal(amount)), "basis": basis, "tax_status": tax,
                     "eligibility": eligibility, "facts_json": json.dumps(facts, sort_keys=True),
                     "comparison_eligible": comparison_eligible,
                     "comparison_key": comparison_key, "review_reason": ";".join(flags)})
    return rows


def public_url(url):
    p = urlsplit(url)
    if p.scheme != "https" or not p.hostname or p.username or p.password:
        raise ValueError("Use an approved public HTTPS URL without credentials")
    host = p.hostname.lower().rstrip(".")
    if host == "localhost" or host.endswith(".localhost"):
        raise ValueError("Do not send local fixtures through a remote proxy")
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        address = None
    if address is not None and not address.is_global:
        raise ValueError("Private IP literals are not supported")
    return p


def make_session(use_proxy):
    session = requests.Session()
    session.trust_env = False
    session.headers["User-Agent"] = "AuthorizedTravelResearch/1.0"
    if use_proxy:
        host = os.environ["SOCKS5IO_HOST"]
        port = int(os.environ["SOCKS5IO_PORT"])
        if not re.fullmatch(r"[A-Za-z0-9.-]+", host) or not 1 <= port <= 65535:
            session.close()
            raise ValueError("Check the issued gateway and port")
        user = quote(os.environ["SOCKS5IO_USER"], safe="")
        password = quote(os.environ["SOCKS5IO_PASSWORD"], safe="")
        proxy = f"socks5h://{user}:{password}@{host}:{port}"
        session.proxies.update(http=proxy, https=proxy)
    return session


def get_json(session, url):
    public_url(url)
    response = session.get(url, timeout=(5, 20), allow_redirects=False)
    if response.status_code != 200:
        raise ValueError(f"HTTP {response.status_code}: stop and review access/quota")
    if "application/json" not in response.headers.get("Content-Type", "").lower():
        raise ValueError("Expected JSON; inspect source content, not just HTTP status")
    return response.json()


def check_exit(session, url):
    address = ipaddress.ip_address(get_json(session, url)["ip"])
    if not address.is_global:
        raise ValueError("Exit checker returned a non-public address")
    return {"observed_ip": str(address),
            "checked_at_utc": datetime.now(timezone.utc).isoformat()}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--url")
    parser.add_argument("--approved", action="store_true")
    parser.add_argument("--proxy", action="store_true")
    parser.add_argument("--check-url", default="https://api.ipify.org?format=json")
    parser.add_argument("--route-label", default="private-run")
    parser.add_argument("--market", choices=["US", "GB"], default="US")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    expected = {**EXPECTED, "requested_market": args.market}
    log_path = args.out.with_suffix(".route.json")
    if args.out.exists() or (args.url and log_path.exists()):
        parser.error("Use new output filenames; existing files are not replaced")
    run_id = uuid4().hex
    if args.url:
        if not args.approved:
            parser.error("Review source permission and request budget before --approved")
        with make_session(args.proxy) as session:
            before = check_exit(session, args.check_url)
            payload = get_json(session, args.url)
            stamp = datetime.now(timezone.utc).isoformat()
            after = check_exit(session, args.check_url)
        manifest = {"run_id": run_id, "route_label": args.route_label,
                    "proxy_requested": args.proxy, "requested_market": args.market,
                    "before": before, "after": after,
                    "same_observed_ip": before["observed_ip"] == after["observed_ip"]}
        with log_path.open("x", encoding="utf-8") as handle:
            json.dump(manifest, handle, indent=2)
        p = public_url(args.url)
        source = urlunsplit((p.scheme, p.netloc, p.path, "", ""))
        source_kind = "user_approved_normalized_feed"
    else:
        if args.proxy:
            parser.error("The embedded fixture is not a proxy test")
        payload = {**DEMO, "context": {**DEMO["context"], "requested_market": args.market}}
        source, source_kind = "local:embedded-demo", "synthetic_fixture"
        stamp = datetime.now(timezone.utc).isoformat()
    rows = normalize(payload, run_id, stamp, source, source_kind, expected)
    with args.out.open("x", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print(f"Saved {len(rows)} observations. Review product keys and review_reason.")


if __name__ == "__main__":
    try:
        main()
    except ValueError as error:
        print("Stopped:", error, file=sys.stderr)
        sys.exit(1)
    except (KeyError, TypeError, AttributeError, OSError, requests.RequestException):
        print("Stopped: inspect fields, access, credentials and output filenames privately.", file=sys.stderr)
        sys.exit(1)

python travel_fares.py --out travel-demo.csv

row["comparison_eligible"] == "True"

$env:SOCKS5IO_HOST = Read-Host 'Issued proxy host'
$env:SOCKS5IO_PORT = Read-Host 'Issued proxy port'
$env:SOCKS5IO_USER = Read-Host 'Issued proxy username'
$travelProxySecret = Read-Host 'Issued proxy password' -AsSecureString
$env:SOCKS5IO_PASSWORD = [System.Net.NetworkCredential]::new('', $travelProxySecret).Password

Remove-Item Env:SOCKS5IO_PASSWORD
Remove-Variable travelProxySecret

python travel_fares.py --url "https://feed.example.invalid/approved-fares" --approved --market US --route-label direct-US --out travel-direct.csv
python travel_fares.py --url "https://feed.example.invalid/approved-fares" --approved --proxy --market US --route-label residential-US --out travel-proxy.csv

python travel_fares.py --url "https://feed.example.invalid/approved-gb-fares" --approved --proxy --market GB --route-label residential-GB --out travel-gb.csv

python travel_fares.py --market GB --out travel-gb-demo.csv