#!/usr/bin/env python3 """Обход сайта по плану маршруты/план-обхода.json. Ограничения берутся из плана: только allowed_domains, глубина не больше max_depth, не больше max_pages страниц. PDF-ссылки, попавшие под условие select (format=pdf, name_contains), скачиваются в output_directory. Реестр URL, статуса, типа содержимого и пути файла пишется в registry. """ from __future__ import annotations import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from collections import deque from datetime import datetime, timezone from html.parser import HTMLParser PLAN_PATH = "маршруты/план-обхода.json" USER_AGENT = "dcd-rag-teaching-crawler/1.0" TIMEOUT = 30 PAUSE_SECONDS = 0.5 class LinkParser(HTMLParser): """Собирает значения href со страницы.""" def __init__(self) -> None: super().__init__(convert_charrefs=True) self.links: list[str] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag.lower() != "a": return for name, value in attrs: if name.lower() == "href" and value: self.links.append(value.strip()) def normalize(url: str, base: str) -> str | None: """Приводит ссылку к абсолютному виду, отбрасывает якоря и не-HTTP схемы.""" if not url or url.startswith(("#", "mailto:", "tel:", "javascript:", "data:")): return None absolute = urllib.parse.urljoin(base, url) parts = urllib.parse.urlsplit(absolute) if parts.scheme not in ("http", "https"): return None path = parts.path or "/" return urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, parts.query, "")) def fetch(url: str) -> tuple[int, str, bytes, str]: """Загружает ресурс: статус, тип содержимого, тело и итоговый URL.""" request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(request, timeout=TIMEOUT) as response: return (response.status, response.headers.get("Content-Type", ""), response.read(), response.geturl()) except urllib.error.HTTPError as error: body = b"" try: body = error.read() except Exception: # noqa: BLE001 - тело ошибки необязательно pass return (error.code, error.headers.get("Content-Type", "") if error.headers else "", body, url) def is_pdf_link(url: str) -> bool: return urllib.parse.urlsplit(url).path.lower().endswith(".pdf") def file_name(url: str) -> str: return os.path.basename(urllib.parse.urlsplit(url).path) or "index" def main() -> int: with open(PLAN_PATH, encoding="utf-8") as handle: plan = json.load(handle) allowed_domains = {d.lower() for d in plan["allowed_domains"]} max_depth = int(plan["max_depth"]) max_pages = int(plan["max_pages"]) select = plan.get("select", {}) select_format = str(select.get("format", "")).lower() name_contains = str(select.get("name_contains", "")).lower() output_directory = plan["output_directory"] registry_path = plan["registry"] started_at = datetime.now(timezone.utc).isoformat(timespec="seconds") os.makedirs(output_directory, exist_ok=True) records: list[dict] = [] seen_pages: set[str] = set() seen_downloads: set[str] = set() queued: set[str] = set() pages_visited = 0 queue: deque[tuple[str, int, str | None]] = deque() for start in plan["start_urls"]: normalized = normalize(start, start) if normalized and normalized not in queued: queued.add(normalized) queue.append((normalized, 0, None)) while queue and pages_visited < max_pages: url, depth, source = queue.popleft() if url in seen_pages: continue seen_pages.add(url) status, content_type, body, final_url = fetch(url) pages_visited += 1 record = { "kind": "page", "url": url, "final_url": final_url, "status": status, "content_type": content_type, "file_path": None, "depth": depth, "source_url": source, "note": "", } records.append(record) print(f"[страница d={depth}] {status} {content_type or '-'} {url}") if status != 200 or "html" not in content_type.lower(): record["note"] = "страница не разобрана: нет HTML-содержимого" time.sleep(PAUSE_SECONDS) continue parser = LinkParser() parser.feed(body.decode("utf-8", "replace")) found_on_page = 0 for raw in parser.links: link = normalize(raw, url) if not link: continue host = urllib.parse.urlsplit(link).netloc.lower() if host not in allowed_domains: continue found_on_page += 1 if is_pdf_link(link) and name_contains in file_name(link).lower(): # Кандидат под условие отбора: скачиваем независимо от глубины ссылки. if link in seen_downloads: continue seen_downloads.add(link) want_pdf = select_format == "pdf" dl_status, dl_type, dl_body, _ = fetch(link) is_pdf = dl_type.lower().startswith("application/pdf") or dl_body[:5] == b"%PDF-" target = os.path.join(output_directory, file_name(link)) if dl_status == 200 and is_pdf and want_pdf: with open(target, "wb") as out: out.write(dl_body) records.append({ "kind": "download", "url": link, "status": dl_status, "content_type": dl_type, "file_path": target, "depth": depth, "source_url": url, "note": "скачано по условию отбора", }) print(f" [файл] {dl_status} {dl_type} -> {target}") else: records.append({ "kind": "download", "url": link, "status": dl_status, "content_type": dl_type, "file_path": None, "depth": depth, "source_url": url, "note": "отклонено: содержимое не является PDF", }) print(f" [файл отклонён] {dl_status} {dl_type or '-'} {link}") time.sleep(PAUSE_SECONDS) continue if depth + 1 <= max_depth and link not in queued: queued.add(link) queue.append((link, depth + 1, url)) record["note"] = f"найдено внутренних ссылок: {found_on_page}" time.sleep(PAUSE_SECONDS) summary = { "pages_limit": max_pages, "pages_visited": sum(1 for r in records if r["kind"] == "page"), "pages_pending_in_queue": len(queue), "downloads": sum(1 for r in records if r["kind"] == "download" and r["file_path"]), "downloads_rejected": sum(1 for r in records if r["kind"] == "download" and not r["file_path"]), } registry = { "plan": plan, "script": "сборка/обход-маршрута.py", "started_at": started_at, "finished_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "summary": summary, "records": records, } with open(registry_path, "w", encoding="utf-8") as out: json.dump(registry, out, ensure_ascii=False, indent=2) out.write("\n") print("\nИтог:", json.dumps(summary, ensure_ascii=False)) print("Реестр:", registry_path) return 0 if __name__ == "__main__": sys.exit(main())