data/script/crawl_tangshi300.py

#!/usr/bin/env python3
"""Crawl 唐诗三百首 from guwendao.net.

Outputs are written to the chosen data directory:
- tangshi300.json
- tangshi300.jsonl
- tangshi300.csv
- tangshi300_summary.txt

Only Python standard library is used.
"""

from __future__ import annotations

import argparse
import csv
import html as html_lib
import ipaddress
import json
import re
import socket
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin, urlparse
from urllib.request import Request, urlopen

BASE_URL = "https://www.guwendao.net"
START_URL = "https://www.guwendao.net/gushi/tangshi.aspx"
ALLOWED_HOSTS = {"www.guwendao.net", "guwendao.net"}
DEFAULT_USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/126.0 Safari/537.36 TangshiCrawler/1.0"
)

SCRIPT_STYLE_RE = re.compile(r"<(script|style)\b[^>]*>.*?</\1>", re.I | re.S)
TAG_RE = re.compile(r"<[^>]+>", re.S)
DIV_TAG_RE = re.compile(r"<(/?)div\b([^>]*)>", re.I | re.S)
ATTR_RE_TEMPLATE = r"\b{}\s*=\s*(['\"])(.*?)\1"


def log(message: str) -> None:
    print(message, flush=True)


def normalize_text(text: str) -> str:
    text = html_lib.unescape(text or "")
    text = text.replace("\xa0", " ").replace("\u3000", " ")
    lines = [re.sub(r"[ \t\r\f\v]+", " ", line).strip() for line in text.split("\n")]
    return "\n".join(line for line in lines if line)


def html_to_text(fragment: str) -> str:
    if not fragment:
        return ""
    fragment = SCRIPT_STYLE_RE.sub("", fragment)
    fragment = re.sub(r"<br\s*/?>", "\n", fragment, flags=re.I)
    fragment = re.sub(r"</(p|div|h1|h2|h3|li|tr)>\s*", "\n", fragment, flags=re.I)
    fragment = re.sub(r"<li\b[^>]*>", "\n", fragment, flags=re.I)
    return normalize_text(TAG_RE.sub("", fragment))


def get_attr(attrs: str, name: str) -> Optional[str]:
    match = re.search(ATTR_RE_TEMPLATE.format(re.escape(name)), attrs or "", re.I | re.S)
    if not match:
        return None
    return html_lib.unescape(match.group(2))


def has_class(attrs: str, class_name: str) -> bool:
    value = get_attr(attrs, "class") or ""
    return class_name in value.split()


def find_div_blocks(
    html: str,
    *,
    class_name: Optional[str] = None,
    id_value: Optional[str] = None,
    start: int = 0,
) -> List[Tuple[str, int, int]]:
    """Return balanced div inner HTML blocks matching class or id."""
    blocks: List[Tuple[str, int, int]] = []
    pos = start
    while True:
        match = DIV_TAG_RE.search(html, pos)
        if not match:
            break
        closing, attrs = match.group(1), match.group(2)
        pos = match.end()
        if closing:
            continue
        if class_name and not has_class(attrs, class_name):
            continue
        if id_value and get_attr(attrs, "id") != id_value:
            continue

        depth = 1
        inner_start = match.end()
        scan = match.end()
        while True:
            next_match = DIV_TAG_RE.search(html, scan)
            if not next_match:
                break
            if next_match.group(1):
                depth -= 1
                if depth == 0:
                    blocks.append((html[inner_start:next_match.start()], match.start(), next_match.end()))
                    pos = next_match.end()
                    break
            else:
                depth += 1
            scan = next_match.end()
        else:
            break
    return blocks


def is_blocked_ip(ip_text: str) -> bool:
    try:
        ip = ipaddress.ip_address(ip_text)
    except ValueError:
        return True
    first_octet = int(ip_text.split(".", 1)[0]) if ip.version == 4 else None
    blocked_first_octets = {9, 10, 11, 21, 30}
    return (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
        or (first_octet in blocked_first_octets)
    )


def validate_url(url: str) -> None:
    parsed = urlparse(url)
    if parsed.scheme != "https":
        raise ValueError(f"Only https URLs are allowed: {url}")
    host = (parsed.hostname or "").lower()
    if host not in ALLOWED_HOSTS:
        raise ValueError(f"URL host is not allowed: {host}")
    try:
        for family, _, _, _, sockaddr in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
            ip_text = sockaddr[0]
            if is_blocked_ip(ip_text):
                raise ValueError(f"Resolved to blocked/internal IP: {host} -> {ip_text}")
    except socket.gaierror as exc:
        raise ValueError(f"Cannot resolve host {host}: {exc}") from exc


def fetch_text(url: str, *, timeout: int = 20, retries: int = 3, user_agent: str = DEFAULT_USER_AGENT) -> str:
    validate_url(url)
    last_error: Optional[Exception] = None
    for attempt in range(1, retries + 1):
        try:
            req = Request(
                url,
                headers={
                    "User-Agent": user_agent,
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
                },
            )
            with urlopen(req, timeout=timeout) as resp:
                final_url = resp.geturl()
                validate_url(final_url)
                charset = resp.headers.get_content_charset() or "utf-8"
                return resp.read().decode(charset, errors="replace")
        except (HTTPError, URLError, TimeoutError, ValueError) as exc:
            last_error = exc
            if attempt < retries:
                time.sleep(min(2 * attempt, 5))
    raise RuntimeError(f"Fetch failed: {url}; last error: {last_error}")


def parse_tangshi_list(page_html: str) -> List[Dict[str, object]]:
    type_blocks = find_div_blocks(page_html, class_name="typecont")
    if not type_blocks:
        raise RuntimeError("Cannot find 唐诗三百首 list block: div.typecont")
    token_re = re.compile(
        r"<div\s+class=['\"]bookMl['\"]\s*>\s*<strong>(.*?)</strong>\s*</div>"
        r"|<span>\s*<a\s+href=['\"]([^'\"]+)['\"][^>]*>(.*?)</a>\s*[（(]([^）)]+)[）)]\s*</span>",
        re.I | re.S,
    )
    items: List[Dict[str, object]] = []
    current_category = ""
    category_order = 0
    for list_html, _, _ in type_blocks:
        for match in token_re.finditer(list_html):
            if match.group(1) is not None:
                current_category = html_to_text(match.group(1))
                category_order += 1
                continue
            href = html_lib.unescape(match.group(2) or "")
            title = html_to_text(match.group(3) or "")
            author = normalize_text(match.group(4) or "")
            if not href or not title:
                continue
            items.append(
                {
                    "index": len(items) + 1,
                    "category": current_category,
                    "category_order": category_order,
                    "title": title,
                    "list_author": author,
                    "url": urljoin(BASE_URL, href),
                }
            )
    if not items:
        raise RuntimeError("No poem links parsed from list page")
    return items


def extract_poem_id(url: str) -> str:
    match = re.search(r"shiwenv_([0-9a-fA-F]+)\.aspx", url)
    return match.group(1) if match else ""


def parse_source(source_html: str) -> Tuple[str, str]:
    anchors = re.findall(r"<a\b[^>]*>(.*?)</a>", source_html or "", flags=re.I | re.S)
    texts = [html_to_text(anchor) for anchor in anchors]
    texts = [text for text in texts if text]
    author = texts[0] if texts else ""
    dynasty = texts[1] if len(texts) > 1 else ""
    dynasty = dynasty.strip("[]【】〔〕()（） ")
    return author, dynasty


def parse_tags(page_html: str) -> List[str]:
    tag_blocks = find_div_blocks(page_html, class_name="tag")
    if not tag_blocks:
        return []
    texts = [html_to_text(anchor) for anchor in re.findall(r"<a\b[^>]*>(.*?)</a>", tag_blocks[0][0], re.I | re.S)]
    return [text for text in texts if text]


def parse_sections(page_html: str) -> List[Dict[str, str]]:
    sections: List[Dict[str, str]] = []
    for block, _, _ in find_div_blocks(page_html, class_name="contyishang"):
        title_match = re.search(r"<h2\b[^>]*>(.*?)</h2>", block, re.I | re.S)
        title = html_to_text(title_match.group(1)) if title_match else ""
        body_html = block
        if title_match:
            body_html = block[: title_match.start()] + block[title_match.end() :]
        # Remove audio/play links and correction links from section body.
        body_html = re.sub(r"<a\b[^>]*javascript:Play[^>]*>.*?</a>", "", body_html, flags=re.I | re.S)
        body_html = re.sub(r"<a\b[^>]*jiucuo\.aspx[^>]*>.*?</a>", "", body_html, flags=re.I | re.S)
        text = html_to_text(body_html)
        if title and text:
            sections.append({"title": title, "text": text})
    return sections


def parse_detail(page_html: str, url: str) -> Dict[str, object]:
    poem_id = extract_poem_id(url)
    zhengwen_blocks = find_div_blocks(page_html, id_value=f"zhengwen{poem_id}") if poem_id else []
    zhengwen_html = zhengwen_blocks[0][0] if zhengwen_blocks else page_html

    title_match = re.search(r"<h1\b[^>]*>(.*?)</h1>", zhengwen_html, re.I | re.S)
    title = html_to_text(title_match.group(1)) if title_match else ""

    source_match = re.search(r"<p\s+class=['\"]source['\"][^>]*>(.*?)</p>", zhengwen_html, re.I | re.S)
    author, dynasty = parse_source(source_match.group(1) if source_match else "")

    content = ""
    if poem_id:
        content_match = re.search(
            rf"<div\s+class=['\"]contson['\"]\s+id=['\"]contson{re.escape(poem_id)}['\"][^>]*>(.*?)</div>",
            zhengwen_html,
            re.I | re.S,
        )
        if content_match:
            content = html_to_text(content_match.group(1))
    if not content:
        content_match = re.search(r"<div\s+class=['\"]contson['\"][^>]*>(.*?)</div>", zhengwen_html, re.I | re.S)
        if content_match:
            content = html_to_text(content_match.group(1))

    sections = parse_sections(page_html)
    extra: Dict[str, str] = {}
    for section in sections:
        section_title = section["title"]
        if "译文" in section_title or "注释" in section_title:
            extra.setdefault("translation_annotation", section["text"])
        elif "赏析" in section_title or "鉴赏" in section_title or "简析" in section_title:
            extra.setdefault("appreciation", section["text"])
        elif "背景" in section_title:
            extra.setdefault("background", section["text"])

    return {
        "poem_id": poem_id,
        "detail_title": title,
        "author": author,
        "dynasty": dynasty,
        "content": content,
        "tags": parse_tags(page_html),
        "sections": sections,
        "translation_annotation": extra.get("translation_annotation", ""),
        "appreciation": extra.get("appreciation", ""),
        "background": extra.get("background", ""),
    }


def write_outputs(records: List[Dict[str, object]], output_dir: Path, source_url: str) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    scraped_at = datetime.now(timezone.utc).isoformat()

    payload = {
        "source": source_url,
        "site": "古文岛/古诗文网",
        "scraped_at": scraped_at,
        "count": len(records),
        "records": records,
    }
    json_path = output_dir / "tangshi300.json"
    json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")

    jsonl_path = output_dir / "tangshi300.jsonl"
    with jsonl_path.open("w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    csv_path = output_dir / "tangshi300.csv"
    fieldnames = [
        "index",
        "category",
        "title",
        "author",
        "dynasty",
        "content",
        "url",
        "tags",
        "translation_annotation",
        "appreciation",
        "background",
        "sections_json",
        "error",
    ]
    with csv_path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for record in records:
            writer.writerow(
                {
                    "index": record.get("index", ""),
                    "category": record.get("category", ""),
                    "title": record.get("title", ""),
                    "author": record.get("author", "") or record.get("list_author", ""),
                    "dynasty": record.get("dynasty", ""),
                    "content": record.get("content", ""),
                    "url": record.get("url", ""),
                    "tags": "；".join(record.get("tags", []) or []),
                    "translation_annotation": record.get("translation_annotation", ""),
                    "appreciation": record.get("appreciation", ""),
                    "background": record.get("background", ""),
                    "sections_json": json.dumps(record.get("sections", []), ensure_ascii=False),
                    "error": record.get("error", ""),
                }
            )

    categories: Dict[str, int] = {}
    errors = 0
    for record in records:
        categories[str(record.get("category", ""))] = categories.get(str(record.get("category", "")), 0) + 1
        if record.get("error"):
            errors += 1
    summary_lines = [
        f"source: {source_url}",
        f"scraped_at: {scraped_at}",
        f"count: {len(records)}",
        f"errors: {errors}",
        "categories:",
    ]
    summary_lines.extend(f"  - {name}: {count}" for name, count in categories.items())
    (output_dir / "tangshi300_summary.txt").write_text("\n".join(summary_lines) + "\n", encoding="utf-8")


def crawl(args: argparse.Namespace) -> List[Dict[str, object]]:
    log(f"Fetch list: {args.start_url}")
    list_html = fetch_text(args.start_url, timeout=args.timeout, retries=args.retries, user_agent=args.user_agent)
    items = parse_tangshi_list(list_html)
    if args.limit:
        items = items[: args.limit]
    log(f"Parsed {len(items)} poem links")

    records: List[Dict[str, object]] = []
    for item in items:
        record = dict(item)
        if args.skip_details:
            records.append(record)
            continue
        url = str(item["url"])
        try:
            log(f"[{item['index']}/{len(items)}] {item['title']} - {url}")
            detail_html = fetch_text(url, timeout=args.timeout, retries=args.retries, user_agent=args.user_agent)
            detail = parse_detail(detail_html, url)
            record.update(detail)
            if not record.get("author"):
                record["author"] = record.get("list_author", "")
            if detail.get("detail_title"):
                record["title"] = detail["detail_title"]
        except Exception as exc:  # keep remaining records available
            record["error"] = str(exc)
            log(f"  ERROR: {exc}")
        records.append(record)
        if args.delay > 0:
            time.sleep(args.delay)
    return records


def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Crawl 唐诗三百首 from guwendao.net")
    parser.add_argument("--start-url", default=START_URL, help="唐诗三百首列表页 URL")
    parser.add_argument("--output-dir", default=str(Path(__file__).resolve().parents[1]), help="数据输出目录")
    parser.add_argument("--delay", type=float, default=0.3, help="每个详情页之间的延迟秒数")
    parser.add_argument("--timeout", type=int, default=20, help="单次请求超时秒数")
    parser.add_argument("--retries", type=int, default=3, help="请求失败重试次数")
    parser.add_argument("--limit", type=int, default=0, help="只爬前 N 条；0 表示全部")
    parser.add_argument("--skip-details", action="store_true", help="只保存列表页条目，不爬详情页")
    parser.add_argument("--user-agent", default=DEFAULT_USER_AGENT, help="请求 User-Agent")
    return parser


def main(argv: Optional[Iterable[str]] = None) -> int:
    args = build_arg_parser().parse_args(argv)
    output_dir = Path(args.output_dir).expanduser().resolve()
    try:
        records = crawl(args)
        write_outputs(records, output_dir, args.start_url)
    except Exception as exc:
        print(f"Fatal: {exc}", file=sys.stderr)
        return 1
    log(f"Done. Files saved to: {output_dir}")
    return 0


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