站内链接审计实战：用 Python 找出孤立文章、断链与过深页面

from __future__ import annotations

import argparse
import csv
import time
from collections import Counter, deque
from dataclasses import dataclass, field
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import urldefrag, urljoin, urlparse, urlunparse

import requests
from urllib import robotparser


SKIP_PREFIXES = ("/admin", "/login", "/logout", "/register", "/search")
SKIP_SUFFIXES = (".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".css", ".js", ".pdf", ".zip")


@dataclass
class Page:
    url: str
    depth: int
    status: int = 0
    content_type: str = ""
    title: str = ""
    canonical: str = ""
    links: list[str] = field(default_factory=list)
    error: str = ""


class LinkParser(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.links: list[str] = []
        self.title = ""
        self.in_title = False
        self.canonical = ""

    def handle_starttag(self, tag, attrs):
        attributes = dict(attrs)
        tag = tag.lower()
        if tag == "a" and attributes.get("href"):
            self.links.append(attributes["href"])
        elif tag == "title":
            self.in_title = True
        elif tag == "link":
            rel = {item.strip().lower() for item in attributes.get("rel", "").split()}
            if "canonical" in rel:
                self.canonical = attributes.get("href", "")

    def handle_endtag(self, tag):
        if tag.lower() == "title":
            self.in_title = False

    def handle_data(self, data):
        if self.in_title:
            self.title += data.strip()


def normalize(url: str) -> str:
    url, _ = urldefrag(url)
    parsed = urlparse(url)
    path = parsed.path or "/"
    if path != "/" and path.endswith("/") is False and "." not in path.rsplit("/", 1)[-1]:
        path += "/"
    return urlunparse((parsed.scheme.lower(), parsed.netloc.lower(), path, "", "", ""))


def is_allowed(url: str, hosts: set[str]) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"} or parsed.netloc not in hosts:
        return False
    if parsed.query or parsed.fragment:
        return False
    path = parsed.path.lower()
    if any(path.startswith(prefix) for prefix in SKIP_PREFIXES):
        return False
    return not path.endswith(SKIP_SUFFIXES)


def make_robot_parser(base_url: str) -> robotparser.RobotFileParser:
    robots_url = urljoin(base_url, "/robots.txt")
    parser = robotparser.RobotFileParser(robots_url)
    try:
        parser.read()
    except OSError:
        # A missing robots file should be reported by the caller, not crash the crawl.
        parser.parse([])
    return parser


def crawl(base_url: str, start_paths: list[str], max_pages: int, delay: float) -> dict[str, Page]:
    base_url = normalize(base_url)
    hosts = {urlparse(base_url).netloc}
    robots = make_robot_parser(base_url)
    session = requests.Session()
    session.headers["User-Agent"] = "BlogLinkAudit/1.0 (+manual-audit)"
    queue = deque((normalize(urljoin(base_url, path)), 0) for path in start_paths)
    pages: dict[str, Page] = {}

    while queue and len(pages) < max_pages:
        url, depth = queue.popleft()
        if url in pages or not is_allowed(url, hosts):
            continue
        page = Page(url=url, depth=depth)
        pages[url] = page
        if not robots.can_fetch(session.headers["User-Agent"], url):
            page.error = "robots.txt disallows this URL"
            continue

        try:
            response = session.get(url, timeout=(3.05, 10), allow_redirects=True)
            page.status = response.status_code
            page.content_type = response.headers.get("content-type", "")
            final_url = normalize(response.url)
            if final_url != url:
                page.error = f"redirected to {final_url}"
            if response.status_code >= 400:
                page.error = page.error or f"HTTP {response.status_code}"
            elif "text/html" in page.content_type.lower():
                if len(response.content) > 2_000_000:
                    page.error = "HTML response exceeds 2 MB limit"
                else:
                    parser = LinkParser()
                    parser.feed(response.text)
                    page.title = " ".join(parser.title.split())
                    page.canonical = normalize(urljoin(response.url, parser.canonical)) if parser.canonical else ""
                    for href in parser.links:
                        target_raw = urljoin(response.url, href)
                        if is_allowed(target_raw, hosts):
                            target = normalize(target_raw)
                            page.links.append(target)
                            if target not in pages:
                                queue.append((target, depth + 1))
        except requests.RequestException as exc:
            page.error = str(exc)
        if delay:
            time.sleep(delay)
    return pages


def build_report(pages: dict[str, Page]) -> list[dict[str, object]]:
    incoming = Counter(link for page in pages.values() for link in page.links)
    rows = []
    for url, page in pages.items():
        rows.append({
            "url": url,
            "status": page.status,
            "depth": page.depth,
            "incoming_links": incoming[url],
            "outgoing_links": len(set(page.links)),
            "title": page.title,
            "canonical": page.canonical,
            "error": page.error,
        })
    return sorted(rows, key=lambda row: (int(row["depth"]), str(row["url"])))


def write_csv(rows: list[dict[str, object]], output: Path) -> None:
    fieldnames = ["url", "status", "depth", "incoming_links", "outgoing_links", "title", "canonical", "error"]
    with output.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main() -> int:
    parser = argparse.ArgumentParser(description="审计同源网站的站内链接结构")
    parser.add_argument("base_url", help="站点根地址，例如 https://blog.zenleak.cn")
    parser.add_argument("--start", action="append", default=["/"], help="抓取入口路径，可重复指定")
    parser.add_argument("--max-pages", type=int, default=300)
    parser.add_argument("--delay", type=float, default=0.3, help="请求间隔秒数")
    parser.add_argument("--output", default="link-audit.csv")
    args = parser.parse_args()
    if args.max_pages < 1 or args.delay < 0:
        parser.error("--max-pages 必须大于 0，--delay 不能为负数")

    pages = crawl(args.base_url, args.start, args.max_pages, args.delay)
    rows = build_report(pages)
    write_csv(rows, Path(args.output))
    isolated = [row["url"] for row in rows if row["incoming_links"] == 0 and row["url"].rstrip("/") != args.base_url.rstrip("/")]
    broken = [row for row in rows if int(row["status"]) >= 400 or row["error"]]
    deep = [row for row in rows if int(row["depth"]) >= 4]
    print(f"抓取 {len(rows)} 个页面，孤立页面 {len(isolated)} 个，异常链接 {len(broken)} 个，深度 >= 4 的页面 {len(deep)} 个")
    print(f"CSV 报告：{args.output}")
    for label, values in (("孤立页面", isolated[:20]), ("异常链接", [row["url"] for row in broken[:20]]), ("过深页面", [row["url"] for row in deep[:20]])):
        print(f"\n{label}：")
        for value in values:
            print(f"- {value}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python link_audit.py \
  https://blog.zenleak.cn \
  --start / --start /topics/seo/ --start /posts/ \
  --max-pages 350 --delay 0.4 --output link-audit.csv