/
Vibek
/
Agent_GG
Обзор
Документация
Войти
/
Vibek
/
Agent_GG
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
agent/tools/parse_progress.py
258 строк
9 KB
VladHudakov
Новый парсер
19 май 2026, 17:18
19 май 2026, 17:18
8db1c68
Код
Авторство
О чём код?
from __future__ import annotations import re from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass, field from threading import Lock from time import time from typing import Any _TTL_SECONDS = 15 * 60 _active_progress_id: ContextVar[str | None] = ContextVar("gg_parse_progress_id", default=None) _PAGE_START_RE = re.compile(r"\[PARSER\] Старт страницы (?P<page>\d+) \(шаг (?P<idx>\d+)/(?P<total>\d+)") _CARD_PROGRESS_RE = re.compile(r"\[\s*(?P<idx>\d+)/(?P<total>\d+)\]\s+Парсинг:") def _now_ts() -> float: return time() def _clamp_percent(value: int | float | None) -> int: if value is None: return 0 return max(0, min(100, int(round(value)))) @dataclass(slots=True) class ParseProgressState: progress_id: str active: bool = True status: str = "running" percent: int = 5 stage: str = "Запуск парсинга" detail: str = "Подготавливаю источник и параметры парсинга..." created_at: float = field(default_factory=_now_ts) updated_at: float = field(default_factory=_now_ts) meta: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "active": self.active, "status": self.status, "percent": self.percent, "stage": self.stage, "detail": self.detail, "created_at": self.created_at, "updated_at": self.updated_at, } class ParseProgressTracker: def __init__(self) -> None: self._items: dict[str, ParseProgressState] = {} self._lock = Lock() def _cleanup_locked(self, now_ts: float) -> None: stale_ids = [ progress_id for progress_id, state in self._items.items() if (now_ts - state.updated_at) > _TTL_SECONDS ] for progress_id in stale_ids: self._items.pop(progress_id, None) def get(self, progress_id: str) -> dict[str, Any]: now_ts = _now_ts() with self._lock: self._cleanup_locked(now_ts) state = self._items.get(progress_id) if state is None: return { "active": False, "status": "idle", "percent": 0, "stage": "", "detail": "", "created_at": now_ts, "updated_at": now_ts, } return state.to_dict() def start(self, progress_id: str, *, city_key: str, start_page: int, end_page: int) -> None: now_ts = _now_ts() total_pages = max(1, end_page - start_page + 1) with self._lock: self._cleanup_locked(now_ts) self._items[progress_id] = ParseProgressState( progress_id=progress_id, percent=8, stage="Запуск парсинга", detail=f"Город: {city_key}. Страницы {start_page}-{end_page}.", created_at=now_ts, updated_at=now_ts, meta={ "city_key": city_key, "start_page": start_page, "end_page": end_page, "total_pages": total_pages, "current_page_idx": 0, }, ) def _update(self, progress_id: str, **kwargs: Any) -> None: now_ts = _now_ts() with self._lock: self._cleanup_locked(now_ts) state = self._items.get(progress_id) if state is None: return percent = kwargs.get("percent") if percent is not None: state.percent = _clamp_percent(percent) stage = kwargs.get("stage") if stage: state.stage = stage detail = kwargs.get("detail") if detail: state.detail = detail status = kwargs.get("status") if status: state.status = status meta = kwargs.get("meta") if isinstance(meta, dict): state.meta.update(meta) active = kwargs.get("active") if active is not None: state.active = bool(active) state.updated_at = now_ts def complete(self, progress_id: str, *, detail: str) -> None: self._update( progress_id, percent=100, stage="Парсинг завершен", detail=detail, status="completed", active=True, ) def fail(self, progress_id: str, *, detail: str) -> None: self._update( progress_id, percent=100, stage="Ошибка парсинга", detail=detail, status="failed", active=True, ) def update_from_message(self, progress_id: str, message: str) -> None: msg = str(message or "").strip() if not msg: return if "[PARSER] Инициализация webdriver" in msg or "[PARSER] Запуск Chrome" in msg: self._update(progress_id, percent=15, stage="Запуск браузера", detail="Поднимаю WebDriver для парсинга.") return if "[PARSER] WebDriver готов." in msg: self._update(progress_id, percent=22, stage="Браузер готов", detail="WebDriver инициализирован.") return if "[ANTIBAN] Заход на главную Avito" in msg or "[ANTIBAN] Заход на главную Cian" in msg: self._update(progress_id, percent=28, stage="Подготовка сессии", detail="Прогреваю сессию перед парсингом.") return if "[ANTIBAN] Сессия прогрета, начинаем парсинг..." in msg: self._update(progress_id, percent=34, stage="Начинаю парсинг", detail="Переход к страницам выдачи.") return page_match = _PAGE_START_RE.search(msg) if page_match: page_number = int(page_match.group("page")) page_idx = int(page_match.group("idx")) total_pages = max(1, int(page_match.group("total"))) base = 34 span = 50 fraction = (page_idx - 1) / total_pages percent = base + span * fraction self._update( progress_id, percent=percent, stage=f"Обрабатываю страницу {page_number}", detail=f"Шаг {page_idx} из {total_pages}.", meta={"current_page_idx": page_idx, "total_pages": total_pages}, ) return card_match = _CARD_PROGRESS_RE.search(msg) if card_match: card_idx = int(card_match.group("idx")) card_total = max(1, int(card_match.group("total"))) with self._lock: state = self._items.get(progress_id) page_idx = int((state.meta if state else {}).get("current_page_idx") or 1) total_pages = int((state.meta if state else {}).get("total_pages") or 1) base = 34 span = 50 fraction = ((page_idx - 1) + (card_idx / card_total)) / total_pages percent = base + span * fraction self._update( progress_id, percent=percent, stage="Сбор объявлений", detail=f"Карточка {card_idx} из {card_total} на текущей странице.", ) return if "Страница " in msg and "полностью обработана" in msg: with self._lock: state = self._items.get(progress_id) page_idx = int((state.meta if state else {}).get("current_page_idx") or 1) total_pages = int((state.meta if state else {}).get("total_pages") or 1) base = 34 span = 50 fraction = min(1.0, page_idx / total_pages) percent = base + span * fraction self._update( progress_id, percent=percent, stage="Страница завершена", detail="Перехожу к следующему шагу парсинга.", ) return if "[DB] SQLite-буфер сохранен" in msg: self._update(progress_id, percent=96, stage="Сохраняю результаты", detail="Фиксирую сохраненные объявления.") _TRACKER = ParseProgressTracker() def get_parse_progress_tracker() -> ParseProgressTracker: return _TRACKER def get_active_progress_id() -> str | None: return _active_progress_id.get() def report_parser_message(message: str) -> None: progress_id = get_active_progress_id() if not progress_id: return _TRACKER.update_from_message(progress_id, message) @contextmanager def bind_parse_progress(progress_id: str | None): if not progress_id: yield return token = _active_progress_id.set(progress_id) try: yield finally: _active_progress_id.reset(token)