feat: многопоточный парсинг + forum survey

- 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+ автофорумов
2026-06-04 10:48:57 +03:00
parent 22ca83adf8
commit c3211f0602
8 changed files with 361 additions and 93 deletions
@@ -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)