feat: single-line progress indicator during crawl and ext-link check

Uses \r to overwrite the same terminal line — no scrolling.
Crawler exposes on_progress callback; checker.check_external_links
gets an optional on_progress parameter.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-12 17:17:37 +02:00
commit 792ca67c08
3 changed files with 35 additions and 1 deletions

View file

@ -82,6 +82,23 @@ def _setup_logging(logs_dir: str, verbose: bool) -> None:
# ---------------------------------------------------------------------------
# Progress display helpers
# ---------------------------------------------------------------------------
def _progress(label: str, done: int, total: int | None = None) -> None:
"""Overwrite the current terminal line with a progress update."""
if total:
line = f" {label}: {done}/{total}"
else:
line = f" {label}: {done}"
print(f"\r{line:<72}", end="", flush=True)
def _progress_end() -> None:
"""Clear the progress line after a phase completes."""
print(f"\r{'':<72}\r", end="", flush=True)
# Crawl + extract helpers
# ---------------------------------------------------------------------------
@ -89,7 +106,11 @@ def _crawl_and_extract(cfg: dict) -> tuple[list[dict], list[dict]]:
"""Run crawler, extract each HTML page. Returns (pages, errors)."""
logger = logging.getLogger(__name__)
crawler = Crawler(cfg)
crawler.on_progress = lambda fetched, queued: _progress(
"Crawling", fetched, fetched + queued if queued else None
)
raw = crawler.crawl()
_progress_end()
pages: list[dict] = []
for raw_page in raw["pages"]:
@ -864,7 +885,13 @@ def _run_ext_links_core(cfg: dict, snap: dict) -> dict:
return {"total": 0, "ok": 0, "broken": 0, "errors": 0,
"broken_links": [], "error_links": []}
results = check_external_links(list(ext_urls.keys()), timeout=cfg.get("request_timeout", 10))
total_ext = len(ext_urls)
results = check_external_links(
list(ext_urls.keys()),
timeout=cfg.get("request_timeout", 10),
on_progress=lambda done, total: _progress("Externe Links prüfen", done, total),
)
_progress_end()
broken = {url: r for url, r in results.items() if r.get("status") and r["status"] >= 400}
conn_errors = {url: r for url, r in results.items() if r.get("error")}
return {

View file

@ -123,6 +123,7 @@ def check_external_links(
urls: list[str],
timeout: int = 10,
delay: float = 0.3,
on_progress: object = None,
) -> dict[str, dict]:
"""
Check a list of URLs for reachability via HEAD (fallback: GET).
@ -158,6 +159,8 @@ def check_external_links(
results[url] = {"status": resp.status_code, "final_url": resp.url, "error": None}
except requests.exceptions.RequestException as exc:
results[url] = {"status": None, "final_url": None, "error": str(exc)}
if on_progress:
on_progress(i + 1, len(urls))
finally:
session.close()

View file

@ -91,6 +91,7 @@ class Crawler:
self.headers = {"User-Agent": cfg["user_agent"]}
self.session = requests.Session()
self.session.headers.update(self.headers)
self.on_progress: object = None # callable(fetched: int, queued: int)
def crawl(self) -> dict:
"""
@ -176,6 +177,9 @@ class Crawler:
if link_url not in seen and link_url not in queue:
queue.append(link_url)
if self.on_progress:
self.on_progress(len(pages), len(queue))
except requests.exceptions.RequestException as exc:
errors.append({"url": url, "error": str(exc)})
logger.warning("Error crawling %s: %s", url, exc)