naeel
5c142096de
- core/ — общий слой: fetch, throttle, output, state, runner - forums/base.py — абстрактный BaseAdapter - forums/xenforo/ — адаптер для XenForo (vwts.ru) - forums/phpbb/ — адаптер для phpBB (нива-лада.рф) - __main__.py — точка входа CLI - history/001-initial-structure.md — журнал изменений Также на ВМ (не в этом коммите): - lada_scraper/parse.py — пропуск sticky-тем при многопоточном парсинге
135 lines
5.8 KiB
Python
135 lines
5.8 KiB
Python
"""Запись данных: JSONL с ротацией по N тем на файл.
|
|
|
|
Формат одной записи (одна тема):
|
|
{
|
|
"section": str, # Название раздела
|
|
"section_id": str, # ID раздела
|
|
"topic_id": int, # ID темы на форуме
|
|
"topic_title": str, # Заголовок темы
|
|
"topic_url": str, # Прямая ссылка на тему
|
|
"tags": [str], # Список тегов (если есть)
|
|
"total_posts": int, # Количество постов в теме
|
|
"posts": [ # Список постов
|
|
{
|
|
"author": str, # Имя автора
|
|
"date": str, # Дата/время поста
|
|
"num": str, # Номер поста (#1, #2...)
|
|
"text": str, # Текст поста
|
|
}
|
|
],
|
|
"scraped_at": str, # TIMESTAMP в ISO-формате
|
|
}
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class TopicWriter:
|
|
"""Потокобезопасная запись тем в JSONL-файлы с ротацией.
|
|
|
|
Каждый воркер пишет в свою папку (worker_N/).
|
|
При достижении TOPICS_PER_FILE — создаётся новый part_*.jsonl.
|
|
|
|
Атрибуты:
|
|
_worker_dir: Путь к папке воркера.
|
|
_topics_file: Счётчик записанных тем в текущий файл.
|
|
_file_index: Индекс текущего part_*.jsonl.
|
|
_handle: Открытый файловый дескриптор.
|
|
"""
|
|
|
|
def __init__(self, worker_dir: Path, topics_per_file: int = 100):
|
|
"""Инициализация.
|
|
|
|
Args:
|
|
worker_dir: Папка воркера (worker_N/).
|
|
topics_per_file: Сколько тем писать в один файл перед ротацией.
|
|
"""
|
|
self._worker_dir = worker_dir
|
|
self._topics_per_file = topics_per_file
|
|
|
|
# Счётчики
|
|
self._topics_file = 0
|
|
self._file_index = 0
|
|
|
|
# Открываем первый файл
|
|
self._handle = self._open_next()
|
|
|
|
# ── Открытие/закрытие файла ─────────────────────────────────────────
|
|
|
|
def _open_next(self):
|
|
"""Закрыть текущий файл (если есть) и открыть следующий part_N.jsonl."""
|
|
# Закрываем предыдущий
|
|
if hasattr(self, "_handle") and self._handle and not self._handle.closed:
|
|
self._handle.close()
|
|
|
|
# Новый путь
|
|
path = self._worker_dir / f"part_{self._file_index:04d}.jsonl"
|
|
|
|
# Сбрасываем счётчик тем в файле
|
|
self._topics_file = 0
|
|
|
|
log.info(" 💾 %s", path.name)
|
|
fh = open(path, "w", encoding="utf-8")
|
|
self._file_index += 1
|
|
return fh
|
|
|
|
def close(self):
|
|
"""Закрыть текущий файл."""
|
|
if self._handle and not self._handle.closed:
|
|
self._handle.close()
|
|
|
|
# ── Запись ──────────────────────────────────────────────────────────
|
|
|
|
def write(self,
|
|
section_name: str,
|
|
section_slug: str,
|
|
topic: dict,
|
|
posts: list,
|
|
tags: list | None = None):
|
|
"""Записать одну тему в JSONL.
|
|
|
|
Если текущий файл заполнен — автоматически создаёт новый.
|
|
|
|
Args:
|
|
section_name: Название раздела (напр. "Бензиновые двигатели").
|
|
section_slug: Короткий ID раздела (напр. "benzinovye-dvigateli").
|
|
topic: Словарь темы: {id, title, url}.
|
|
posts: Список постов: [{author, date, num, text}].
|
|
tags: Список тегов (опционально).
|
|
"""
|
|
# ── Ротация ─────────────────────────────────────────────────────
|
|
if self._topics_file >= self._topics_per_file:
|
|
self._open_next()
|
|
|
|
# ── Формируем запись ────────────────────────────────────────────
|
|
record = {
|
|
"section": section_name,
|
|
"section_id": section_slug,
|
|
"topic_id": topic["id"],
|
|
"topic_title": topic["title"],
|
|
"topic_url": topic.get("url", ""),
|
|
"tags": tags or [],
|
|
"total_posts": len(posts),
|
|
"posts": posts,
|
|
"scraped_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
# ── Пишем в файл ────────────────────────────────────────────────
|
|
line = json.dumps(record, ensure_ascii=False)
|
|
self._handle.write(line + "\n")
|
|
self._handle.flush()
|
|
self._topics_file += 1
|
|
|
|
# ── Контекстный менеджер ────────────────────────────────────────────
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.close()
|