PythonでXMLとHTMLをスクレイピングする方法：解析・検証・CSV出力

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

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

html_tree = BeautifulSoup(HTML, "html.parser")
html_card = html_tree.select_one("article.product")
html_name = html_card.select_one("h2 a").get_text(strip=True)
html_price = html_card.select_one(".price").get_text(strip=True)
xml_parser = etree.XMLParser(resolve_entities=False, load_dtd=False,
                             no_network=True, recover=False, huge_tree=False)
xml_tree = etree.fromstring(XML, parser=xml_parser)
xml_product = xml_tree.find("c:product", NS)
xml_name = xml_product.findtext("c:name", namespaces=NS)
xml_price = xml_product.findtext("c:price", namespaces=NS)

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, urljoin, urlsplit, urlunsplit
from uuid import uuid4

import requests
from bs4 import BeautifulSoup
from lxml import etree

MAX_BYTES = 2_000_000
NS = {"c": "urn:demo:catalog"}
BASE = "https://catalog.example.invalid/"
HTML = b'''<!doctype html><html><body><main id="catalog">
<article class="product" data-sku="SKU-DEMO-01">
<h2><a href="/p/coffee">Demo Coffee</a></h2>
<span class="price" data-currency="USD">24.90</span><p class="stock">in_stock</p></article>
<article class="product" data-sku="SKU-DEMO-02">
<h2><a href="/p/tea">Demo Tea</a></h2>
<span class="price" data-currency="USD">18.50</span></article>
</main></body></html>'''
XML = b'''<?xml version="1.0" encoding="UTF-8"?>
<catalog xmlns="urn:demo:catalog"><product sku="SKU-DEMO-01">
<name>Demo Coffee</name><price currency="USD">24.90</price>
<path>/p/coffee</path><stock>in_stock</stock></product>
<product sku="SKU-DEMO-02"><name>Demo Tea</name>
<price currency="USD">18.50</price><path>/p/tea</path></product></catalog>'''


def only(nodes):
    if len(nodes) != 1:
        raise ValueError("Expected exactly one required field or container")
    return nodes[0]


def parse_html(data):
    root = only(BeautifulSoup(data, "html.parser").select("main#catalog"))
    cards = root.select("article.product")
    records = []
    for card in cards:
        link = only(card.select("h2 a"))
        price = only(card.select(".price"))
        stock = card.select(".stock")
        if len(stock) > 1:
            raise ValueError("Ambiguous stock field")
        records.append({"sku": card.get("data-sku", ""),
                        "name": link.get_text(" ", strip=True),
                        "amount": price.get_text(" ", strip=True),
                        "currency": price.get("data-currency", ""),
                        "path": link.get("href", ""),
                        "stock": stock[0].get_text(" ", strip=True) if stock else "unknown"})
    return records


def parse_xml(data):
    parser = etree.XMLParser(resolve_entities=False, load_dtd=False, no_network=True,
                             recover=False, huge_tree=False)
    root = etree.fromstring(data, parser=parser)
    if root.getroottree().docinfo.doctype:
        raise ValueError("DTD-bearing XML is outside this tutorial contract")
    if root.tag != "{urn:demo:catalog}catalog":
        raise ValueError("Unexpected XML root or namespace")
    def text(node):
        return " ".join(part.strip() for part in node.itertext() if part.strip())
    records = []
    for product in root.findall("c:product", NS):
        name = only(product.findall("c:name", NS))
        price = only(product.findall("c:price", NS))
        path = only(product.findall("c:path", NS))
        stock = product.findall("c:stock", NS)
        if len(stock) > 1:
            raise ValueError("Ambiguous stock field")
        records.append({"sku": product.get("sku", ""), "name": text(name),
                        "amount": text(price), "currency": price.get("currency", ""),
                        "path": text(path), "stock": text(stock[0]) if stock else "unknown"})
    return records


def normalize(records, base_url, kind, fmt, run_id, stamp):
    if not records:
        raise ValueError("No product records; inspect the source instead of claiming sold out")
    rows, seen = [], set()
    for raw in records:
        row = {key: str(value).strip() for key, value in raw.items()}
        if not row["sku"] or not row["name"] or not row["path"] or row["sku"] in seen:
            raise ValueError("Missing required identity/text or duplicate SKU")
        seen.add(row["sku"])
        if not re.fullmatch(r"\d+(?:\.\d+)?", row["amount"]) or Decimal(row["amount"]) <= 0:
            raise ValueError("Require an explicit positive decimal amount")
        if not re.fullmatch(r"[A-Z]{3}", row["currency"]):
            raise ValueError("Require an explicit three-letter uppercase currency label")
        target = urljoin(base_url, row.pop("path"))
        parsed = urlsplit(target)
        if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
            raise ValueError("Expected HTTPS product URL without embedded credentials")
        row["url"] = target
        row["amount"] = str(Decimal(row["amount"]))
        row["review_reason"] = ""
        if row["stock"] not in {"in_stock", "out_of_stock"}:
            row["stock"], row["review_reason"] = "unknown", "stock_unknown"
        row.update(source_kind=kind, source_format=fmt, run_id=run_id, retrieved_at_utc=stamp)
        rows.append(row)
    return rows


def public_source(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 source without credentials")
    host = p.hostname.lower().rstrip(".")
    if host == "localhost" or host.endswith(".localhost"):
        raise ValueError("Use local fixtures, not network mode, for localhost")
    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 outside network mode")
    return p


def make_session(use_proxy):
    session = requests.Session()
    session.trust_env = False
    session.headers["User-Agent"] = "AuthorizedMarkupTutorial/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 issued proxy host 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_body(session, url, accepted):
    with session.get(url, timeout=(5, 20), allow_redirects=False, stream=True) as response:
        if response.status_code != 200:
            raise ValueError(f"HTTP {response.status_code}: stop and review access/quota")
        mime = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
        if mime not in accepted:
            raise ValueError("Unexpected response type; check for a changed source or challenge")
        data = bytearray()
        for chunk in response.iter_content(65536):
            data.extend(chunk)
            if len(data) > MAX_BYTES:
                raise ValueError("Response exceeds tutorial limit")
        return bytes(data)


def check_exit(session):
    body = get_body(session, "https://api.ipify.org?format=json", {"application/json"})
    address = ipaddress.ip_address(json.loads(body)["ip"])
    if not address.is_global:
        raise ValueError("Checker returned a non-public address")
    return {"observed_ip": str(address), "checked_at_utc": datetime.now(timezone.utc).isoformat()}


def safe_diagnostic(error):
    if isinstance(error, etree.XMLSyntaxError):
        return "XML_SYNTAX: inspect XML well-formedness privately."
    if isinstance(error, requests.RequestException):
        return "NETWORK: inspect connection, proxy authentication and TLS privately."
    if isinstance(error, OSError):
        return "OUTPUT_IO: check destination and write permissions."
    if isinstance(error, KeyError):
        return "CONFIG_FIELD: check required configuration or checker fields."
    message = str(error)
    status = re.fullmatch(r"HTTP (\d{3}): stop and review access/quota", message)
    if status:
        return "HTTP_STATUS_" + status[1] + ": review authorization, quota or source status."
    codes = {
        "Unexpected response type; check for a changed source or challenge": "RESPONSE_MIME",
        "Response exceeds tutorial limit": "INPUT_SIZE",
        "Input exceeds tutorial limit": "INPUT_SIZE",
        "Unexpected XML root or namespace": "XML_NAMESPACE",
        "DTD-bearing XML is outside this tutorial contract": "XML_DTD",
        "Expected exactly one required field or container": "FIELD_CARDINALITY",
        "Ambiguous stock field": "FIELD_CARDINALITY",
        "No product records; inspect the source instead of claiming sold out": "NO_RECORDS",
        "Missing required identity/text or duplicate SKU": "FIELD_IDENTITY",
        "Require an explicit positive decimal amount": "FIELD_AMOUNT",
        "Require an explicit three-letter uppercase currency label": "FIELD_CURRENCY",
        "Expected HTTPS product URL without embedded credentials": "FIELD_URL",
    }
    return codes.get(message, "INPUT_OR_CONFIG") + ": inspect the corresponding input or configuration privately."


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--format", choices=["html", "xml"], required=True)
    parser.add_argument("--url")
    parser.add_argument("--approved", action="store_true")
    parser.add_argument("--proxy", action="store_true")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    log_path = args.out.with_suffix(".run.json")
    if args.out.exists() or log_path.exists():
        parser.error("Use new output filenames")
    run_id = uuid4().hex
    log = {"run_id": run_id, "format": args.format, "proxy_requested": args.proxy}
    if args.url:
        p = public_source(args.url)
        if not args.approved:
            parser.error("Review source permission and request budget before --approved")
        accepted = {"text/html"} if args.format == "html" else {"application/xml", "text/xml"}
        with make_session(args.proxy) as session:
            log["before"] = check_exit(session)
            data = get_body(session, args.url, accepted)
            stamp = datetime.now(timezone.utc).isoformat()
            log["after"] = check_exit(session)
        log["same_observed_ip"] = log["before"]["observed_ip"] == log["after"]["observed_ip"]
        base, kind = args.url, "user_approved_source"
        log["source"] = urlunsplit((p.scheme, p.netloc, p.path, "", ""))
    else:
        if args.proxy:
            parser.error("A local fixture is not a proxy test")
        data = HTML if args.format == "html" else XML
        base, kind = BASE, "synthetic_fixture"
        stamp = datetime.now(timezone.utc).isoformat()
        log["source"] = "local:embedded-" + args.format
    if len(data) > MAX_BYTES:
        raise ValueError("Input exceeds tutorial limit")
    records = parse_html(data) if args.format == "html" else parse_xml(data)
    rows = normalize(records, base, kind, args.format, run_id, stamp)
    log.update(source_kind=kind, records_exported=len(rows), retrieved_at_utc=stamp)
    with log_path.open("x", encoding="utf-8") as handle:
        json.dump(log, handle, indent=2)
    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)} {args.format.upper()} records; inspect review_reason.")


if __name__ == "__main__":
    try:
        main()
    except (ValueError, etree.XMLSyntaxError, KeyError, OSError, requests.RequestException) as error:
        print("Stopped:", safe_diagnostic(error), file=sys.stderr)
        sys.exit(1)

.\.venv\Scripts\python.exe markup_scraper.py --format html --out products-html.csv
.\.venv\Scripts\python.exe markup_scraper.py --format xml --out products-xml.csv

.venv/bin/python markup_scraper.py --format html --out products-html.csv
.venv/bin/python markup_scraper.py --format xml --out products-xml.csv

.\.venv\Scripts\python.exe markup_scraper.py --format html --url "https://catalog.example.invalid/approved-products" --approved --out approved-html.csv
.\.venv\Scripts\python.exe markup_scraper.py --format xml --url "https://catalog.example.invalid/approved-export.xml" --approved --out approved-xml.csv

.venv/bin/python markup_scraper.py --format html --url "https://catalog.example.invalid/approved-products" --approved --out approved-html.csv
.venv/bin/python markup_scraper.py --format xml --url "https://catalog.example.invalid/approved-export.xml" --approved --out approved-xml.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'
$markupProxySecret = Read-Host 'Issued proxy password' -AsSecureString
$env:SOCKS5IO_PASSWORD = [System.Net.NetworkCredential]::new('', $markupProxySecret).Password

.\.venv\Scripts\python.exe markup_scraper.py --format xml --url "https://catalog.example.invalid/approved-export.xml" --approved --proxy --out proxy-xml.csv

read -r -p 'Issued proxy host: ' SOCKS5IO_HOST
read -r -p 'Issued proxy port: ' SOCKS5IO_PORT
read -r -p 'Issued proxy username: ' SOCKS5IO_USER
read -r -s -p 'Issued proxy password: ' SOCKS5IO_PASSWORD; printf '\n'
export SOCKS5IO_HOST SOCKS5IO_PORT SOCKS5IO_USER SOCKS5IO_PASSWORD
.venv/bin/python markup_scraper.py --format xml --url "https://catalog.example.invalid/approved-export.xml" --approved --proxy --out proxy-xml.csv
unset SOCKS5IO_PASSWORD

Remove-Item Env:SOCKS5IO_PASSWORD
Remove-Variable markupProxySecret