Итоговый практический экзамен

__version__ = "0.4.2"

"""Настройки. Читаются из файла рядом с модулем."""
from __future__ import annotations

import json
from pathlib import Path

DEFAULTS = {
    "host": "127.0.0.1",
    "port": 8080,
    "workers": 4,
    "timeout_seconds": 10,
    "cache_dir": "/var/cache/linkcheck",
    "report_dir": "/var/lib/linkcheck/reports",
    "warmup_seconds": 12,
}

CONFIG_PATH = Path(__file__).with_name("settings.json")


def load() -> dict:
    settings = dict(DEFAULTS)
    if CONFIG_PATH.exists():
        settings.update(json.loads(CONFIG_PATH.read_text(encoding="utf-8")))
    return settings

{
  "port": 8080,
  "workers": 4
}

"""Хранение результатов проверок на диске."""
from __future__ import annotations

import json
import time
from pathlib import Path


class ReportStore:
    def __init__(self, report_dir: str) -> None:
        self.dir = Path(report_dir)
        self.dir.mkdir(parents=True, exist_ok=True)

    def save(self, job_id: str, payload: dict) -> Path:
        path = self.dir / f"{job_id}.json"
        path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
        return path

    def load(self, job_id: str) -> dict | None:
        path = self.dir / f"{job_id}.json"
        if not path.exists():
            return None
        return json.loads(path.read_text(encoding="utf-8"))

    def list_ids(self) -> list[str]:
        return sorted(p.stem for p in self.dir.glob("*.json"))

    def prune(self, older_than_seconds: int) -> int:
        now = time.time()
        removed = 0
        for p in self.dir.glob("*.json"):
            if now - p.stat().st_mtime > older_than_seconds:
                p.unlink()
                removed += 1
        return removed

"""Проверка доступности ссылок.

Работает в пуле потоков: сетевое ожидание, а не вычисления.
"""
from __future__ import annotations

import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor


def check_one(url: str, timeout: int) -> dict:
    started = time.perf_counter()
    try:
        with urllib.request.urlopen(url, timeout=timeout) as response:
            status = response.status
            error = None
    except urllib.error.HTTPError as exc:
        status, error = exc.code, None
    except Exception as exc:
        status, error = None, f"{type(exc).__name__}: {exc}"
    return {
        "url": url,
        "status": status,
        "error": error,
        "ms": round((time.perf_counter() - started) * 1000, 1),
    }


def check_all(urls: list[str], workers: int, timeout: int) -> list[dict]:
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(lambda u: check_one(u, timeout), urls))

"""HTTP-интерфейс на стандартной библиотеке."""
from __future__ import annotations

import json
import logging
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse

from . import __version__
from .checker import check_all
from .config import load
from .store import ReportStore

log = logging.getLogger("linkcheck")

SETTINGS = load()
STORE = ReportStore(SETTINGS["report_dir"])
STARTED_AT = time.time()
JOBS: dict[str, str] = {}


def warmed_up() -> bool:
    """Прогрев: первые секунды сервис отвечает, но работает медленно."""
    return time.time() - STARTED_AT >= SETTINGS["warmup_seconds"]


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def _send(self, code: int, payload: dict) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt: str, *args) -> None:
        log.info(fmt % args)

    def do_GET(self) -> None:
        path = urlparse(self.path).path
        if path == "/health":
            self._send(200, {"status": "ok", "version": __version__,
                             "warmed_up": warmed_up()})
        elif path == "/jobs":
            self._send(200, {"jobs": STORE.list_ids()})
        elif path.startswith("/jobs/"):
            job_id = path.split("/")[-1]
            report = STORE.load(job_id)
            if report is None:
                self._send(404, {"error": "не найдено", "job_id": job_id})
            else:
                self._send(200, report)
        else:
            self._send(404, {"error": "неизвестный путь", "path": path})

    def do_POST(self) -> None:
        if urlparse(self.path).path != "/check":
            self._send(404, {"error": "неизвестный путь"})
            return
        length = int(self.headers.get("Content-Length", 0))
        try:
            payload = json.loads(self.rfile.read(length))
            urls = payload["urls"]
        except Exception as exc:
            self._send(400, {"error": f"неверное тело запроса: {exc}"})
            return

        job_id = uuid.uuid4().hex[:12]
        JOBS[job_id] = "running"

        def work() -> None:
            results = check_all(urls, SETTINGS["workers"], SETTINGS["timeout_seconds"])
            STORE.save(job_id, {"job_id": job_id, "results": results,
                                "finished_at": time.time()})
            JOBS[job_id] = "done"

        threading.Thread(target=work, daemon=True).start()
        self._send(202, {"job_id": job_id, "urls": len(urls)})


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    server = ThreadingHTTPServer((SETTINGS["host"], SETTINGS["port"]), Handler)
    log.info("linkcheck %s слушает %s:%s", __version__,
             SETTINGS["host"], SETTINGS["port"])
    server.serve_forever()


if __name__ == "__main__":
    main()

from .server import main

main()

python3 -m linkcheck &
sleep 1
curl -s localhost:8080/health
curl -s -X POST localhost:8080/check -H 'content-type: application/json' \
     -d '{"urls": ["https://example.com"]}'

git clone . /tmp/проверка && cd /tmp/проверка
docker compose build
docker compose up -d
sleep 5

curl -s localhost:8080/health                              # 1, 2
docker compose exec -T linkcheck sh -c 'exit 0' || true    # 4: оболочки может не быть
docker image inspect ЛОКАЛЬНЫЙ_ОБРАЗ --format '{{.Config.User}}'  # 4
time docker compose stop                                   # 6
docker compose up -d && sleep 5
curl -s localhost:8080/jobs                                # 7: отчёты на месте
docker compose logs --no-log-prefix | head -3 | \
  python3 -c 'import json,sys; [json.loads(l) for l in sys.stdin]; print("логи разбираются")'  # 9
docker build --target test .                               # 12