编写数据爬取脚本

以下脚本使用 requests 与 Beautiful Soup 从指定栏目页抓取新闻数据，并包含基础的数据预处理（文本清洗、去重）、分页与速率控制、robots.txt 合规检查，以及可选的详情页补全。输出支持 CSV 与 JSONL，便于后续数据挖掘（特征工程、建模）的输入。

说明与假设：

• 由于站点结构未知，脚本采用多组通用选择器与回退策略，尽可能抽取标题、链接、发布时间、作者与摘要。
• 脚本会检查并遵守 robots.txt；请确保抓取行为符合目标站点的使用条款与法律法规。
• 若页面依赖 JavaScript 动态渲染，需改用渲染型方案（如 Playwright/Selenium）；本脚本仅处理静态 HTML。

使用方法：

1. 安装依赖：pip install requests beautifulsoup4
2. 运行示例：python scrape_marketing_insights.py 可选参数示例：python scrape_marketing_insights.py --url https://news.example.net/industry/marketing-insights --output-csv insights.csv --output-jsonl insights.jsonl --max-pages 3 --detail

Python脚本（保存为 scrape_marketing_insights.py）：

import argparse
import csv
import json
import logging
import random
import re
import sys
import time
from dataclasses import dataclass, asdict
from typing import List, Optional, Set
from urllib.parse import urljoin, urlparse
from urllib.robotparser import RobotFileParser

import requests
from bs4 import BeautifulSoup
from bs4.element import Tag

# -----------------------
# 配置与工具函数
# -----------------------

USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0 Safari/537.36"
DEFAULT_URL = "https://news.example.net/industry/marketing-insights"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)

@dataclass
class Article:
    title: str
    url: str
    published_at: Optional[str]
    author: Optional[str]
    summary: Optional[str]
    source: str

def normalize_whitespace(text: Optional[str]) -> Optional[str]:
    if text is None:
        return None
    return re.sub(r"\s+", " ", text).strip()

def build_robot_parser(start_url: str) -> RobotFileParser:
    parsed = urlparse(start_url)
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
    rp = RobotFileParser()
    rp.set_url(robots_url)
    try:
        rp.read()
        logging.info(f"robots.txt 加载完成: {robots_url}")
    except Exception as e:
        logging.warning(f"robots.txt 加载失败: {robots_url} -> {e}")
    return rp

def can_fetch(rp: RobotFileParser, url: str, user_agent: str = USER_AGENT) -> bool:
    # 如果 robots.txt 无法读取，RobotFileParser 默认返回 True
    try:
        return rp.can_fetch(user_agent, url)
    except Exception:
        return True

def get_session() -> requests.Session:
    s = requests.Session()
    s.headers.update({
        "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",
    })
    return s

# -----------------------
# 解析函数（列表页与详情页）
# -----------------------

def find_article_containers(soup: BeautifulSoup) -> List[Tag]:
    # 通用选择器组合，尽可能兼容不同站点结构
    containers: List[Tag] = []
    containers.extend(soup.find_all("article"))
    containers.extend(soup.select('div[itemtype*="NewsArticle"], div[itemtype*="Article"]'))
    for cls in ["news", "article", "post", "story", "card", "entry"]:
        containers.extend(soup.select(f"div[class*='{cls}'], li[class*='{cls}']"))
    # 去重：以元素的字符串位置标识符（sourceline）+名称近似去重
    seen_ids = set()
    unique = []
    for tag in containers:
        key = (tag.name, getattr(tag, "sourceline", None), getattr(tag, "sourcepos", None))
        if key not in seen_ids:
            seen_ids.add(key)
            unique.append(tag)
    return unique

def extract_title_and_url(container: Tag, base_url: str) -> (Optional[str], Optional[str]):
    # 优先从标题标签提取
    title_tag = None
    for sel in ["h1 a", "h2 a", "h3 a", "h1", "h2", "h3", "a"]:
        title_tag = container.select_one(sel)
        if title_tag:
            break
    title = None
    url = None
    if title_tag:
        if title_tag.name == "a":
            title = title_tag.get_text(strip=True) or None
            href = title_tag.get("href")
            if href:
                url = urljoin(base_url, href)
        else:
            title = title_tag.get_text(strip=True) or None
            a = title_tag.find("a", href=True)
            if a and a.get("href"):
                url = urljoin(base_url, a["href"])
    # 回退：任何可点击链接
    if not url:
        a = container.find("a", href=True)
        if a and a.get_text(strip=True):
            url = urljoin(base_url, a["href"])
            title = title or a.get_text(strip=True)
    return normalize_whitespace(title), url

def extract_published_at(container: Tag) -> Optional[str]:
    # time 标签
    t = container.find("time")
    if t:
        dt = t.get("datetime") or t.get_text(strip=True)
        if dt:
            return normalize_whitespace(dt)
    # meta 标签（结构化）
    for sel in [
        'meta[itemprop="datePublished"]',
        'meta[property="article:published_time"]',
        'meta[name="pubdate"]',
        'span[itemprop="datePublished"]',
    ]:
        m = container.select_one(sel)
        if m:
            content = m.get("content") or m.get_text(strip=True)
            if content:
                return normalize_whitespace(content)
    return None

def extract_author(container: Tag) -> Optional[str]:
    for sel in [
        ".author", ".byline", '[itemprop="author"]', 'a[rel="author"]', 'meta[name="author"]',
    ]:
        el = container.select_one(sel)
        if el:
            content = el.get("content") if el.name == "meta" else el.get_text(strip=True)
            if content:
                return normalize_whitespace(content)
    return None

def extract_summary(container: Tag) -> Optional[str]:
    for sel in [".summary", ".dek", ".abstract", "p"]:
        el = container.select_one(sel)
        if el:
            text = el.get_text(" ", strip=True)
            if text and len(text.split()) >= 3:
                return normalize_whitespace(text)
    return None

def parse_detail_page(soup: BeautifulSoup) -> dict:
    # 从详情页提取更规范的字段（若存在）
    data = {"title": None, "published_at": None, "author": None, "summary": None}
    # 标题
    title_tag = soup.find(["h1", "title"])
    if title_tag:
        data["title"] = normalize_whitespace(title_tag.get_text(strip=True))
    og_title = soup.select_one('meta[property="og:title"]')
    if og_title and og_title.get("content"):
        data["title"] = normalize_whitespace(og_title["content"]) or data["title"]
    # 摘要
    og_desc = soup.select_one('meta[property="og:description"], meta[name="description"]')
    if og_desc and og_desc.get("content"):
        data["summary"] = normalize_whitespace(og_desc["content"])
    # 发布时间
    for sel in [
        'meta[property="article:published_time"]',
        'meta[itemprop="datePublished"]',
        'time[datetime]',
    ]:
        el = soup.select_one(sel)
        if el:
            content = el.get("content") or el.get("datetime") or el.get_text(strip=True)
            if content:
                data["published_at"] = normalize_whitespace(content)
                break
    # 作者
    for sel in [
        'meta[name="author"]',
        '[itemprop="author"]',
        '.author', '.byline', 'a[rel="author"]',
    ]:
        el = soup.select_one(sel)
        if el:
            content = el.get("content") if el.name == "meta" else el.get_text(strip=True)
            if content:
                data["author"] = normalize_whitespace(content)
                break
    # JSON-LD（可选增强）
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            payload = json.loads(script.string or "")
        except Exception:
            continue
        objs = payload if isinstance(payload, list) else [payload]
        for obj in objs:
            t = obj.get("@type") or obj.get("@graph", [{}])[0].get("@type")
            if isinstance(t, list):
                t = ",".join(t)
            if t and ("NewsArticle" in t or "Article" in t):
                data["title"] = normalize_whitespace(obj.get("headline")) or data["title"]
                data["summary"] = normalize_whitespace(obj.get("description")) or data["summary"]
                pub = obj.get("datePublished") or obj.get("dateCreated")
                if pub:
                    data["published_at"] = normalize_whitespace(pub)
                author = obj.get("author")
                if isinstance(author, dict):
                    data["author"] = normalize_whitespace(author.get("name")) or data["author"]
                elif isinstance(author, list) and author:
                    if isinstance(author[0], dict):
                        data["author"] = normalize_whitespace(author[0].get("name")) or data["author"]
                    elif isinstance(author[0], str):
                        data["author"] = normalize_whitespace(author[0]) or data["author"]
                break
    return data

# -----------------------
# 抓取主流程
# -----------------------

def scrape_list(start_url: str,
                max_pages: int = 1,
                sleep_min: float = 1.0,
                sleep_max: float = 2.5,
                fetch_detail: bool = False) -> List[Article]:
    session = get_session()
    rp = build_robot_parser(start_url)
    articles: List[Article] = []
    seen_urls: Set[str] = set()

    def fetch(url: str) -> Optional[BeautifulSoup]:
        if not can_fetch(rp, url):
            logging.warning(f"robots 不允许抓取：{url}")
            return None
        try:
            resp = session.get(url, timeout=30)
            if resp.status_code != 200:
                logging.warning(f"请求失败 {resp.status_code}: {url}")
                return None
            return BeautifulSoup(resp.text, "html.parser")
        except requests.RequestException as e:
            logging.error(f"请求异常: {url} -> {e}")
            return None

    page_url = start_url
    base = f"{urlparse(start_url).scheme}://{urlparse(start_url).netloc}"

    for page_idx in range(max_pages):
        soup = fetch(page_url)
        if soup is None:
            break

        containers = find_article_containers(soup)
        if not containers:
            logging.info("未找到文章容器，尝试使用更宽松的链接抽取")
            # 宽松抽取：从主体区块抽取链接
            for a in soup.select("main a[href], .content a[href], .container a[href]"):
                title = normalize_whitespace(a.get_text(strip=True))
                if not title or len(title) < 5:
                    continue
                url = urljoin(base, a.get("href"))
                if url in seen_urls:
                    continue
                articles.append(Article(
                    title=title,
                    url=url,
                    published_at=None,
                    author=None,
                    summary=None,
                    source=start_url,
                ))
                seen_urls.add(url)
        else:
            for c in containers:
                title, url = extract_title_and_url(c, base)
                if not url or not title:
                    continue
                if url in seen_urls:
                    continue
                published_at = extract_published_at(c)
                author = extract_author(c)
                summary = extract_summary(c)

                # 详情页补全（可选）
                if fetch_detail and can_fetch(rp, url):
                    time.sleep(random.uniform(sleep_min, sleep_max))
                    dsoup = fetch(url)
                    if dsoup:
                        detail = parse_detail_page(dsoup)
                        title = detail["title"] or title
                        summary = detail["summary"] or summary
                        published_at = detail["published_at"] or published_at
                        author = detail["author"] or author

                articles.append(Article(
                    title=title,
                    url=url,
                    published_at=published_at,
                    author=author,
                    summary=summary,
                    source=start_url,
                ))
                seen_urls.add(url)

        # 分页：查找下一页链接
        next_link = soup.select_one('a[rel="next"], a.next, a[aria-label*="下一页"], a[aria-label*="Next"], li.next a')
        if next_link and next_link.get("href"):
            next_url = urljoin(base, next_link.get("href"))
            if next_url == page_url:
                break
            page_url = next_url
            logging.info(f"跳转下一页: {page_url}")
            time.sleep(random.uniform(sleep_min, sleep_max))
        else:
            break

    return articles

# -----------------------
# 输出与 CLI
# -----------------------

def write_csv(path: str, items: List[Article]) -> None:
    with open(path, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["title", "url", "published_at", "author", "summary", "source"])
        w.writeheader()
        for it in items:
            w.writerow(asdict(it))

def write_jsonl(path: str, items: List[Article]) -> None:
    with open(path, "w", encoding="utf-8") as f:
        for it in items:
            f.write(json.dumps(asdict(it), ensure_ascii=False) + "\n")

def main():
    parser = argparse.ArgumentParser(description="Marketing Insights 数据抓取（BeautifulSoup）")
    parser.add_argument("--url", type=str, default=DEFAULT_URL, help="起始列表页 URL")
    parser.add_argument("--max-pages", type=int, default=1, help="最大分页页数")
    parser.add_argument("--sleep-min", type=float, default=1.0, help="最小请求间隔（秒）")
    parser.add_argument("--sleep-max", type=float, default=2.5, help="最大请求间隔（秒）")
    parser.add_argument("--detail", action="store_true", help="是否抓取详情页补全字段")
    parser.add_argument("--output-csv", type=str, default=None, help="CSV 输出路径")
    parser.add_argument("--output-jsonl", type=str, default=None, help="JSONL 输出路径")
    args = parser.parse_args()

    items = scrape_list(
        start_url=args.url,
        max_pages=args.max_pages,
        sleep_min=args.sleep_min,
        sleep_max=args.sleep_max,
        fetch_detail=args.detail,
    )

    logging.info(f"抓取完成，共 {len(items)} 条")
    if args.output_csv:
        write_csv(args.output_csv, items)
        logging.info(f"CSV 已写入: {args.output_csv}")
    if args.output_jsonl:
        write_jsonl(args.output_jsonl, items)
        logging.info(f"JSONL 已写入: {args.output_jsonl}")

    # 如果未指定输出文件，打印示例前几条
    if not args.output_csv and not args.output_jsonl:
        for it in items[:5]:
            print(json.dumps(asdict(it), ensure_ascii=False))

if __name__ == "__main__":
    main()

数据挖掘建议（采集到数据后的处理要点）：

• 字段标准化：将 published_at 统一转换为 ISO8601；作者名进行规范化与实体消歧。
• 去重策略：依据 URL、标题+发布时间的组合键进行二次去重。
• 内容质量过滤：摘要长度阈值、关键词过滤（如“营销”“洞察”等）、语言检测，提升样本纯度。
• 结构化增强：若可用，优先采用页面中的结构化数据（JSON-LD）填补缺失字段。
• 合规与速率：遵守 robots 与站点条款；设置合理的间隔与并发限流，避免对站点造成压力。