如何用Python爬取需要登录的网站？从模拟登录到数据导出

python demo_site.py --host 127.0.0.1 --port 8765

python -m pip install requests==2.33.1 beautifulsoup4==4.14.3 playwright==1.58.0
python -m playwright install chromium

$env:DEMO_BASE_URL = "http://127.0.0.1:8765"
$env:DEMO_USERNAME = "editor@example.test"
$env:DEMO_PASSWORD = "demo-pass-123"

import csv
import os
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup

BASE = os.environ["DEMO_BASE_URL"].rstrip("/") + "/"
USER = os.environ["DEMO_USERNAME"]
PASSWORD = os.environ["DEMO_PASSWORD"]
OUTPUT = Path(os.getenv("OUTPUT_CSV", "products_requests.csv"))


def collect():
    with requests.Session() as session:
        session.headers.update({"User-Agent": "AuthorizedDemoResearch/1.0"})
        login_page = session.get(urljoin(BASE, "login"), timeout=10)
        login_page.raise_for_status()
        soup = BeautifulSoup(login_page.text, "html.parser")
        token = soup.select_one('input[name="csrf_token"]')
        if token is None or not token.get("value"):
            raise RuntimeError("未找到当前登录表单的 CSRF Token")

        login = session.post(
            urljoin(BASE, "login"),
            data={
                "csrf_token": token["value"],
                "email": USER,
                "password": PASSWORD,
            },
            timeout=10,
        )
        login.raise_for_status()
        if "demo_session" not in session.cookies:
            raise RuntimeError("未保存演示站的会话 Cookie")

        protected = session.get(urljoin(BASE, "products"), timeout=10)
        protected.raise_for_status()
        if urlparse(protected.url).path.rstrip("/") != "/products":
            raise RuntimeError("未进入受保护商品页，请检查登录结果")

        soup = BeautifulSoup(protected.text, "html.parser")
        products = []
        for row in soup.select("tr.product"):
            item = {}
            for key in ("sku", "name", "price"):
                cell = row.select_one("." + key)
                if cell is None:
                    raise RuntimeError(f"商品行缺少 {key} 字段")
                item[key] = cell.get_text(strip=True)
            products.append(item)
        if not products:
            raise RuntimeError("未找到商品，请检查登录状态和选择器")
        return products, sorted(session.cookies.keys())


def main():
    products, cookie_names = collect()
    with OUTPUT.open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.DictWriter(handle, fieldnames=("sku", "name", "price"))
        writer.writeheader()
        writer.writerows(products)
    print("Authenticated page: /products")
    print("Cookie names:", ", ".join(cookie_names))
    print(f"Exported {len(products)} product rows to {OUTPUT}")
    for item in products:
        print(item["sku"], item["name"], item["price"])


if __name__ == "__main__":
    main()

python requests_login.py

import csv
import os
from pathlib import Path
from urllib.parse import urlparse

from playwright.sync_api import sync_playwright

BASE = os.environ["DEMO_BASE_URL"].rstrip("/")
USER = os.environ["DEMO_USERNAME"]
PASSWORD = os.environ["DEMO_PASSWORD"]
OUTPUT = Path(os.getenv("OUTPUT_CSV", "products_playwright.csv"))


def collect():
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=True)
        try:
            context = browser.new_context()
            page = context.new_page()
            page.set_default_timeout(15000)
            page.goto(BASE + "/login", wait_until="domcontentloaded")
            page.get_by_label("Email").fill(USER)
            page.get_by_label("Password").fill(PASSWORD)
            page.get_by_role("button", name="Sign in").click()
            page.wait_for_url(BASE + "/products")
            page.locator("tr.product").first.wait_for()

            page.goto(BASE + "/dynamic", wait_until="domcontentloaded")
            if urlparse(page.url).path.rstrip("/") != "/dynamic":
                raise RuntimeError("动态商品页未通过登录验证")
            rows = page.locator('#catalog[data-state="ready"] tr.product')
            rows.first.wait_for()
            products = rows.evaluate_all("""rows => rows.map(row => ({
                sku: row.querySelector('.sku').textContent.trim(),
                name: row.querySelector('.name').textContent.trim(),
                price: row.querySelector('.price').textContent.trim()
            }))""")
            if not products:
                raise RuntimeError("页面未渲染出商品数据")
            return products
        finally:
            browser.close()


def main():
    products = collect()
    with OUTPUT.open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.DictWriter(handle, fieldnames=("sku", "name", "price"))
        writer.writeheader()
        writer.writerows(products)
    print("Authenticated dynamic page: /dynamic")
    print(f"Exported {len(products)} JavaScript-rendered rows to {OUTPUT}")
    for item in products:
        print(item["sku"], item["name"], item["price"])


if __name__ == "__main__":
    main()

python playwright_login.py

account_login001-country-us-sessiontime-10

import os
from urllib.parse import quote


def proxy_settings():
    # 使用基础账号名，不要重复追加已有的会话或地区后缀。
    account = os.environ["ROLA_PROXY_USER"]
    password = os.environ["ROLA_PROXY_PASSWORD"]
    session_id = os.getenv("ROLA_SESSION_ID", "login001")
    country = os.getenv("ROLA_COUNTRY", "us").lower()
    minutes = int(os.getenv("ROLA_SESSION_MINUTES", "10"))
    gateway = os.getenv("ROLA_PROXY_GATEWAY", "gate.rola.vip:1000")
    if not 1 <= minutes <= 120:
        raise ValueError("会话时长必须在 1—120 分钟之间")

    username = f"{account}_{session_id}-country-{country}-sessiontime-{minutes}"
    url = (
        f"http://{quote(username, safe='')}:{quote(password, safe='')}"
        f"@{gateway}"
    )
    return (
        {"http": url, "https": url},
        {"server": "http://" + gateway,
         "username": username, "password": password},
    )

from rola_proxy import proxy_settings

# 以下两行放在 with requests.Session() as session: 内部。
requests_proxies, _ = proxy_settings()
session.proxies.update(requests_proxies)

_, browser_proxy = proxy_settings()
browser = playwright.chromium.launch(headless=True, proxy=browser_proxy)

"""Small local storefront used by the login-scraping tutorial.

This server uses only Python's standard library. It is intentionally local and
contains synthetic products and credentials; do not expose it to the internet.
"""

import argparse
import html
import http.cookies
import json
import secrets
import urllib.parse
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


DEMO_USER = "editor@example.test"
DEMO_PASSWORD = "demo-pass-123"
PRODUCTS = [
    {"sku": "DR-101", "name": "Meridian Desk Lamp", "price": "$42.00"},
    {"sku": "NT-204", "name": "Grid Notebook", "price": "$18.50"},
    {"sku": "CB-307", "name": "Canvas Carry Bag", "price": "$29.00"},
]
SESSIONS = {}


def cookie_value(headers, name):
    parsed = http.cookies.SimpleCookie()
    parsed.load(headers.get("Cookie", ""))
    morsel = parsed.get(name)
    return morsel.value if morsel else None


class DemoHandler(BaseHTTPRequestHandler):
    server_version = "LoginScrapeDemo/1.0"

    def send_html(self, body, status=HTTPStatus.OK, extra_headers=None):
        payload = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        for key, value in (extra_headers or {}).items():
            if isinstance(value, (list, tuple)):
                for item in value:
                    self.send_header(key, item)
            else:
                self.send_header(key, value)
        self.end_headers()
        self.wfile.write(payload)

    def send_json(self, value, status=HTTPStatus.OK):
        payload = json.dumps(value).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def session(self):
        sid = cookie_value(self.headers, "demo_session")
        return sid, SESSIONS.get(sid)

    def require_auth(self):
        sid, state = self.session()
        if not state or not state.get("authenticated"):
            self.send_response(HTTPStatus.FOUND)
            self.send_header("Location", "/login")
            self.end_headers()
            return None
        return sid, state

    def do_GET(self):  # noqa: N802 - required by BaseHTTPRequestHandler
        path = urllib.parse.urlparse(self.path).path
        if path == "/login":
            sid = cookie_value(self.headers, "demo_session") or secrets.token_urlsafe(18)
            token = secrets.token_urlsafe(24)
            SESSIONS[sid] = {"authenticated": False, "csrf_token": token}
            body = f"""<!doctype html>
<html><head><meta charset='utf-8'><title>Demo sign in</title></head>
<body><h1>Demo storefront sign in</h1>
<form method='post' action='/login'>
<label for='email'>Email</label><input id='email' name='email' type='email'><br>
<label for='password'>Password</label><input id='password' name='password' type='password'><br>
<input type='hidden' name='csrf_token' value='{html.escape(token)}'>
<button type='submit'>Sign in</button>
</form></body></html>"""
            self.send_html(
                body,
                extra_headers={
                    "Set-Cookie": [
                        f"demo_session={sid}; Path=/; HttpOnly",
                        "demo_browser=1; Path=/; HttpOnly",
                    ]
                },
            )
            return

        if path == "/products":
            if not self.require_auth():
                return
            rows = "".join(
                f"<tr class='product'><td class='sku'>{html.escape(p['sku'])}</td>"
                f"<td class='name'>{html.escape(p['name'])}</td>"
                f"<td class='price'>{html.escape(p['price'])}</td></tr>"
                for p in PRODUCTS
            )
            self.send_html(
                "<!doctype html><html><body><h1>Products</h1>"
                f"<table><thead><tr><th>SKU</th><th>Name</th><th>Price</th></tr></thead>"
                f"<tbody>{rows}</tbody></table></body></html>"
            )
            return

        if path == "/dynamic":
            if not self.require_auth():
                return
            body = """<!doctype html><html><body><h1>Dynamic catalog</h1>
<table id='catalog' data-state='loading'><tbody></tbody></table>
<script>
fetch('/api/products').then(r => r.json()).then(products => {
  const catalog = document.querySelector('#catalog');
  catalog.querySelector('tbody').innerHTML = products.map(p =>
    `<tr class="product"><td class="sku">${p.sku}</td><td class="name">${p.name}</td><td class="price">${p.price}</td></tr>`
  ).join('');
  catalog.dataset.state = 'ready';
});
</script></body></html>"""
            self.send_html(body)
            return

        if path == "/api/products":
            sid, state = self.session()
            if not state or not state.get("authenticated"):
                self.send_json({"error": "authentication required"}, status=HTTPStatus.UNAUTHORIZED)
                return
            self.send_json(PRODUCTS)
            return

        self.send_error(HTTPStatus.NOT_FOUND)

    def do_POST(self):  # noqa: N802 - required by BaseHTTPRequestHandler
        path = urllib.parse.urlparse(self.path).path
        if path != "/login":
            self.send_error(HTTPStatus.NOT_FOUND)
            return
        sid, state = self.session()
        if not sid or not state:
            self.send_error(HTTPStatus.FORBIDDEN, "Open /login first")
            return
        length = int(self.headers.get("Content-Length", "0"))
        values = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8"))
        token = values.get("csrf_token", [""])[0]
        email = values.get("email", [""])[0]
        password = values.get("password", [""])[0]
        if token != state["csrf_token"]:
            self.send_error(HTTPStatus.FORBIDDEN, "Invalid CSRF token")
            return
        if email != DEMO_USER or password != DEMO_PASSWORD:
            self.send_error(HTTPStatus.UNAUTHORIZED, "Invalid demo credentials")
            return
        state["authenticated"] = True
        self.send_response(HTTPStatus.FOUND)
        self.send_header("Location", "/products")
        self.end_headers()


def main():
    parser = argparse.ArgumentParser(description="Run the local login-scraping demo")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8765)
    args = parser.parse_args()
    server = ThreadingHTTPServer((args.host, args.port), DemoHandler)
    print(f"Demo site listening at http://{args.host}:{args.port}/login")
    print(f"Demo credentials: {DEMO_USER} / {DEMO_PASSWORD}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopping demo site")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()