242 lines
7.8 KiB
Python
242 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from collections import deque
|
|
from typing import Any
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from requests import Response
|
|
from requests.exceptions import RequestException
|
|
|
|
|
|
WEBSITE_URL = r"""<< website_url >>"""
|
|
MAX_PAGES_DEFAULT = int(r"""<< max_pages >>""")
|
|
REQUEST_TIMEOUT_SECONDS_DEFAULT = int(r"""<< request_timeout_seconds >>""")
|
|
VERIFY_TLS_DEFAULT = r"""<< verify_tls >>"""
|
|
FOLLOW_REDIRECTS_DEFAULT = r"""<< follow_redirects >>"""
|
|
REQUEST_HEADERS_JSON_DEFAULT = r"""<< request_headers_json >>"""
|
|
USER_AGENT_DEFAULT = r"""<< user_agent >>"""
|
|
|
|
|
|
def str_to_bool(value: Any) -> bool:
|
|
return str(value).strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
|
|
|
|
def parse_json_object(value: str, field_name: str) -> dict[str, Any]:
|
|
if not value.strip():
|
|
return {}
|
|
try:
|
|
parsed = json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise SystemExit(f"{field_name} must be valid JSON: {exc}") from exc
|
|
if not isinstance(parsed, dict):
|
|
raise SystemExit(f"{field_name} must be a JSON object.")
|
|
return parsed
|
|
|
|
|
|
def build_session(headers: dict[str, str], user_agent: str) -> requests.Session:
|
|
session = requests.Session()
|
|
session.headers.update({"User-Agent": user_agent})
|
|
session.headers.update(headers)
|
|
return session
|
|
|
|
|
|
def sanitize_url(url: str) -> str | None:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme in {"mailto", "javascript", "tel", "data"}:
|
|
return None
|
|
cleaned = parsed._replace(fragment="").geturl()
|
|
return cleaned or None
|
|
|
|
|
|
def same_host(url: str, website_host: str) -> bool:
|
|
host = (urlparse(url).hostname or "").lower()
|
|
return bool(host) and host == website_host
|
|
|
|
|
|
def is_html_response(response: Response) -> bool:
|
|
content_type = response.headers.get("Content-Type", "").lower()
|
|
return "text/html" in content_type or "application/xhtml+xml" in content_type
|
|
|
|
|
|
def probe_url(
|
|
session: requests.Session,
|
|
url: str,
|
|
timeout: int,
|
|
verify_tls: bool,
|
|
follow_redirects: bool,
|
|
) -> tuple[Response | None, str | None]:
|
|
try:
|
|
response = session.get(
|
|
url,
|
|
timeout=timeout,
|
|
verify=verify_tls,
|
|
allow_redirects=follow_redirects,
|
|
)
|
|
return response, None
|
|
except RequestException as exc:
|
|
return None, str(exc)
|
|
|
|
|
|
def main() -> int:
|
|
website_url = WEBSITE_URL.strip()
|
|
if not website_url:
|
|
raise SystemExit("website_url is required.")
|
|
website_host = (urlparse(website_url).hostname or "").lower()
|
|
if not website_host:
|
|
raise SystemExit("website_url must include a valid hostname.")
|
|
|
|
verify_tls = str_to_bool(VERIFY_TLS_DEFAULT)
|
|
follow_redirects = str_to_bool(FOLLOW_REDIRECTS_DEFAULT)
|
|
|
|
request_headers = parse_json_object(REQUEST_HEADERS_JSON_DEFAULT, "request_headers_json")
|
|
session = build_session(request_headers, USER_AGENT_DEFAULT.strip())
|
|
|
|
queue: deque[tuple[str, str]] = deque([(website_url, website_url)])
|
|
queued_urls = {website_url}
|
|
visited_pages: set[str] = set()
|
|
checked_resources: dict[str, dict[str, Any]] = {}
|
|
broken: dict[str, dict[str, Any]] = {}
|
|
|
|
pages_crawled = 0
|
|
|
|
while queue and pages_crawled < MAX_PAGES_DEFAULT:
|
|
page_url, source = queue.popleft()
|
|
if page_url in visited_pages:
|
|
continue
|
|
visited_pages.add(page_url)
|
|
|
|
response, error = probe_url(
|
|
session=session,
|
|
url=page_url,
|
|
timeout=REQUEST_TIMEOUT_SECONDS_DEFAULT,
|
|
verify_tls=verify_tls,
|
|
follow_redirects=follow_redirects,
|
|
)
|
|
pages_crawled += 1
|
|
|
|
if error:
|
|
broken.setdefault(
|
|
page_url,
|
|
{"url": page_url, "kind": "page", "status_code": None, "error": error, "sources": []},
|
|
)["sources"].append(source)
|
|
continue
|
|
|
|
if response is None:
|
|
continue
|
|
|
|
if response.status_code >= 400:
|
|
broken.setdefault(
|
|
page_url,
|
|
{
|
|
"url": page_url,
|
|
"kind": "page",
|
|
"status_code": response.status_code,
|
|
"error": f"HTTP {response.status_code}",
|
|
"sources": [],
|
|
},
|
|
)["sources"].append(source)
|
|
continue
|
|
|
|
if not is_html_response(response):
|
|
continue
|
|
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
|
|
discovered: list[tuple[str, str]] = []
|
|
for tag_name, attribute, kind in [
|
|
("a", "href", "link"),
|
|
("img", "src", "asset"),
|
|
("script", "src", "asset"),
|
|
("link", "href", "asset"),
|
|
]:
|
|
for tag in soup.find_all(tag_name):
|
|
raw_value = tag.get(attribute)
|
|
if not raw_value:
|
|
continue
|
|
normalized = sanitize_url(urljoin(page_url, raw_value))
|
|
if not normalized:
|
|
continue
|
|
discovered.append((normalized, kind))
|
|
|
|
for discovered_url, kind in discovered:
|
|
if not same_host(discovered_url, website_host):
|
|
continue
|
|
|
|
if kind == "link" and discovered_url not in visited_pages and discovered_url not in queued_urls:
|
|
queue.append((discovered_url, page_url))
|
|
queued_urls.add(discovered_url)
|
|
|
|
if discovered_url in checked_resources:
|
|
checked_resources[discovered_url]["sources"].add(page_url)
|
|
if discovered_url in broken:
|
|
broken[discovered_url]["sources"].append(page_url)
|
|
continue
|
|
|
|
resource_response, resource_error = probe_url(
|
|
session=session,
|
|
url=discovered_url,
|
|
timeout=REQUEST_TIMEOUT_SECONDS_DEFAULT,
|
|
verify_tls=verify_tls,
|
|
follow_redirects=follow_redirects,
|
|
)
|
|
|
|
checked_resources[discovered_url] = {
|
|
"url": discovered_url,
|
|
"kind": kind,
|
|
"sources": {page_url},
|
|
}
|
|
|
|
if resource_error:
|
|
broken[discovered_url] = {
|
|
"url": discovered_url,
|
|
"kind": kind,
|
|
"status_code": None,
|
|
"error": resource_error,
|
|
"sources": [page_url],
|
|
}
|
|
continue
|
|
|
|
if resource_response is not None and resource_response.status_code >= 400:
|
|
broken[discovered_url] = {
|
|
"url": discovered_url,
|
|
"kind": kind,
|
|
"status_code": resource_response.status_code,
|
|
"error": f"HTTP {resource_response.status_code}",
|
|
"sources": [page_url],
|
|
}
|
|
|
|
for value in broken.values():
|
|
value["sources"] = sorted(set(value["sources"]))
|
|
|
|
summary = {
|
|
"pages_crawled": pages_crawled,
|
|
"urls_checked": len(set(checked_resources) | visited_pages),
|
|
"broken_urls": len(broken),
|
|
}
|
|
overall_status = "fail" if broken else "pass"
|
|
sorted_broken = sorted(broken.values(), key=lambda item: item["url"])
|
|
|
|
print(f"Broken Link Crawler: {overall_status.upper()}")
|
|
print(f"Website: {website_url}")
|
|
print(f"Pages crawled: {pages_crawled}")
|
|
print(f"URLs checked: {summary['urls_checked']}")
|
|
print(f"Broken URLs: {summary['broken_urls']}")
|
|
if sorted_broken:
|
|
print("")
|
|
print("Broken targets:")
|
|
for item in sorted_broken:
|
|
source = ", ".join(item["sources"][:3]) or "-"
|
|
print(f"- [{item['kind']}] {item['url']} | source={source} | error={item['error']}")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|