From c3211f0602fc5fd29652f095702e422138e39c6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Thu, 4 Jun 2026 10:48:57 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=BC=D0=BD=D0=BE=D0=B3=D0=BE=D0=BF?= =?UTF-8?q?=D0=BE=D1=82=D0=BE=D1=87=D0=BD=D1=8B=D0=B9=20=D0=BF=D0=B0=D1=80?= =?UTF-8?q?=D1=81=D0=B8=D0=BD=D0=B3=20+=20forum=20survey?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Throttle: Thread-safe координация задержек - BanDetector: автоопределение бана - fetch: thread-local сессии вместо глобальной - run: ThreadPoolExecutor + диапазоны страниц + мерж - output/state: поддержка worker-папок и merge_states() - __main__: --workers, --no-delay, --no-ban-test - VWTS/forums-survey.md: обзор 20+ автофорумов --- VWTS/forums-survey.md | 32 ++++++ vwts_scraper/__main__.py | 33 +++++- vwts_scraper/config.py | 2 + vwts_scraper/fetch.py | 29 ++++- vwts_scraper/output.py | 13 ++- vwts_scraper/run.py | 239 ++++++++++++++++++++++++++------------- vwts_scraper/state.py | 28 ++++- vwts_scraper/throttle.py | 78 +++++++++++++ 8 files changed, 361 insertions(+), 93 deletions(-) create mode 100644 VWTS/forums-survey.md create mode 100644 vwts_scraper/throttle.py diff --git a/VWTS/forums-survey.md b/VWTS/forums-survey.md new file mode 100644 index 0000000..596ac69 --- /dev/null +++ b/VWTS/forums-survey.md @@ -0,0 +1,32 @@ +# Обзор автомобильных форумов РФ и мира + +Дата: 2026-06-04 +Цель: найти живые автофорумы с доступом для парсинга (RAG-база знаний по ремонту). + +## Результаты проверки + +| Форум | Марка | Движок | Доступ | Примечание | +|-------|-------|--------|--------|------------| +| **vwts.ru** | VW/Audi/Skoda/SEAT | **XenForo** | ✅ | Отлично, наш основной источник | +| **lexus-club.ru** | Lexus/Toyota | phpBB | ✅ | Большой, активный | +| **renault-club.ru** | Renault | vBulletin 3.6 | ✅ | Активный, 84 раздела | +| ffclub.ru | Ford | Invision Power Board | 🟡 | Жив, другой движок | +| kia-forums.com | Kia (en) | XenForo-like | 🟡 | 741K сообщений, но сабфорумы timeout | +| kianiroforum.com | Kia Niro (en) | XenForo-like | 🟡 | Малый, timeout | +| lada-* (все) | Lada | — | ❌ | Все домены мертвы | +| kia-rio.net | Kia (ru) | — | ❌ | CloudFlare | +| bmwclub.ru | BMW | — | ❌ | DDoS-Guard | +| motor-talk.de | VAG (de) | — | ❌ | CloudFlare | +| kia-forum.de | Kia (de) | — | ❌ | Недоступен | +| toyota-club.net | Toyota (ru) | — | ❌ | 403 | +| toyota-forum.de | Toyota (de) | — | ❌ | CloudFlare | + +## Выводы + +1. **XenForo с открытым доступом — только vwts.ru.** Остальные либо под защитой, либо мёртвы. +2. **Живые форумы на других движках** — потребуют отдельных парсеров: + - phpBB (`viewtopic.php`) — lexus-club.ru + - vBulletin (`forumdisplay.php`) — renault-club.ru + - Invision Power Board (`ipsPagination`) — ffclub.ru +3. **Англоязычные Kia-форумы** — крупные (kia-forums.com, 741K сообщений), но сабфорумы не грузятся с нашего IP (возможно гео- или CDN-ограничение). +4. **Российские автофорумы массово под CloudFlare/DDoS-Guard** — парсинг через requests невозможен. diff --git a/vwts_scraper/__main__.py b/vwts_scraper/__main__.py index b1009dd..fa45e3d 100644 --- a/vwts_scraper/__main__.py +++ b/vwts_scraper/__main__.py @@ -2,13 +2,15 @@ Использование: python -m vwts_scraper https://vwts.ru/forum/vag/benzinovye-dvigateli/ + python -m vwts_scraper URL --workers 8 --no-delay + python -m vwts_scraper URL --workers 1 # однопоточно (как раньше) Пауза: Ctrl+C → состояние сохраняется. Продолжить: та же команда. """ import sys, logging -from .config import OUTPUT_DIR +from .config import OUTPUT_DIR, WORKERS OUTPUT_DIR.mkdir(parents=True, exist_ok=True) @@ -23,14 +25,37 @@ logging.basicConfig( ) if __name__ == "__main__": - if len(sys.argv) < 2 or "vwts.ru" not in sys.argv[1]: + # ── Парсинг аргументов ────────────────────────────────────────── + args = sys.argv[1:] + if len(args) < 1 or "vwts.ru" not in args[0]: print("❌ Укажи URL раздела vwts.ru. Пример:") print(" python -m vwts_scraper " "https://vwts.ru/forum/vag/benzinovye-dvigateli/") + print("\n⚙️ Опции:") + print(" --workers N количество потоков (по умолчанию 4)") + print(" --no-delay без координации задержек") + print(" --no-ban-test не проверять бан") sys.exit(1) + section_url = args[0].rstrip("/") + "/" + workers = WORKERS + no_delay = False + ban_test = True + + for a in args[1:]: + if a == "--no-delay": + no_delay = True + elif a == "--no-ban-test": + ban_test = False + elif a == "--workers" or a == "-w": + idx = args.index(a) + if idx + 1 < len(args): + try: + workers = int(args[idx + 1]) + except ValueError: + pass + from .run import run - section_url = sys.argv[1].rstrip("/") + "/" log = logging.getLogger(__name__) log.info("=" * 60) @@ -40,7 +65,7 @@ if __name__ == "__main__": log.info("=" * 60) try: - run(section_url) + run(section_url, workers=workers, no_delay=no_delay, ban_test=ban_test) except KeyboardInterrupt: log.info("\n🛑 ПАУЗА. Продолжить: та же команда.") sys.exit(0) diff --git a/vwts_scraper/config.py b/vwts_scraper/config.py index 5df7f32..e9722b8 100644 --- a/vwts_scraper/config.py +++ b/vwts_scraper/config.py @@ -10,6 +10,8 @@ OUTPUT_DIR = Path("vwts_data") # куда сохраняем данны DELAY = 1.5 # секунд между запросами TIMEOUT = (10, 30) # (connect, read) TOPICS_PER_FILE = 100 +WORKERS = 4 # потоков по умолчанию +BAN_TEST = 5 # запросов для автоопределения бана HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } diff --git a/vwts_scraper/fetch.py b/vwts_scraper/fetch.py index 5480866..fb7181c 100644 --- a/vwts_scraper/fetch.py +++ b/vwts_scraper/fetch.py @@ -7,13 +7,32 @@ from .config import DELAY, TIMEOUT, HEADERS log = logging.getLogger(__name__) -session = requests.Session() -session.headers.update(HEADERS) +# Глобальный throttle — устанавливается из run() +throttle = None -def get(url: str) -> BeautifulSoup | None: - """GET + ретраи (до 3). 404 — без ретрая, сразу None.""" - time.sleep(DELAY) +def make_session() -> requests.Session: + """Создать новую сессию (thread-safe).""" + s = requests.Session() + s.headers.update(HEADERS) + return s + + +def get(url: str, session: requests.Session = None) -> BeautifulSoup | None: + """GET + ретраи (до 3). 404 — без ретрая, сразу None. + + Args: + url: URL для загрузки + session: опциональная сессия (если None — создаётся временная) + """ + if session is None: + session = make_session() + + if throttle: + throttle.wait() + else: + time.sleep(DELAY) + for n in range(3): try: r = session.get(url, timeout=TIMEOUT) diff --git a/vwts_scraper/output.py b/vwts_scraper/output.py index 64e128b..7cd7183 100644 --- a/vwts_scraper/output.py +++ b/vwts_scraper/output.py @@ -2,14 +2,21 @@ import json, logging from datetime import datetime +from pathlib import Path from .config import TOPICS_PER_FILE, OUTPUT_DIR log = logging.getLogger(__name__) def save_topic(section_slug: str, section_name: str, topic: dict, - posts: list, tags: list, state: dict): - """Записать тему в part_XXXX.jsonl. Ротирует файл каждые N тем.""" + posts: list, tags: list, state: dict, + worker_dir: Path = None): + """Записать тему в part_XXXX.jsonl. Ротирует файл каждые N тем. + + Args: + worker_dir: если задано — писать в эту папку (для многопоточности), + иначе в OUTPUT_DIR / section_slug + """ st = state # ротация @@ -27,7 +34,7 @@ def save_topic(section_slug: str, section_name: str, topic: dict, "scraped_at": datetime.now().isoformat(), } - d = OUTPUT_DIR / section_slug + d = worker_dir if worker_dir else (OUTPUT_DIR / section_slug) d.mkdir(parents=True, exist_ok=True) fpath = d / f"part_{st['file_index']:04d}.jsonl" diff --git a/vwts_scraper/run.py b/vwts_scraper/run.py index b34e0b6..ccb0777 100644 --- a/vwts_scraper/run.py +++ b/vwts_scraper/run.py @@ -1,35 +1,148 @@ -"""Основной цикл: обход страниц раздела → темы → посты → сохранение.""" +"""Основной цикл: ThreadPoolExecutor + диапазоны страниц → мерж.""" -import sys, logging +import sys, logging, shutil, json +from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from urllib.parse import urlparse -from .config import OUTPUT_DIR -from .fetch import get, session +from .config import OUTPUT_DIR, WORKERS, BAN_TEST, DELAY +from .fetch import get, make_session from .paginate import count, verify_last from .parse import parse_topics, parse_posts, parse_tags -from .state import load, save +from .state import load, save, merge_states from .output import save_topic +from .throttle import Throttle, BanDetector log = logging.getLogger(__name__) -def run(section_url: str): - """Главный цикл скрейпера. +# ── Глобальный throttle (устанавливается в run()) ───────────────────── +_throttle = None +import fetch as _fetch + + +def _worker(section_url: str, section_slug: str, section_name: str, + pages: range, worker_id: int): + """Один поток: обходит свой диапазон страниц, пишет в worker_N/.""" + session = make_session() + st = {"topics_done": {}, "pages_done": [], "file_index": 0, "topic_count": 0} + base = f"{urlparse(section_url).scheme}://{urlparse(section_url).netloc}" + consecutive_404 = 0 + + # Папка воркера + worker_dir = OUTPUT_DIR / section_slug / f"worker_{worker_id}" + worker_dir.mkdir(parents=True, exist_ok=True) + + for pg in pages: + url = section_url if pg == 1 else f"{section_url}page-{pg}" + log.info(f"[W{worker_id}] 📄 стр.{pg}") + + page_soup = get(url, session) + if not page_soup: + consecutive_404 += 1 + if consecutive_404 >= 3: + log.info(f"[W{worker_id}] ⏹️ 3 пустых — завершаем") + break + continue + consecutive_404 = 0 + + topics = parse_topics(page_soup) + log.info(f"[W{worker_id}] Тем: {len(topics)}") + + for i, tp in enumerate(topics, 1): + tid = tp["id"] + if str(tid) in st["topics_done"]: + continue + + log.info(f"[W{worker_id}] [{i}/{len(topics)}] #{tid}: {tp['title'][:80]}") + topic_soup = get(tp["url"], session) + if not topic_soup: + st["topics_done"][str(tid)] = tp["title"] + continue + + topic_pages = count(topic_soup) + posts, tags = [], parse_tags(topic_soup) + + for tpp in range(1, topic_pages + 1): + if tpp == 1: + posts += parse_posts(topic_soup) + else: + tp2 = get(f"{base}/forum/topic/{tid}/page-{tpp}", session) + if tp2: + posts += parse_posts(tp2) + + log.info(f"[W{worker_id}] {len(posts)} постов ({topic_pages} стр.), теги: {tags}") + save_topic(section_slug, section_name, tp, posts, tags, st, + worker_dir=worker_dir) + st["topics_done"][str(tid)] = tp["title"] + + st["pages_done"].append(pg) + + # Сохраняем состояние воркера + _state_file = worker_dir / "state.json" + _state_file.write_text(json.dumps(st, ensure_ascii=False, indent=2), + encoding="utf-8") + session.close() + return worker_id, st["topic_count"] + + +def _merge_workers(section_slug: str): + """Собрать part_*.jsonl из всех worker_N/ в корень раздела с единой нумерацией.""" + root = OUTPUT_DIR / section_slug + + # Собираем все part_*.jsonl из worker_N/ + all_parts = [] + for wdir in sorted(root.glob("worker_*")): + if wdir.is_dir(): + for f in sorted(wdir.glob("part_*.jsonl")): + all_parts.append(f) + + log.info(f"🔗 Мерж {len(all_parts)} файлов из {len(list(root.glob('worker_*')))} воркеров") + + # Переименовываем с единой нумерацией + for idx, src in enumerate(all_parts): + dst = root / f"part_{idx:04d}.jsonl" + shutil.move(str(src), str(dst)) + + # Удаляем пустые worker-папки + for wdir in root.glob("worker_*"): + if wdir.is_dir(): + try: + wdir.rmdir() + except OSError: + pass + + # Пишем объединённый state.json + merge_states(section_slug) + + log.info(f"✅ Мерж: {len(all_parts)} файлов → {root}") + + +def run(section_url: str, workers: int = WORKERS, + no_delay: bool = False, ban_test: bool = True): + """Главный цикл с многопоточностью. Args: - section_url: полный URL раздела, напр. - https://vwts.ru/forum/vag/benzinovye-dvigateli/ + section_url: полный URL раздела + workers: количество потоков + no_delay: True = без координации задержек + ban_test: True = проверить бан перед стартом """ section_slug = section_url.rstrip("/").split("/")[-1] - # загружаем состояние (если было прервано — продолжим) - st = load(section_slug) - if st["topics_done"]: - log.info(f"🔄 Продолжаем: {len(st['topics_done'])} тем, " - f"part_{st['file_index']:04d}.jsonl") + # ── Автоопределение бана ────────────────────────────────────────── + throttle_enabled = not no_delay + if ban_test and throttle_enabled and workers > 1: + if not BanDetector.test(section_url, BAN_TEST): + throttle_enabled = False - # ── Страница 1: название раздела + количество страниц ──────────────── - soup = get(section_url) + global _throttle + _throttle = Throttle(DELAY, enabled=throttle_enabled) + _fetch.throttle = _throttle + + log.info(f"⚙️ Потоков: {workers}, координация: {'вкл' if throttle_enabled else 'выкл'}") + + # ── Страница 1: название + количество страниц ───────────────────── + soup = get(section_url, make_session()) if not soup: log.error("❌ Сайт недоступен") sys.exit(1) @@ -38,78 +151,44 @@ def run(section_url: str): section_name = section_name.text.strip() if section_name else section_slug total_pages = count(soup) - # Проверяем: последняя страница реально существует? - total_pages = verify_last(section_url, total_pages, session) + total_pages = verify_last(section_url, total_pages, make_session()) log.info(f"📋 {section_name} | страниц: {total_pages}") - # ── Обход страниц раздела ─────────────────────────────────────────── - consecutive_404 = 0 + # ── Диапазоны страниц ───────────────────────────────────────────── + if workers > total_pages: + workers = total_pages - for pg in range(1, total_pages + 1): - if pg in st["pages_done"]: - consecutive_404 = 0 - continue + base_size = total_pages // workers + remainder = total_pages % workers + ranges = [] + start = 1 + for w in range(workers): + size = base_size + (1 if w < remainder else 0) + ranges.append(range(start, start + size)) + start += size - # page-1 = базовый URL (чтобы не было лишнего редиректа) - url = section_url if pg == 1 else f"{section_url}page-{pg}" - log.info(f"📄 [{pg}/{total_pages}]") + log.info(f"📦 Диапазоны: {[f'{r.start}-{r[-1]}' for r in ranges]}") - page_soup = get(url) - if not page_soup: - consecutive_404 += 1 - # 3 пустых страницы подряд = раздел закончился раньше - if consecutive_404 >= 3: - log.info(f" ⏹️ {consecutive_404} пустых подряд — завершаем") - break - log.warning(f" ⚠️ Пропущена стр.{pg} " - f"(пустых подряд: {consecutive_404})") - continue - consecutive_404 = 0 + # ── Запуск потоков ──────────────────────────────────────────────── + total_topics = 0 + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = [ + pool.submit(_worker, section_url, section_slug, section_name, + rng, i) + for i, rng in enumerate(ranges) + ] + for fut in as_completed(futures): + wid, cnt = fut.result() + total_topics += cnt + log.info(f"[W{wid}] ✅ завершён: {cnt} тем") - topics = parse_topics(page_soup) - log.info(f" Тем: {len(topics)}") + # ── Мерж ────────────────────────────────────────────────────────── + _merge_workers(section_slug) - # ── Обход тем на странице ────────────────────────────────────── - for i, tp in enumerate(topics, 1): - tid = tp["id"] - if str(tid) in st["topics_done"]: - continue - - log.info(f" [{i}/{len(topics)}] #{tid}: {tp['title'][:80]}") - topic_soup = get(tp["url"]) - if not topic_soup: - # не удалось — помечаем чтобы не ретраить бесконечно - st["topics_done"][str(tid)] = tp["title"] - save(section_slug, st) - continue - - topic_pages = count(topic_soup) - posts, tags = [], parse_tags(topic_soup) - - # ── Обход страниц темы ──────────────────────────────────── - base = f"{urlparse(section_url).scheme}://{urlparse(section_url).netloc}" - - for tpp in range(1, topic_pages + 1): - if tpp == 1: - posts += parse_posts(topic_soup) - else: - tp2 = get(f"{base}/forum/topic/{tid}/page-{tpp}") - if tp2: - posts += parse_posts(tp2) - - log.info(f" {len(posts)} постов ({topic_pages} стр.), " - f"теги: {tags}") - save_topic(section_slug, section_name, tp, posts, tags, st) - st["topics_done"][str(tid)] = tp["title"] - save(section_slug, st) - - st["pages_done"].append(pg) - save(section_slug, st) - - # ── Итог ──────────────────────────────────────────────────────────── + # ── Итог ────────────────────────────────────────────────────────── log.info("=" * 60) - log.info(f"✅ '{section_name}' — {st['topic_count']} тем " - f"в {st['file_index']+1} частях") + log.info(f"✅ '{section_name}' — {total_topics} тем " + f"в {len(list((OUTPUT_DIR/section_slug).glob('part_*.jsonl')))} частях") for f in sorted((OUTPUT_DIR / section_slug).glob("part_*.jsonl")): log.info(f" 📄 {f.name} ({f.stat().st_size/1024/1024:.1f} MB)") log.info("=" * 60) diff --git a/vwts_scraper/state.py b/vwts_scraper/state.py index 6917c95..b56cdbf 100644 --- a/vwts_scraper/state.py +++ b/vwts_scraper/state.py @@ -1,8 +1,11 @@ """Управление состоянием (state.json в папке раздела).""" -import json +import json, logging +from pathlib import Path from .config import OUTPUT_DIR +log = logging.getLogger(__name__) + def _state_file(section_slug: str) -> Path: d = OUTPUT_DIR / section_slug @@ -20,3 +23,26 @@ def load(section_slug: str) -> dict: def save(section_slug: str, st: dict): _state_file(section_slug).write_text( json.dumps(st, ensure_ascii=False, indent=2), encoding="utf-8") + + +def merge_states(section_slug: str): + """Собрать состояния всех воркеров в единый state.json.""" + root = OUTPUT_DIR / section_slug + merged = {"topics_done": {}, "pages_done": [], "file_index": 0, "topic_count": 0} + + for wdir in sorted(root.glob("worker_*/state.json")): + try: + wst = json.loads(wdir.read_text(encoding="utf-8")) + merged["topics_done"].update(wst.get("topics_done", {})) + merged["pages_done"].extend(wst.get("pages_done", [])) + merged["topic_count"] += wst.get("topic_count", 0) + except Exception as e: + log.warning(f"⚠️ Ошибка чтения {wdir}: {e}") + + merged["pages_done"] = sorted(set(merged["pages_done"])) + merged["file_index"] = len(list(root.glob("part_*.jsonl"))) - 1 + if merged["file_index"] < 0: + merged["file_index"] = 0 + + _state_file(section_slug).write_text( + json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8") diff --git a/vwts_scraper/throttle.py b/vwts_scraper/throttle.py new file mode 100644 index 0000000..d0f24c5 --- /dev/null +++ b/vwts_scraper/throttle.py @@ -0,0 +1,78 @@ +"""Общая координация задержек и автоопределение бана.""" + +import time, threading, logging +import requests +from .config import DELAY, TIMEOUT, HEADERS + +log = logging.getLogger(__name__) + + +class Throttle: + """Thread-safe координатор задержек. + + Все потоки ждут своей очереди — максимум 1 запрос в DELAY секунд. + """ + + def __init__(self, delay: float = DELAY, enabled: bool = True): + self._delay = delay + self._enabled = enabled + self._lock = threading.Lock() + self._last = 0.0 # монотонное время последнего запроса + + @property + def enabled(self) -> bool: + return self._enabled + + def wait(self): + """Ждать своей очереди. Без координации — просто sleep(delay).""" + if not self._enabled: + time.sleep(self._delay) + return + + with self._lock: + elapsed = time.monotonic() - self._last + if elapsed < self._delay: + time.sleep(self._delay - elapsed) + self._last = time.monotonic() + + def disable(self): + self._enabled = False + + +class BanDetector: + """Проверяет, банит ли сервер за быстрые запросы.""" + + @staticmethod + def test(base_url: str, count: int = 5) -> bool: + """Шлёт count запросов без задержки. Возвращает True если банят. + + Args: + base_url: любой URL того же хоста (напр. главная раздела) + count: сколько запросов подряд + + Returns: + True — банит (есть 403/429), False — не банит + """ + log.info(f"🔍 Проверка бана: {count} запросов подряд...") + s = requests.Session() + s.headers.update(HEADERS) + + banned = False + for i in range(count): + try: + r = s.get(base_url, timeout=TIMEOUT) + log.info(f" [{i+1}/{count}] HTTP {r.status_code}") + if r.status_code in (403, 429): + banned = True + break + except Exception as e: + log.warning(f" [{i+1}/{count}] ошибка: {e}") + banned = True + break + + if banned: + log.warning("⚠️ Сервер банит быстрые запросы — включаю координацию") + else: + log.info("✅ Бан не обнаружен — без координации") + s.close() + return banned