status-django

  1"""HTTP fetching, robots.txt, and sitemap loading."""
  2import logging
  3import urllib.robotparser
  4from dataclasses import dataclass, field
  5from urllib.parse import urlparse
  6
  7import requests
  8from bs4 import BeautifulSoup
  9
 10
 11logger = logging.getLogger(__name__)
 12
 13USER_AGENT = "status (+https://status.bythewood.me)"
 14PAGE_CAP = 500
 15CONCURRENCY = 4
 16REQUEST_TIMEOUT = (5, 15)
 17EXTERNAL_LINK_TIMEOUT = (3, 8)
 18# Hard deadline for a single site crawl. Scheduler JOIN_TIMEOUT must exceed this.
 19CRAWL_DEADLINE_SECONDS = 540
 20
 21
 22@dataclass
 23class FetchResult:
 24    url: str
 25    requested_url: str
 26    status: int
 27    headers: dict
 28    body: bytes
 29    content_type: str
 30    elapsed_ms: int
 31    redirect_chain: list = field(default_factory=list)
 32    error: str = ""
 33
 34
 35def make_session():
 36    s = requests.Session()
 37    s.headers.update({"User-Agent": USER_AGENT})
 38    return s
 39
 40
 41def fetch(session, url):
 42    """GET a URL and return a FetchResult. Body is only captured for HTML."""
 43    try:
 44        r = session.get(url, timeout=REQUEST_TIMEOUT, allow_redirects=True)
 45        chain = [(h.status_code, h.url) for h in r.history]
 46        chain.append((r.status_code, r.url))
 47        content_type = r.headers.get("content-type", "").lower()
 48        body = r.content if "text/html" in content_type else b""
 49        return FetchResult(
 50            url=r.url,
 51            requested_url=url,
 52            status=r.status_code,
 53            headers=dict(r.headers),
 54            body=body,
 55            content_type=content_type,
 56            elapsed_ms=int(r.elapsed.total_seconds() * 1000),
 57            redirect_chain=chain,
 58        )
 59    except requests.RequestException as e:
 60        return FetchResult(
 61            url=url,
 62            requested_url=url,
 63            status=0,
 64            headers={},
 65            body=b"",
 66            content_type="",
 67            elapsed_ms=0,
 68            error=str(e),
 69        )
 70
 71
 72def head_status(session, url):
 73    """Cheap check for external links. Returns the status code (0 on error)."""
 74    try:
 75        r = session.head(url, timeout=EXTERNAL_LINK_TIMEOUT, allow_redirects=True)
 76        # Some servers reject HEAD with 405/403 but accept GET.
 77        if r.status_code in (403, 405, 501):
 78            r = session.get(
 79                url, timeout=EXTERNAL_LINK_TIMEOUT, allow_redirects=True, stream=True
 80            )
 81            r.close()
 82        return r.status_code
 83    except requests.RequestException:
 84        return 0
 85
 86
 87def load_robots(session, base_origin):
 88    """Fetch robots.txt. Returns (RobotFileParser, robots_url, raw_text_or_None)."""
 89    robots_url = f"{base_origin}/robots.txt"
 90    rp = urllib.robotparser.RobotFileParser()
 91    text = None
 92    try:
 93        r = session.get(robots_url, timeout=REQUEST_TIMEOUT)
 94        if r.status_code == 200:
 95            text = r.text
 96            rp.parse(text.splitlines())
 97    except requests.RequestException:
 98        pass
 99    return rp, robots_url, text
100
101
102def _parse_sitemap_xml(body):
103    """Return all <loc> values. Distinguishing index vs urlset is handled upstream."""
104    soup = BeautifulSoup(body, "xml")
105    return [loc.get_text(strip=True) for loc in soup.find_all("loc")]
106
107
108def load_sitemap(session, base_origin, robots_text):
109    """Return list of URLs from sitemap(s). Follows sitemap indexes one level.
110
111    Checks Sitemap: entries in robots.txt first, falls back to /sitemap.xml.
112    """
113    candidates = []
114    if robots_text:
115        for line in robots_text.splitlines():
116            if line.lower().startswith("sitemap:"):
117                candidates.append(line.split(":", 1)[1].strip())
118    if not candidates:
119        candidates.append(f"{base_origin}/sitemap.xml")
120
121    seen = set()
122    urls = []
123    to_fetch = list(candidates)
124    while to_fetch and len(seen) < 20:
125        smurl = to_fetch.pop()
126        if smurl in seen:
127            continue
128        seen.add(smurl)
129        try:
130            r = session.get(smurl, timeout=REQUEST_TIMEOUT)
131            if r.status_code != 200:
132                continue
133            for loc in _parse_sitemap_xml(r.content):
134                # Sub-sitemaps end in .xml; everything else is a page URL.
135                if loc.lower().endswith(".xml") or "sitemap" in loc.lower():
136                    to_fetch.append(loc)
137                else:
138                    urls.append(loc)
139        except requests.RequestException:
140            continue
141    return urls
142
143
144def same_site(url, host):
145    """True if `url` is on `host` or its www./apex counterpart."""
146    u = urlparse(url).netloc.lower()
147    h = host.lower()
148    if not u:
149        return False
150    if u == h:
151        return True
152    if u == "www." + h or h == "www." + u:
153        return True
154    return False