From 29e1ee26fbca98ae01036b455fa1e5ff82581e26 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Sun, 26 Apr 2026 19:50:23 +0300 Subject: [PATCH] =?UTF-8?q?=D0=92=D1=8B=D0=BD=D0=BE=D1=81=20=D0=BB=D0=BE?= =?UTF-8?q?=D0=B3=D0=B8=D0=BA=D0=B8=20=D0=BF=D1=80=D0=B5=D0=B4=D0=BC=D0=B5?= =?UTF-8?q?=D1=82=D0=BD=D0=BE=D0=B3=D0=BE=20=D1=83=D0=BA=D0=B0=D0=B7=D0=B0?= =?UTF-8?q?=D1=82=D0=B5=D0=BB=D1=8F=20=D0=B2=20=D0=BE=D1=82=D0=B4=D0=B5?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20=D0=BC=D0=BE=D0=B4=D1=83=D0=BB?= =?UTF-8?q?=D1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 29 +++++------- modules/crawler.py | 81 +++++++++++++++++++++++++++++++-- modules/index.py | 65 +++++++++++++++++++++++++++ modules/loader.py | 21 +++++++-- modules/processor.py | 104 ++++++++++++++++++------------------------- 5 files changed, 214 insertions(+), 86 deletions(-) create mode 100644 modules/index.py diff --git a/main.py b/main.py index ea87714..0d6677e 100644 --- a/main.py +++ b/main.py @@ -1,31 +1,26 @@ from modules.crawler import crawl_document -from modules.processor import ( - build_subject_index, - prepare_text, - write_subject_index_csv, - write_subject_index_json, -) +from modules.processor import prepare_text +from modules.index import build_subject_index, write_subject_index_csv, write_subject_index_json +from pathlib import Path SOURCE_URL = "https://www.consultant.ru/document/cons_doc_LAW_10699/" OUTPUT_DIR = "output" def main(): - from pathlib import Path - - output_dir = Path(OUTPUT_DIR) + output_dir = Path(OUTPUT_DIR) # Служебные вызовы для создания директории вывода output_dir.mkdir(exist_ok=True) - pages = crawl_document(SOURCE_URL) - original_text = "\n".join(page["text"] for page in pages) + pages = crawl_document(SOURCE_URL) # Вызов краулера на исходный url для рекурсивного обхода + original_text = "\n".join(page["text"] for page in pages) # Складываем результат работы краулера в единую строку - prepared_text = prepare_text(original_text) - subject_index = build_subject_index(original_text, top_n=100) + prepared_text = prepare_text(original_text) # Вызов препаратора на полученную строку для обработки + subject_index = build_subject_index(original_text, top_n=100) # Строим предметный указатель - (output_dir / "uk_rf_original.txt").write_text(original_text, encoding="utf-8") - (output_dir / "uk_rf_prepared.txt").write_text(prepared_text, encoding="utf-8") - write_subject_index_csv(subject_index, output_dir / "uk_rf_subject_index.csv") - write_subject_index_json(subject_index, output_dir / "uk_rf_subject_index.json") + (output_dir / "uk_rf_original.txt").write_text(original_text, encoding="utf-8") # Вывод исходного текста в файл + (output_dir / "uk_rf_prepared.txt").write_text(prepared_text, encoding="utf-8") # Вывод токенизированного и обработанного текста в файл + write_subject_index_csv(subject_index, output_dir / "uk_rf_subject_index.csv") # Вывод предметного указателя в csv + write_subject_index_json(subject_index, output_dir / "uk_rf_subject_index.json") # Предметные указатель в json print(f"Загружено статей: {len(pages)}") print(f"Строк исходного текста: {len(original_text.splitlines())}") diff --git a/modules/crawler.py b/modules/crawler.py index d2af68c..8d0b164 100644 --- a/modules/crawler.py +++ b/modules/crawler.py @@ -7,34 +7,52 @@ from lxml import html from modules.loader import download_html -START_URL = "https://www.consultant.ru/document/cons_doc_LAW_10699/" -FIRST_ARTICLE_URL = START_URL + "e8ecf933c52a85d9223094e0e7fbf52f0128d399/" +START_URL = "https://www.consultant.ru/document/cons_doc_LAW_10699/" # Базовый адрес УК РФ (откуда мы начинаем обход) +FIRST_ARTICLE_URL = START_URL + "e8ecf933c52a85d9223094e0e7fbf52f0128d399/" # Первая статья (для детерминированности обхода) -ARTICLE_RE = re.compile(r"^\s*(?:УК РФ,?\s+)?Статья\s+\d+(?:\.\d+)?\.", re.IGNORECASE) +ARTICLE_RE = re.compile(r"^\s*(?:УК РФ,?\s+)?Статья\s+\d+(?:\.\d+)?\.", re.IGNORECASE) STRUCTURE_PREFIX_RE = re.compile(r"^\s*(?:\d+(?:\.\d+)*|[а-яё])[\.)]\s+", re.IGNORECASE) SPACE_RE = re.compile(r"[ \t\r\f\v]+") def normalize_url(base_url: str, href: str) -> str: + """ + Преобразует относительный href в абсолютный URL и отрезает фрагмент (#...). + """ url = urljoin(base_url, href) url, _ = urldefrag(url) return url def document_prefix(url: str) -> str: + """ + Возвращает двухуровневый префикс пути URL — например /document/cons_doc_LAW_10699/. + + Используется для проверки, что ссылка ведёт в тот же документ. + """ parts = [part for part in urlparse(url).path.split("/") if part] return f"/{parts[0]}/{parts[1]}/" def is_same_document(url: str, prefix: str) -> bool: + """ + Проверяет, что URL принадлежит тому же документу на consultant.ru. + """ parsed = urlparse(url) return parsed.netloc in {"", "consultant.ru", "www.consultant.ru"} and parsed.path.startswith(prefix) def download_document_page(url: str, allow_partial: bool = False) -> str: + """ + Загружает страницу документа, повторяя запрос с возрастающими таймаутами (8, 30, 30 с). + + По умолчанию считает страницу полной только если HTML заканчивается тегом . + allow_partial=True отключает эту проверку (нужно для индексной страницы, которая может + грузиться медленно). Между попытками — пауза 0.5 с. + """ last_html = "" - for timeout in (8, 30, 30): + for timeout in (8, 15, 30): try: page_html = download_html(url, retries=1, timeout=timeout) except RuntimeError: @@ -54,6 +72,9 @@ def download_document_page(url: str, allow_partial: bool = False) -> str: def clean_text(text: str) -> str: + """ + Нормализует пробелы и срезает нумерационные префиксы вида «1. », «а) », «2.1 ». + """ text = text.replace("\xa0", " ") text = SPACE_RE.sub(" ", text) text = STRUCTURE_PREFIX_RE.sub("", text) @@ -61,11 +82,17 @@ def clean_text(text: str) -> str: def main_content(tree): + """ + Возвращает корневой div.document-page__content или None, если он не найден. + """ nodes = tree.xpath("//div[contains(concat(' ', normalize-space(@class), ' '), ' document-page__content ')]") return nodes[0] if nodes else None def article_title(tree) -> str: + """ + Извлекает заголовок статьи: сначала ищет

, fallback — первый div.doc-style. + """ content = main_content(tree) if content is None: return "" @@ -81,6 +108,12 @@ def article_title(tree) -> str: def service_line(line: str) -> bool: + """ + Возвращает True для редакционных пометок и реквизитов, которые не нужны в тексте. + + Фильтрует: пометки «(в ред. ФЗ ...)», «(введён ...)», «(см. текст ...)», + подпись президента, дату подписания и номер закона вида «N 63-ФЗ». + """ low = line.lower() if low.startswith("(см. текст") or low.startswith("(в ред.") or low.startswith("(введен"): @@ -100,6 +133,14 @@ def service_line(line: str) -> bool: def extract_page_text(tree) -> str: + """ + Извлекает смысловой текст статьи из HTML-дерева. + + Перед сбором абзацев удаляет из дерева заголовок (h1), блоки doc-style, + doc-insert и doc-roll (вставки, примечания, раскрывающиеся блоки), чтобы + их текст не попал в результат. Затем собирает все

, фильтрует служебные + строки через service_line() и возвращает текст через \n. + """ content = main_content(tree) if content is None: return "" @@ -122,6 +163,12 @@ def extract_page_text(tree) -> str: def extract_article_urls(tree, base_url: str) -> list[str]: + """ + Собирает ссылки на статьи со страницы — те, чей текст совпадает с ARTICLE_RE. + + Возвращает список уникальных URL в порядке появления на странице. + Внешние ссылки (не на тот же документ) отбрасываются. + """ prefix = document_prefix(base_url) result = [] @@ -135,6 +182,9 @@ def extract_article_urls(tree, base_url: str) -> list[str]: def extract_next_document_url(tree, base_url: str) -> str | None: + """ + Возвращает URL следующей страницы через кнопку-стрелку pages__right, или None. + """ prefix = document_prefix(base_url) links = tree.xpath("//a[contains(concat(' ', normalize-space(@class), ' '), ' pages__right ')][@href]") if not links: @@ -145,6 +195,12 @@ def extract_next_document_url(tree, base_url: str) -> str | None: def next_known_article(current_url: str, known_urls: list[str], visited: set[str]) -> str | None: + """ + Находит следующую непосещённую статью из known_urls после current_url. + + Используется как fallback, когда страница не содержит кнопки pages__right — + например, когда статья занимает ровно одну страницу и стрелки нет. + """ if current_url not in known_urls: return None @@ -157,12 +213,29 @@ def next_known_article(current_url: str, known_urls: list[str], visited: set[str def add_known_urls(known_urls: list[str], urls: list[str]) -> None: + """ + Добавляет новые URL в known_urls, не нарушая порядок и без дублей.""" for url in urls: if url not in known_urls: known_urls.append(url) def crawl_document(start_url: str = START_URL, max_pages: int = 800) -> list[dict]: + """ + Обходит документ УК РФ и возвращает список статей. + + Алгоритм: + 1. Загружает индексную страницу и собирает ссылки на все статьи (known_urls). + 2. Начинает с первой статьи и идёт по ним в порядке оглавления. + 3. На каждой странице: + - добавляет найденные ссылки на статьи в known_urls; + - если заголовок — статья, сохраняет текст; + - переходит на следующую страницу через pages__right или next_known_article(). + 4. Останавливается на «Статья 361.» (последняя статья УК), по лимиту max_pages + или если следующий URL недоступен. + + Возвращает list[dict] с ключами: url, title, text. + """ pages = [] visited = set() known_urls = [] diff --git a/modules/index.py b/modules/index.py new file mode 100644 index 0000000..6c449ed --- /dev/null +++ b/modules/index.py @@ -0,0 +1,65 @@ +import csv +import json +from collections import Counter +from pathlib import Path + +from modules.processor import good_words_from_line + + +def prepared_terms(text: str): + """Генератор: выдаёт (лемма, номер_строки, позиция_символа, исходное_слово) для каждого хорошего слова.""" + for line_number, line in enumerate(text.splitlines(), start=1): + for word in good_words_from_line(line): + yield word["lemma"], line_number, word["char"], word["source"] + + +def build_subject_index(text: str, top_n: int = 100) -> list[dict]: + """Строит предметный указатель: топ-N лемм по частоте с позицией первого вхождения. + + first_place.setdefault гарантирует, что запоминается именно первое вхождение — + prepared_terms обходит текст сверху вниз, поэтому первый же setdefault выигрывает. + """ + counts = Counter() + first_place = {} + + for lemma, line, char, source in prepared_terms(text): + counts[lemma] += 1 + first_place.setdefault(lemma, (line, char, source)) + + result = [] + for lemma, count in counts.most_common(top_n): + line, char, source = first_place[lemma] + result.append( + { + "word": lemma, + "count": count, + "line": line, + "char": char, + "source_word": source, + } + ) + + return result + + +def write_subject_index_csv(entries: list[dict], path: Path) -> None: + """Сохраняет предметный указатель в CSV с разделителем «;».""" + with path.open("w", encoding="utf-8", newline="") as file: + writer = csv.writer(file, delimiter=";") + writer.writerow(["word", "count", "line", "char", "source_word"]) + for entry in entries: + writer.writerow( + [ + entry["word"], + entry["count"], + entry["line"], + entry["char"], + entry["source_word"], + ] + ) + + +def write_subject_index_json(entries: list[dict], path: Path) -> None: + """Сохраняет предметный указатель в JSON с отступами и без ASCII-экранирования кириллицы.""" + with path.open("w", encoding="utf-8") as file: + json.dump(entries, file, ensure_ascii=False, indent=2) diff --git a/modules/loader.py b/modules/loader.py index fe417eb..2f73b39 100644 --- a/modules/loader.py +++ b/modules/loader.py @@ -1,7 +1,6 @@ import ssl import subprocess import time -from urllib.parse import urlparse from urllib.request import Request, urlopen @@ -13,6 +12,13 @@ HEADERS = { def download_with_curl(url: str, timeout: float) -> str: + """ + Загружает URL через системный curl с поддержкой gzip и редиректов. + + Флаги: -L следует редиректам, --compressed принимает gzip/br, + --silent подавляет прогресс, --show-error выводит ошибки в stderr. + Если stdout пуст (curl вернул ошибку), бросает RuntimeError со stderr. + """ command = [ "curl", "-L", @@ -35,6 +41,9 @@ def download_with_curl(url: str, timeout: float) -> str: def download_with_urllib(url: str, timeout: float) -> str: + """ + Загружает URL через стандартный urllib с отключённой проверкой SSL-сертификата (иначе может лечь). + """ context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE @@ -45,9 +54,13 @@ def download_with_urllib(url: str, timeout: float) -> str: def download_html(url: str, retries: int = 3, timeout: float = 30.0) -> str: - """Загружает HTML. Для ConsultantPlus сначала пробует curl, для остальных сайтов urllib.""" - is_consultant = urlparse(url).netloc.endswith("consultant.ru") - loaders = [download_with_curl, download_with_urllib] if is_consultant else [download_with_urllib, download_with_curl] + """Загружает HTML-страницу с retry-логикой. + + Порядок загрузчиков: curl → urllib (curl обходит часть защит consultant.ru). + На каждой итерации retry пробует оба; пауза 0.5 с между попытками. + Если все попытки исчерпаны — бросает RuntimeError с причиной последней ошибки. + """ + loaders = [download_with_curl, download_with_urllib] last_error = None for _ in range(retries): diff --git a/modules/processor.py b/modules/processor.py index 21561a8..12a91f7 100644 --- a/modules/processor.py +++ b/modules/processor.py @@ -1,15 +1,11 @@ -from collections import Counter from functools import lru_cache -import csv -import json -from pathlib import Path import re WORD_RE = re.compile(r"[А-Яа-яЁё]+(?:-[А-Яа-яЁё]+)?") DIGIT_BEFORE_WORD_RE = re.compile(r"\d[\d\s.,-]*$") -STOP_WORDS = set( +STOP_WORDS = set( # Стоп-слова (выведены вручную) """ а без более бы бывший был была были было быть в во весь вместе вне вновь все всего всей всем всеми всех вследствие вы где да для до его ее если есть же за из или им @@ -21,7 +17,7 @@ STOP_WORDS = set( """.split() ) -NUMERAL_WORDS = set( +NUMERAL_WORDS = set( # Числительные (выведены вручную) """ ноль один два три четыре пять шесть семь восемь девять десять одиннадцать двенадцать тринадцать четырнадцать пятнадцать шестнадцать семнадцать @@ -32,20 +28,20 @@ NUMERAL_WORDS = set( """.split() ) -NUMBER_UNITS = set( +NUMBER_UNITS = set( # Слова-спутники числительных (тоже малоинформативны) """ год месяц неделя день сутки час минута рубль копейка процент метр километр грамм килограмм литр """.split() ) -PROPER_WORDS = set( +PROPER_WORDS = set( # Имена собственные """ ельцин интернет конституция кремль москва россия рф снг ссср """.split() ) -PROPER_PHRASES = [ +PROPER_PHRASES = [ # Имена собственные в формате словосочетаний phrase.split() for phrase in [ "арбитражный процессуальный кодекс", @@ -82,11 +78,17 @@ try: except ImportError: MORPH = None else: - MORPH = pymorphy3.MorphAnalyzer() + MORPH = pymorphy3.MorphAnalyzer() # Почему не SpaCy? - потому что использовать NLP для простой лемматизации - это пушкой по воробьям. А PyMorphy - это просто большой словарь (и все равно меньше, чем модель), который анализирует русскую морфологию - по-умному - конечный автомат по словарю. @lru_cache(maxsize=100_000) def parse_word(word: str): + """Возвращает (лемма, тег) для слова. Кешируем, чтобы не прогонять одно и то же через модуль постоянно. + + «ё» → «е» для единообразия леммы в словаре и при сравнении. + Если pymorphy3 не установлен, тег пустой, лемма = lower(). + Берём первый (наиболее вероятный) разбор pymorphy3. + """ w = word.lower().replace("ё", "е") if MORPH is None: return w, "" @@ -101,22 +103,31 @@ def parse_word(word: str): @lru_cache(maxsize=10_000) def _tag_parts(tag: str) -> frozenset[str]: + """Разбивает строку тега pymorphy3 на множество граммем для быстрого поиска.""" return frozenset(re.split(r"[, ]+", tag)) def has_tag(tag: str, names: set[str]) -> bool: + """Проверяет, содержит ли тег хотя бы одну из граммем из names.""" return bool(_tag_parts(tag) & names) def is_numeral(lemma: str, tag: str) -> bool: + """True если слово — числительное: по словарю NUMERAL_WORDS или тегу NUMR/Anum.""" return lemma in NUMERAL_WORDS or has_tag(tag, {"NUMR", "Anum"}) def is_proper_name(lemma: str, tag: str) -> bool: + """True если слово — имя собственное: по словарю PROPER_WORDS или тегу Name/Surn/Patr/Geox/Orgn.""" return lemma in PROPER_WORDS or has_tag(tag, {"Name", "Surn", "Patr", "Geox", "Orgn"}) def find_phrase_positions(lemmas: list[str]) -> set[int]: + """Возвращает индексы лемм, входящих в устойчивые фразы-собственные из PROPER_PHRASES. + + Скользящим окном ищет каждую фразу в списке лемм и помечает все её позиции, + чтобы is_bad_word мог отфильтровать их целиком. + """ positions = set() for phrase in PROPER_PHRASES: @@ -129,6 +140,11 @@ def find_phrase_positions(lemmas: list[str]) -> set[int]: def words_from_line(line: str): + """Разбирает строку на слова и возвращает список dict с морфологическими данными. + + Каждый dict: source — оригинальное слово, lemma, tag, char — 1-based позиция + в строке, after_digit — стоит ли перед словом число (для фильтрации единиц измерения). + """ words = [] for match in WORD_RE.finditer(line): @@ -149,6 +165,13 @@ def words_from_line(line: str): def is_bad_word(word: dict, phrase_positions: set[int], index: int, after_number: bool) -> bool: + """Возвращает True если слово нужно отфильтровать. + + Причины отсева: длина ≤ 2, стоп-слово, часть устойчивой фразы-собственной, + имя собственное, числительное, единица измерения после числа. + after_number — признак того, что предыдущее значимое слово было числительным + (передаётся из good_words_from_line для цепочек вида «пять лет»). + """ lemma = word["lemma"] tag = word["tag"] @@ -165,6 +188,12 @@ def is_bad_word(word: dict, phrase_positions: set[int], index: int, after_number def good_words_from_line(line: str): + """Генератор: выдаёт только «хорошие» слова строки после всех фильтров. + + Сначала вычисляет позиции фраз-собственных, затем идёт по словам, + отслеживая флаг after_number — он сбрасывается на каждом не-числительном, + чтобы точно отловить единицы измерения, стоящие сразу после числа. + """ words = words_from_line(line) phrase_positions = find_phrase_positions([word["lemma"] for word in words]) after_number = False @@ -179,13 +208,11 @@ def good_words_from_line(line: str): yield word -def prepared_terms(text: str): - for line_number, line in enumerate(text.splitlines(), start=1): - for word in good_words_from_line(line): - yield word["lemma"], line_number, word["char"], word["source"] - - def prepare_text(text: str) -> str: + """Возвращает отфильтрованный текст: каждая строка — леммы через пробел. + + Пустые строки (все слова отсеяны) пропускаются. + """ lines = [] for line in text.splitlines(): @@ -194,48 +221,3 @@ def prepare_text(text: str) -> str: lines.append(" ".join(lemmas)) return "\n".join(lines) - - -def build_subject_index(text: str, top_n: int = 100) -> list[dict]: - counts = Counter() - first_place = {} - - for lemma, line, char, source in prepared_terms(text): - counts[lemma] += 1 - first_place.setdefault(lemma, (line, char, source)) - - result = [] - for lemma, count in counts.most_common(top_n): - line, char, source = first_place[lemma] - result.append( - { - "word": lemma, - "count": count, - "line": line, - "char": char, - "source_word": source, - } - ) - - return result - - -def write_subject_index_csv(entries: list[dict], path: Path) -> None: - with path.open("w", encoding="utf-8", newline="") as file: - writer = csv.writer(file, delimiter=";") - writer.writerow(["word", "count", "line", "char", "source_word"]) - for entry in entries: - writer.writerow( - [ - entry["word"], - entry["count"], - entry["line"], - entry["char"], - entry["source_word"], - ] - ) - - -def write_subject_index_json(entries: list[dict], path: Path) -> None: - with path.open("w", encoding="utf-8") as file: - json.dump(entries, file, ensure_ascii=False, indent=2)