Dev2

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+max-line-length = 120
+extend-ignore = E203, W503

pyproject.toml

@@ -0,0 +1,3 @@
+[flake8]
+max-line-length = 120
+extend-ignore = ["E203", "W503"]

src/content_storage.py

@@ -62,3 +62,13 @@ def clear_pages() -> None:
     with closing(get_conn()) as conn:
         conn.execute("DELETE FROM pages;")
         conn.commit()
+
+
+def iter_all_urls() -> list[str]:
+    """
+    проходится по всем урлам
+    :return:
+    """
+    with closing(get_conn()) as conn:
+        rows = conn.execute("SELECT url FROM pages").fetchall()
+    return [r[0] for r in rows]

src/crawler.py

@@ -1,9 +1,51 @@
+import shutil
 from typing import List
 from urllib.parse import urljoin, urlparse
 from src.parser import get_links_from_url
 from src.url_storage import put_url_to_queue_if_not_met, get_next_url, r
-from src.content_storage import save_page
+from src.content_storage import save_page, iter_all_urls, clear_pages
+import src.mirror as mirror_mod
 import threading
+import os
+
+DENY_EXT = (
+    ".jpg",
+    ".jpeg",
+    ".png",
+    ".gif",
+    ".svg",
+    ".css",
+    ".js",
+    ".ico",
+    ".pdf",
+    ".zip",
+    ".rar",
+    ".7z",
+    ".mp3",
+    ".mp4",
+    ".avi",
+    ".mov",
+    ".webm",
+)
+
+
+def should_follow(url: str, allowed_domains: set[str] | None = None) -> bool:
+    """
+    решает, идти ли по ссылке
+    :param url: абсолютный URL
+    :param allowed_domains: домены, по которым разрешено ходить
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        return False
+    # фильтр по домену
+    if allowed_domains and parsed.netloc not in allowed_domains:
+        return False
+    # фильтр по расширениям
+    lower_path = parsed.path.lower()
+    if any(lower_path.endswith(ext) for ext in DENY_EXT):
+        return False
+    return True
 
 
 def normalize_url(base: str, link: str) -> str | None:
@@ -18,15 +60,17 @@ def normalize_url(base: str, link: str) -> str | None:
     parsed_url = urlparse(abs_url)
     if parsed_url.scheme not in ("http", "https"):
         return None
-    abs_url = abs_url.split("#", 1)[0]  # анкор
+    abs_url = abs_url.split("#", 1)[0]
     return abs_url
 
 
 processed_lock = threading.Lock()
 processed_pages = 0
 
 
-def worker(max_pages: int) -> None:
+def worker(
+    max_pages: int, mirror: bool = False, allowed_domains: set[str] | None = None
+) -> None:
     """
     Итеративно проходится по страницам в очереди, обрабатывая их.
     :param max_pages: максимальное кол-во страниц для краулинга (глубина)
@@ -53,16 +97,25 @@ def worker(max_pages: int) -> None:
         if status != 200:
             print(f"[{current_num}] got status {status}")
             continue
-
-        save_page(url, status, html)
+        if mirror:
+            mirror_mod.save_mirrored_page(url, html)
+        else:
+            save_page(url, status, html)
 
         for link in links:
+            clear_pages()
             new_url = normalize_url(url, link)
-            if new_url:
+            if new_url and should_follow(new_url, allowed_domains):
                 put_url_to_queue_if_not_met(new_url)
 
 
-def crawl(seed_url: str, max_pages: int, max_workers: int) -> None:
+def crawl(
+    seed_url: str,
+    max_pages: int,
+    max_workers: int,
+    update_mode: bool = False,
+    mirror_mode: bool = False,
+) -> None:
     """
     В несколько потоков краулит веб
     :param seed_url: стартовая веб-страница
@@ -72,11 +125,24 @@ def crawl(seed_url: str, max_pages: int, max_workers: int) -> None:
     global processed_pages
     processed_pages = 0
     r.flushdb()
-    put_url_to_queue_if_not_met(seed_url)
+    seed_host = urlparse(seed_url).netloc
+    allowed_domains = {seed_host}
+    if mirror_mode:
+        if os.path.isdir(mirror_mod.MIRROR_ROOT):
+            shutil.rmtree(mirror_mod.MIRROR_ROOT)
+    if update_mode:
+        for url in iter_all_urls():
+            put_url_to_queue_if_not_met(url, update=True)
+        if not iter_all_urls():
+            put_url_to_queue_if_not_met(seed_url, update=True)
+    else:
+        put_url_to_queue_if_not_met(seed_url)
     threads: List[threading.Thread] = []
 
     for i in range(max_workers):
-        thread = threading.Thread(target=worker, args=(max_pages,))
+        thread = threading.Thread(
+            target=worker, args=(max_pages, mirror_mode, allowed_domains)
+        )
         thread.start()
         threads.append(thread)
 

src/main.py

@@ -1,7 +1,6 @@
 from src.content_storage import init_db
 from src.crawler import crawl
 
-
 def main() -> None:
     """
     Сканирует интернет, переходя по ссылкам между страницами
@@ -12,7 +11,9 @@ def main() -> None:
     url = input("Введите seed-url: ")
     workers = int(input("Введите количество потоков: "))
     depth = int(input("Введите глубину парсинга: "))
-    crawl(url, depth, workers)
+    update_mode = input("Режим обновления? (y/n): ").strip().lower() == "y"
+    mirror_mode = input("Зеркало сайта? (y/n): ").strip().lower() == "y"
+    crawl(url, depth, workers, update_mode=update_mode, mirror_mode=mirror_mode)
 
 
 if __name__ == "__main__":

src/mirror.py

@@ -0,0 +1,100 @@
+import os
+from urllib.parse import urlparse, urljoin
+from bs4 import BeautifulSoup
+
+MIRROR_ROOT = "mirror"
+
+
+def url_to_local_path(url: str) -> str:
+    """
+    превращает url в путь к файлу внутри папки mirror
+    :param url: веб адрес
+    :return: путь к файлц
+    """
+    parsed = urlparse(url)
+    host = parsed.netloc
+    path = parsed.path
+    if not path or path.endswith("/"):
+        # https://example.com/ -> mirror/example.com/index.html
+        path = (path or "/") + "index.html"
+    else:
+        # https://example.com/a/b -> mirror/example.com/a/b/index.html
+        filename = os.path.basename(path)
+        if "." not in filename:
+            path = path.rstrip("/") + "/index.html"
+    full_path = os.path.join(MIRROR_ROOT, host, path.lstrip("/"))
+    os.makedirs(os.path.dirname(full_path), exist_ok=True)
+    return full_path
+
+
+def make_relative_path(from_path: str, to_path: str) -> str:
+    """
+
+    :param from_path:
+    :param to_path:
+    :return:
+    """
+    rel = os.path.relpath(to_path, start=os.path.dirname(from_path))
+    return rel.replace(os.sep, "/")
+
+
+def rewrite_links_to_local(base_url: str, html: str, current_local_path: str) -> str:
+    """
+    переписывает ссылки на локальные
+    :param base_url:
+    :param html:
+    :param current_local_path:
+    :return:
+    """
+    soup = BeautifulSoup(html, "html.parser")
+    base_parsed = urlparse(base_url)
+
+    def convert_url(href: str | None) -> str | None:
+        if not href:
+            return href
+        if href.startswith(("mailto:", "javascript:", "tel:", "#")):
+            return href
+
+        abs_url = urljoin(base_url, href)
+        parsed = urlparse(abs_url)
+
+        if parsed.scheme not in ("http", "https"):
+            return href
+        if parsed.netloc != base_parsed.netloc:
+            return href
+
+        target_local = url_to_local_path(abs_url)
+        rel = make_relative_path(current_local_path, target_local)
+        return rel
+
+    for a in soup.find_all("a", href=True):
+        new = convert_url(a["href"])
+        if new:
+            a["href"] = new
+    for img in soup.find_all("img", src=True):
+        new = convert_url(img["src"])
+        if new:
+            img["src"] = new
+    for link in soup.find_all("link", href=True):
+        new = convert_url(link["href"])
+        if new:
+            link["href"] = new
+    for script in soup.find_all("script", src=True):
+        new = convert_url(script["src"])
+        if new:
+            script["src"] = new
+
+    return str(soup)
+
+
+def save_mirrored_page(url: str, html: str) -> None:
+    """
+
+    :param url:
+    :param html:
+    :return:
+    """
+    local_path = url_to_local_path(url)
+    html_rewritten = rewrite_links_to_local(url, html, local_path)
+    with open(local_path, "w", encoding="utf-8") as f:
+        f.write(html_rewritten)

src/url_storage.py

@@ -17,12 +17,15 @@
 SEEN_KEY = "seen_urls"
 
 
-def put_url_to_queue_if_not_met(url: str) -> None:
+def put_url_to_queue_if_not_met(url: str, update: bool = False) -> None:
     """
     Добавляет url в очередь, если его там нет
     :param url: адрес веб страницы
     :return:
     """
+    if update:
+        r.rpush(FRONTIER_KEY, url)
+        return
     is_new = r.sadd(SEEN_KEY, url)
     if is_new:
         r.rpush(FRONTIER_KEY, url)

tests/test_crawler.py

@@ -79,7 +79,7 @@ def test_crawl_real() -> None:
 
 
 def test_crawl_on_fake_net(
-    monkeypatch: MonkeyPatch, fake_redis: Redis, temp_db: None
+        monkeypatch: MonkeyPatch, fake_redis: Redis, temp_db: None
 ) -> None:
     """
     Тест логики краулинга:

tests/test_filtration.py

@@ -0,0 +1,67 @@
+import src.crawler as crawler
+
+
+def test_should_follow_same_domain_html_page() -> None:
+    """
+    обычная HTML-страница на разрешённом домене — True
+    """
+    url = "https://example.com/path/page"
+    allowed = {"example.com"}
+    assert crawler.should_follow(url, allowed_domains=allowed) is True
+
+
+def test_should_follow_other_domain_false() -> None:
+    """
+    ссылка на другой домен должна отфильтровываться
+    """
+    url = "https://google.com/search?q=test"
+    allowed = {"example.com"}
+    assert crawler.should_follow(url, allowed_domains=allowed) is False
+
+
+def test_should_follow_denies_static_extensions() -> None:
+    """
+    ссылки на статические файлы должны блокироваться
+    """
+    allowed = {"example.com"}
+
+    urls = [
+        "https://example.com/image.jpg",
+        "https://example.com/image.JPEG",
+        "https://example.com/static/style.css",
+        "https://example.com/file.PDF",
+        "https://example.com/video.Mp4",
+    ]
+
+    for url in urls:
+        assert crawler.should_follow(url, allowed_domains=allowed) is False
+
+
+def test_should_follow_allows_when_no_allowed_domains() -> None:
+    """
+    если allowed_domains=None, фильтр по домену не применяется,
+    остаётся только фильтрация по схеме/расширениям
+    """
+    # домен любой, но расширения нет True
+    url1 = "https://some-other-site.org/page"
+    # домен любой, но это картинка False
+    url2 = "https://another.com/img/logo.png"
+
+    assert crawler.should_follow(url1, allowed_domains=None) is True
+    assert crawler.should_follow(url2, allowed_domains=None) is False
+
+
+def test_should_follow_rejects_non_http_schemes() -> None:
+    """
+    не HTTP схемы должны отбрасываться
+    """
+    allowed = {"example.com"}
+
+    urls = [
+        "mailto:test@example.com",
+        "ftp://example.com/file.txt",
+        "javascript:alert(1)",
+    ]
+
+    for url in urls:
+        assert crawler.should_follow(url, allowed_domains=allowed) is False