Парсинг XML и HTML на Python: извлечение данных, CSV и настройка прокси

Парсинг XML и HTML начинается с получения документа, но дальше требует разных подходов. В HTML нужные данные находят по элементам и атрибутам страницы. В XML учитывают иерархию элементов и пространства имён. В обоих случаях перед выгрузкой нужно проверить значения: найденный элемент ещё не означает, что получены правильные данные.

В этом руководстве один скрипт извлекает два вымышленных товара из HTML и XML и сохраняет их в CSV с одинаковым набором полей. Сначала запустим встроенные примеры без сети, затем разберём адаптацию к разрешённому источнику и подключение Socks5.IO, если задаче нужен определённый сетевой маршрут.

Главное в трёх пунктах:

• Разделяйте разбор форматов. Для HTML используйте подходящие селекторы, для XML — пути с учётом пространств имён.
• Проверяйте данные до сохранения. Не подменяйте неизвестный остаток предположением о наличии товара.
• Настраивайте прокси отдельно от парсера. Другой выходной IP не исправит селектор, пространство имён или отсутствующее поле.

Полный код содержит исходные примеры; скачивать отдельный пакет не нужно. Для быстрого перехода найдите разделы «Полный скрипт», «Запуск и проверка CSV», «Настройка SOCKS5-прокси» или «Ошибки парсинга».

Чем отличается XML от HTML при парсинге?

HTML описывает структуру и смысл элементов веб-страницы, а XML позволяет задавать собственную структуру документов и обменных данных. Для парсинга важны правила синтаксиса, имена элементов и способ выбора нужных узлов.

HTML не ограничивается отображением: текст, атрибуты, таблицы и встроенные данные тоже содержат информацию для извлечения. XML, в свою очередь, не гарантирует ни полноту записей, ни правильность цен.

Стандарт разбора HTML и спецификация XML W3C задают разные правила обработки документов. Если браузер показывает страницу с ошибками разметки, это не повод принимать неверный результат извлечения.

Отдельно учитывайте XHTML, который сервер отдаёт как XML. Он отличается от обычного text/html. Выбирайте обработку по формату ответа и спецификации источника, а не только по расширению файла.

Какие библиотеки нужны для парсинга на Python?

В примере Requests получает документ, Beautiful Soup извлекает данные из HTML, а lxml разбирает XML. Разделение сетевых запросов, разбора документа и проверки полей упрощает поиск ошибок.

Для HTML укажем стандартный Python-парсер html.parser. Для XML создадим lxml.etree.XMLParser с явными ограничениями. Выбор парсера и работу с деревом описывает документация Beautiful Soup.

Эти парсеры не запускают JavaScript страницы. Если поле появляется только после разрешённого рендеринга или доступно в официальном фиде, потребуется отдельный способ получения данных. Изменение CSS-селектора не создаст значение, которого нет в загруженном ответе.

Парсинг XML и HTML на Python: пример с одинаковыми товарами

Для двух форматов нужны отдельные функции извлечения и общие правила проверки. Локальные примеры позволяют проверить эту логику до подключения сети и прокси.

Используем два вымышленных товара:

• SKU-DEMO-01 — Demo Coffee, цена 24.90 USD, наличие in_stock.
• SKU-DEMO-02 — Demo Tea, цена 18.50 USD, сведения о наличии намеренно отсутствуют.

Для второго товара скрипт сохраняет unknown и добавляет отметку для проверки. Он не объявляет товар доступным или распроданным.

1. Определите поля и правила проверки

Префикс c в коде обозначает пространство имён urn:demo:catalog. Он не обязан совпадать с префиксом в XML. Пространство имён по умолчанию тоже относится к именам элементов, поэтому поиск просто по product не найдёт такой элемент.

2. Подготовьте окружение

Нужен Python 3.10 или новее. Создайте отдельное виртуальное окружение и установите зависимости.

Windows, PowerShell:

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

Все последующие Windows-команды тоже явно вызывают Python из .venv, поэтому активировать окружение не требуется.

macOS/Linux:

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

Диапазоны версий ограничивают зависимости, но не гарантируют одинаковое поведение всех будущих выпусков. Запишите установленные версии и повторяйте проверку примеров после обновления.

3. Сравните CSS-селекторы и пути XML

Фрагмент показывает различие в извлечении полей. Он использует HTML, XML и NS из полного скрипта ниже и не предназначен для отдельного запуска.

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)

HTML-парсер выбирает article.product, затем h2 a и .price. XML-парсер ищет c:product, c:name и c:price с заданным пространством имён.

Здесь нет проверки результатов. Для запуска копируйте полный скрипт: в нём последовательно размещены учебные данные → функции извлечения → нормализация → сетевые проверки → параметры командной строки и экспорт.

4. Полный скрипт

Сохраните код как markup_scraper.py. Без сетевых параметров он работает со встроенными данными: --format html или --format xml выбирает формат. Для сетевого режима нужны --url и --approved.

Названия полей, диагностические коды и сообщения внутри кода сохранены из оригинала, чтобы команды и результаты можно было сопоставить без переименования.

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)

5. Запуск и проверка CSV

Windows, PowerShell:

.\.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

macOS/Linux:

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

Оба запуска должны дать одинаковые товарные поля:

Ссылки на товары будут вести на HTTPS-адреса вымышленного каталога. Служебные поля отличаются:

• source_format указывает HTML или XML.
• run_id и время получения создаются для каждого запуска.
• Все строки из одного ответа получают общую временную метку.
• Пустое review_reason означает только прохождение заданных проверок. Оно не подтверждает актуальный остаток, учёт налогов или совпадение товара между источниками.

Рядом с CSV скрипт сохраняет журнал запуска .run.json. Существующие файлы он не перезаписывает. Запись двух файлов не является атомарной: при сбое один файл может остаться без другого.

Если открываете недоверенный текст в табличном редакторе, используйте импорт как текст, чтобы значения не выполнялись как формулы.

6. Проверьте поведение при ошибках

Меняйте по одному условию и каждый раз задавайте новое имя выходного файла.

После проверки восстановите исходные данные. В этом примере одному SKU соответствует одна запись в ответе, поэтому дубликаты запрещены. Для вариантов товара, разных складов или нескольких предложений сначала определите другой состав уникального идентификатора.

Парсинг XML на Python: пространства имён, XPath и безопасность

Если XML не возвращает ожидаемые элементы, сначала проверьте путь и пространство имён. Корректный синтаксис документа не подтверждает его безопасность, соответствие схеме или правильность бизнес-данных.

Проверяйте URI пространства имён, а не только префикс

В root.findall("c:product", NS) префикс c связан с URI пространства имён. Другой префикс в XML при том же URI обозначает то же полное имя. Изменение URI уже меняет имя элемента, а не только его запись.

В примерах атрибуты sku и currency не имеют пространства имён, а элементы используют пространство имён по умолчанию. Для RSS, Atom и карт сайта проверяйте фактический корневой элемент, иерархию и поля: один набор путей не подходит всем XML-документам.

Различайте findall и XPath

Метод findall использует ограниченный язык путей в стиле ElementTree. Для более сложных запросов в lxml есть отдельный интерфейс XPath. Смена метода поиска не отменяет проверку извлечённых значений.

Ограничивайте обработку недоверенных документов

Руководство Python по безопасности XML описывает риски исчерпания ресурсов. Значение настроек парсера объясняет документация lxml.

В скрипте заданы следующие ограничения:

• resolve_entities=False отключает подстановку сущностей.
• load_dtd=False отключает загрузку внешних DTD.
• no_network=True запрещает сетевой доступ при обращении к внешним документам.
• recover=False отключает автоматическое исправление некорректного XML.
• huge_tree=False сохраняет ограничения парсера для защиты от чрезмерно сложных деревьев.
• Документы с DTD отклоняются; XInclude и XSLT не выполняются.
• Объём получаемого содержимого ограничен 2_000_000 байт.

Это настройки для небольшого учебного примера, а не полноценная изолированная среда обработки. В рабочей системе обновляйте библиотеки и ограничивайте память и время выполнения процесса.

Корректный синтаксис XML не равен проверке по XML Schema. Этот код проверяет ожидаемый корень и обязательные поля, но не проверяет документ по XSD или DTD.

Для больших разрешённых XML-файлов рассмотрите iterparse или другую потоковую обработку. Освобождайте обработанные элементы и ограничивайте объём вывода. Пример строит небольшое дерево в памяти: он не демонстрирует постоянное потребление памяти и не доказывает превосходство одного парсера по скорости.

Как подключить реальный источник данных?

Сначала адаптируйте извлечение к структуре разрешённого источника и проверьте сохранённый ответ. Только после этого меняйте URL и включайте сетевой режим.

Порядок работы:

1. Подтвердите разрешение на получение данных и допустимую частоту запросов.
2. Сохраните пример ответа и зафиксируйте формат, кодировку, корень или контейнер записей.
3. Определите идентификатор записи, обязательные поля и обработку пропусков.
4. Измените соответствующую функцию разбора и проверьте результат на сохранённом документе.
5. Подключите разрешённый URL и отдельно проверьте сетевое получение.

Селектор article.product и пространство имён urn:demo:catalog относятся только к учебным данным. Это не актуальные селекторы коммерческого сайта и не универсальный парсер карты сайта.

Команды ниже — шаблоны с заведомо нерабочим доменом .invalid. После адаптации кода замените его адресом разрешённого источника.

Windows:

.\.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

macOS/Linux:

.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

Сетевой режим принимает только заданные MIME-типы HTML или XML и не следует перенаправлениям автоматически. Для application/xhtml+xml, специальных MIME-типов RSS или другой структуры XML нужна явная адаптация.

У примера есть дополнительные границы:

• Источник должен быть общедоступным HTTPS-ресурсом, получение данных с которого разрешено.
• Код отклоняет localhost и явно указанные непубличные IP-адреса. Это не универсальная защита от SSRF или DNS rebinding.
• Загружается один документ. Скрипт не обходит товарные ссылки, не переключает страницы, не выполняет JavaScript и не ищет закрытые API.
• Относительные ссылки строятся от URL источника. HTML-элемент base не обрабатывается; при необходимости добавьте его поддержку для конкретного сайта.
• XML с сообщением об ошибке может быть синтаксически корректным. Проверки MIME-типа недостаточно: нужны ожидаемый корень и обязательные поля.

Настройка SOCKS5-прокси в Python через Socks5.IO

Socks5.IO подключается на этапе сетевого запроса, когда задаче нужен определённый выходной регион или маршрут. Прокси меняет путь получения документа, но не правила разбора HTML и XML.

Для небольшого разрешённого сбора без требований к маршруту может хватить прямого подключения. Если нужны поддерживаемый выход через домашние подключения или конкретный рынок, рассмотрите ротационные резидентские прокси Socks5.IO.

До использования уточните доступные локации, способ авторизации, поведение сессий, срок действия трафика и условия покупки. Размер IP-пула сам по себе не подтверждает доступность нужного города или конкретного сайта. Прокси также не увеличивает разрешённый источником лимит запросов.

1. Получите адрес и данные авторизации

Используйте официальную инструкцию Socks5.IO по авторизации с логином и паролем на английском. Ориентируйтесь на текущий интерфейс и возможности выбранного продукта.

1. Откройте Rotating Residential IP → Proxy Setup.
2. Выберите авторизацию по учётным данным и нужного прокси-пользователя.
3. Проверьте требования к региону и сессии, выберите поддерживаемый протокол.
4. Сгенерируйте подключение и скопируйте хост, порт, логин и пароль прокси.

При необходимости создайте прокси-пользователя через Sub Account Management. Логин для авторизации на прокси не совпадает по назначению с логином личного кабинета. Используйте именно выданные прокси-реквизиты.

2. Задайте переменные окружения и запустите сбор

В отдельном процессе PowerShell введите выданные значения:

$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

После адаптации парсера и проверки разрешения на доступ запустите источник с флагом --proxy:

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

На macOS/Linux выполните команды в Bash. Пароль вводится отдельно, а не записывается непосредственно в команду:

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

Параметры read рассчитаны на Bash. Если используете другую оболочку, сначала перейдите в Bash.

Скрипт кодирует специальные символы учётных данных и использует socks5h: имя целевого сервера разрешается на стороне прокси. Это поведение описано в разделе Requests о SOCKS. Сам клиент по-прежнему должен разрешить имя прокси-шлюза и подключиться к нему.

Дополнительные настройки подключения:

• Проверка TLS-сертификатов остаётся включённой.
• trust_env=False отключает использование прокси и путей к наборам сертификатов из переменных окружения.
• При ошибке прокси код не переходит автоматически на прямое соединение.
• Если организации нужен собственный удостоверяющий центр, утверждённый набор сертификатов требуется настроить явно.

SOCKS5 не является автоматически более быстрым, анонимным или подходящим для XML, чем HTTP(S). Выбирайте протокол по поддержке сервера и клиента. Подробности приведены в сравнении SOCKS5 и HTTP-прокси.

Подключение браузера требует отдельной проверки. Нельзя считать, что настройки авторизации Requests без изменений подойдут Chromium.

3. Проверьте выходной IP тем же клиентом

В сетевом режиме один объект Session последовательно обращается к публичному сервису ipify, источнику данных и снова к ipify. Проверьте условия использования стороннего сервиса и последствия передачи ему сетевых данных. Если проверка IP не проходит, сбор останавливается.

В журнале запуска посмотрите before, after и same_observed_ip. При необходимости сравните результат с разрешённым прямым подключением. Географию выхода проверяйте отдельно.

Совпадение IP до и после получения документа не доказывает, что запрос к самому источнику прошёл через тот же адрес. При ротации выход может меняться для каждого запроса; один объект Session его не закрепляет.

Если нужен постоянный выход, настройте поддерживаемый фиксированный IP или закрепление сессии — sticky session — и отдельно проверьте выполнение этого требования. Для замены ipify нужны реальный адрес другого разрешённого сервиса и его формат ответа. Не подставляйте вымышленный API проверки Socks5.IO.

4. Защитите журнал и учётные данные

CSV сохраняет товарные URL из источника. В журнале запуска из адреса источника удаляются параметры запроса и фрагмент. Если для воспроизведения нужен полный разрешённый URL, храните его отдельно с ограниченным доступом и связывайте с запуском по run_id.

Не публикуйте токены бронирования, ссылки на аккаунты, персональные данные и учётные данные прокси. Ограничивайте доступ к окружению процесса. После работы удалите пароль; для PowerShell используйте:

Remove-Item Env:SOCKS5IO_PASSWORD
Remove-Variable markupProxySecret

Ошибки парсинга: что проверять перед увеличением нагрузки?

Разделяйте ошибки доступа, формата ответа, разбора, полей и соединения. Начинайте с диагностического кода и проверки ответа источника.

Скрипт выводит заранее определённые диагностические категории вместо полного текста исключений, URL и учётных данных. Например, HTTP_STATUS_403 и HTTP_STATUS_429 различают коды HTTP-ответа, а XML_NAMESPACE указывает на несоответствие ожидаемого корня или пространства имён. Нераспознанные ошибки получают общую категорию.

Подробные ответы исследуйте в закрытой рабочей среде, а не в публичных журналах. Не настраивайте автоматическую смену IP для обхода отказа в доступе и не пытайтесь устранить ошибку 429 неограниченным числом потоков.

У каждого источника должен быть общий лимит запросов для всех обработчиков. Используйте кеширование только там, где оно разрешено; ограничивайте параллелизм, число повторов и полное время задания. timeout=(5, 20) задаёт тайм-ауты подключения и чтения, а не максимальную длительность всей задачи.

Оценивайте качество записей отдельно от успешности запросов

Один ответ может содержать несколько товаров. Поэтому число принятых записей, делённое на число запросов, не является долей успешных запросов.

Отдельно отслеживайте:

• ответы с ожидаемым содержимым;
• ошибки из-за изменений структуры;
• долю записей, которым нужна проверка полей;
• дубликаты и конфликты идентификаторов;
• время, прошедшее с получения данных;
• длительность обработки, расход трафика и относимые к задаче затраты;
• стоимость поддержки кода.

Более широкий разбор сетевых вариантов и расходов есть в руководстве по выбору прокси для парсинга на английском. Качество извлечённых данных всё равно нужно проверять отдельно.

Отличайте локальные тесты от проверки реального подключения

В приложенном к английскому оригиналу отчёте указаны 21 проверка, 0 неудачных проверок и 0 ошибок. HTTP-поведение проверялось с помощью имитаций — mock-объектов. Производительность реальных серверов Socks5.IO не измерялась.

Пример применения: сверка HTML-каталога с XML-выгрузкой поставщика

Сравнивать цены имеет смысл только после согласования идентификаторов товаров, валюты и состава цены. Пустой результат извлечения нельзя автоматически трактовать как отсутствие товара.

Допустим, магазин имеет право ежедневно сопоставлять собственный HTML-каталог с XML-выгрузкой поставщика. Команда выбирает 20 заранее сопоставленных SKU из каждого источника.

Это 40 плановых проверок сочетания «источник—SKU», а не 40 гарантированных запросов или совпавших записей. Один документ может содержать несколько SKU; часть ожидаемых записей может отсутствовать.

Рабочая схема:

1. Отдельные функции извлекают HTML и XML, затем приводят записи к общей структуре.
2. Проверяются валюта, смысл цены и идентификаторы товаров или вариантов.
3. Неизвестный остаток направляется на проверку и не запускает уведомление «нет в наличии».
4. Изменение пространства имён XML считается несовместимостью с текущим кодом, а не удалением всех товаров поставщиком.
5. HTTP 200 со страницей согласия не считается корректным товарным ответом.
6. Прокси с подходящим регионом добавляется только при соответствующем требовании к разрешённому сбору.

До увеличения нагрузки команда оценивает долю пригодных записей и затраты на обслуживание. Это пример организации работы, а не история клиента, измеренная экономия или тест доступности сайтов.

Итог: проверьте результат перед увеличением нагрузки

Надёжный парсинг XML и HTML начинается со структуры источника и правил проверки полей. Прежде чем увеличивать объём сбора, убедитесь, что полученные записи соответствуют этим правилам.

Запустите оба локальных примера, сравните CSV и отметки для проверки. Затем адаптируйте извлечение к одному разрешённому источнику.

Перед расширением сбора также проверьте допустимую частоту запросов, расход ресурсов и затраты.