酒店数据采集教程：用Python校验房价，配置Socks5.IO代理并导出CSV

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

python3 -m venv .venv
.venv/bin/python -m pip install "requests[socks]>=2.32,<3" "beautifulsoup4>=4.13,<5"

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

import requests
from bs4 import BeautifulSoup

EXPECTED = dict(checkin="2026-11-10", checkout="2026-11-12",
                adults="2", children="0", rooms="1", currency="USD", locale="en-US")
DEMO = '''<section id="results" data-checkin="2026-11-10" data-checkout="2026-11-12"
data-adults="2" data-children="0" data-rooms="1" data-currency="USD" data-locale="en-US">
<article class="hotel-offer" data-property="DEMO-101" data-room="Standard Double"
data-rate="Flexible" data-meal="Room only" data-cancel="Free until stated deadline"
data-basis="stay_total" data-tax="included" data-eligibility="public" data-availability="available">
<h2>Fictional Harbor Hotel</h2><span class="price" data-amount="420.00">USD 420 total; taxes included</span>
</article>
<article class="hotel-offer" data-property="DEMO-101" data-room="Standard Double"
data-rate="Non-refundable" data-meal="Room only" data-cancel="Non-refundable"
data-basis="stay_total" data-tax="excluded" data-eligibility="public" data-availability="available">
<h2>Fictional Harbor Hotel</h2><span class="price" data-amount="360.00">USD 360 total; taxes excluded</span>
</article></section>'''


def one(node, selector):
    found = node.select(selector)
    if len(found) != 1:
        raise ValueError("Expected exactly one matching node")
    return found[0]


def required(node, name):
    value = node.get("data-" + name, "").strip()
    if not value:
        raise ValueError("Missing required offer or context attribute")
    return value


def extract(html, source, kind, requested_market, retrieved_at_utc=None, run_id=None):
    retrieved_at_utc = retrieved_at_utc or datetime.now(timezone.utc).isoformat()
    run_id = run_id or uuid4().hex
    root = one(BeautifulSoup(html, "html.parser"), "#results")
    context = {key: required(root, key) for key in EXPECTED}
    if context != EXPECTED:
        raise ValueError("Returned stay context does not match the requested context")
    nights = (date.fromisoformat(context["checkout"]) - date.fromisoformat(context["checkin"])).days
    if nights <= 0 or context["children"] != "0":
        raise ValueError("Check dates; add an explicit child-age adapter before including children")
    cards = root.select("article.hotel-offer")
    if not cards:
        raise ValueError("No offer cards; do not interpret this as sold out")
    rows, seen = [], set()
    for card in cards:
        price = one(card, ".price")
        raw = price.get("data-amount", "")
        if not re.fullmatch(r"\d+(?:\.\d+)?", raw) or Decimal(raw) <= 0:
            raise ValueError("Expected an explicit positive numeric amount")
        row = {"source_property_id": required(card, "property"),
               "property_name": one(card, "h2").get_text(" ", strip=True),
               **context, "nights": nights, "displayed_price": price.get_text(" ", strip=True),
               "amount": str(Decimal(raw)), "room": required(card, "room"),
               "rate": required(card, "rate"), "meal": required(card, "meal"),
               "cancellation_text": required(card, "cancel"), "price_basis": required(card, "basis"),
               "tax_status": required(card, "tax"), "eligibility": required(card, "eligibility"),
               "availability": required(card, "availability")}
        if not row["property_name"] or not row["displayed_price"]:
            raise ValueError("Missing displayed text")
        key = (row["source_property_id"], row["room"], row["rate"], row["meal"],
               row["cancellation_text"], row["eligibility"], row["price_basis"], row["tax_status"])
        if key in seen:
            raise ValueError("Duplicate offer key; inspect pagination or source identifiers")
        seen.add(key)
        reasons = [label for label, ok in (
            ("basis_not_stay_total", row["price_basis"] == "stay_total"),
            ("taxes_not_confirmed_included", row["tax_status"] == "included"),
            ("restricted_eligibility", row["eligibility"] == "public"),
            ("availability_not_confirmed", row["availability"] == "available"),
            ("cancellation_unknown", row["cancellation_text"] != "unknown"),
            ("meal_unknown", row["meal"] != "unknown")) if not ok]
        row.update(review_reason=";".join(reasons), requested_market=requested_market,
                   retrieved_at_utc=retrieved_at_utc, run_id=run_id,
                   source_url=source, source_kind=kind)
        rows.append(row)
    return rows



# A second page layout owned by this tutorial, not an OTA response.
OWNED_DEMO = '''<main class="rate-search">
<dl class="stay-summary"><dd class="arrival">2026-11-10</dd>
<dd class="departure">2026-11-12</dd><dd class="adults">2</dd>
<dd class="children">0</dd><dd class="rooms">1</dd>
<dd class="currency">USD</dd><dd class="locale">en-US</dd></dl>
<section class="room-card" data-listing-id="OWNED-201">
<h3>Fictional Garden Hotel</h3><p class="room-name">Garden Double</p>
<p class="rate-name">Public flexible</p><p class="meal">Room only</p>
<p class="price-value">440.00</p><p class="price-label">USD 440 for the stay</p>
<p class="basis">stay_total</p><p class="eligibility">public</p>
<p class="availability">available</p></section></main>'''


def observed(node, selector, fallback=None):
    matches = node.select(selector)
    if len(matches) == 0 and fallback is not None:
        return fallback
    return one(node, selector).get_text(" ", strip=True)


def adapt_owned(html):
    source = BeautifulSoup(html, "html.parser")
    summary = one(source, ".stay-summary")
    target = BeautifulSoup('<section id="results"></section>', "html.parser")
    root = target.section
    mapping = {"checkin": "arrival", "checkout": "departure",
               "adults": "adults", "children": "children", "rooms": "rooms",
               "currency": "currency", "locale": "locale"}
    for field, css in mapping.items():
        root["data-" + field] = observed(summary, "." + css)
    cards = source.select("section.room-card")
    if not cards:
        raise ValueError("No owned-layout cards; inspect source")
    for card in cards:
        identity = card.get("data-listing-id", "").strip()
        if not identity:
            raise ValueError("Missing source property identifier")
        out = target.new_tag("article", attrs={"class": "hotel-offer"})
        fields = {"property": identity,
                  "room": observed(card, ".room-name"),
                  "rate": observed(card, ".rate-name"),
                  "meal": observed(card, ".meal", "unknown"),
                  "cancel": observed(card, ".cancellation", "unknown"),
                  "basis": observed(card, ".basis", "unknown"),
                  "tax": {"Taxes included": "included", "Taxes excluded": "excluded"}.get(
                      observed(card, ".tax-note", "unknown"), "unknown"),
                  "eligibility": observed(card, ".eligibility", "unknown"),
                  "availability": observed(card, ".availability", "unknown")}
        for field, value in fields.items():
            out["data-" + field] = value
        name = target.new_tag("h2")
        name.string = observed(card, "h3")
        price = target.new_tag("span", attrs={"class": "price",
                              "data-amount": observed(card, ".price-value")})
        price.string = observed(card, ".price-label")
        out.extend([name, price])
        root.append(out)
    return str(target)


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


def validate_url(url):
    parsed = urlsplit(url)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
        raise ValueError("Use an approved public HTTPS URL without embedded credentials")
    host = parsed.hostname.lower().rstrip(".")
    if host == "localhost" or host.endswith(".localhost"):
        raise ValueError("Local source is not supported in network mode")
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        address = None
    if address is not None and not address.is_global:
        raise ValueError("Use a public remote source")
    return parsed


def check_exit(session, url):
    validate_url(url)
    response = session.get(url, timeout=(5, 20), allow_redirects=False)
    if response.status_code != 200:
        raise ValueError("Exit-check endpoint did not return HTTP 200")
    # ipify's documented JSON contract; adapt this if using your own endpoint.
    address = ipaddress.ip_address(response.json()["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 fetch(url, use_proxy, session=None):
    validate_url(url)
    manager = nullcontext(session) if session is not None else make_session(use_proxy)
    with manager as active:
        response = active.get(url, timeout=(5, 20), allow_redirects=False)
        if response.status_code != 200:
            raise ValueError(f"HTTP {response.status_code}: stop and review source access or quota")
        if "text/html" not in response.headers.get("Content-Type", "").lower():
            raise ValueError("Expected HTML; use a separate adapter for a documented JSON interface")
        return response.content


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--url")
    parser.add_argument("--adapter-demo", action="store_true")
    parser.add_argument("--layout", choices=["canonical", "owned"], default="canonical")
    parser.add_argument("--check-url", default="https://api.ipify.org?format=json")
    parser.add_argument("--route-label", default="private-run")
    parser.add_argument("--approved", action="store_true")
    parser.add_argument("--proxy", action="store_true")
    parser.add_argument("--market", default="US")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    run_id = uuid4().hex
    route_path = args.out.with_suffix(".route.json")
    if args.out.exists() or (args.url and route_path.exists()):
        parser.error("Choose unused CSV and route-log filenames")
    if args.url and args.adapter_demo:
        parser.error("Do not mix a network source with --adapter-demo")
    manifest = None
    if args.url:
        if not args.approved:
            parser.error("Review permission, context and rate limits before --approved")
        # Same Session and proxy settings for both exit checks and the source.
        with make_session(args.proxy) as session:
            before = check_exit(session, args.check_url)
            html = fetch(args.url, args.proxy, session=session)
            retrieved_at_utc = 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, "before": before, "after": after,
                    "same_observed_ip": before["observed_ip"] == after["observed_ip"]}
        # Save privately even if the later page validation fails.
        with route_path.open("x", encoding="utf-8") as log:
            json.dump(manifest, log, indent=2)
        if args.layout == "owned":
            html = adapt_owned(html)
        p = urlsplit(args.url)
        source = urlunsplit((p.scheme, p.netloc, p.path, "", ""))
        kind = "user_approved_html"
    else:
        if args.proxy:
            parser.error("The synthetic demo is not a proxy test")
        html = adapt_owned(OWNED_DEMO) if args.adapter_demo else DEMO
        source = "local:owned-layout-demo" if args.adapter_demo else "local:embedded-demo"
        kind = "synthetic_fixture"
        retrieved_at_utc = datetime.now(timezone.utc).isoformat()
    rows = extract(html, source, kind, args.market, retrieved_at_utc, run_id)
    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; inspect review_reason and offer conditions.")


if __name__ == "__main__":
    try:
        main()
    except ValueError as error:
        print("Stopped:", error, file=sys.stderr)
        sys.exit(1)
    except (KeyError, OSError, requests.RequestException):
        print("Stopped: check endpoint fields, source access or a new output filename.", file=sys.stderr)
        sys.exit(1)

.\.venv\Scripts\python.exe hotel_scraper.py --out hotel-demo.csv

.\.venv\Scripts\python.exe hotel_scraper.py --adapter-demo --out hotel-adapter.csv

tax_status=unknown
cancellation_text=unknown
review_reason=taxes_not_confirmed_included;cancellation_unknown

.\.venv\Scripts\python.exe hotel_scraper.py --url "https://hotel.example.invalid/approved-search" --approved --market US --out hotel-direct.csv

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

.\.venv\Scripts\python.exe hotel_scraper.py --url "https://hotel.example.invalid/approved-search" --approved --proxy --market US --out hotel-proxy.csv

https://api.ipify.org?format=json

{"ip":"<observed-public-IP>"}

Remove-Item Env:SOCKS5IO_PASSWORD
Remove-Variable hotelProxySecret