企业官网建设流程全解析

前言

「爬一个新闻站」和「做一个新闻爬取系统」的区别在于：前者抓一页就结束，后者要不停地把「列表页发现链接、详情页抽取正文、结果去重入库」这条流水线跑通，而且中途任何一个环节出错都不能让整批任务崩掉。

常见误解是「新闻站结构简单，写个正则就行」。实际新闻页恰恰是最不规整的一类：正文里嵌着广告区块、相关阅读、图片说明，标题可能同时存在于title和h1，日期格式五花八门。用事件驱动的解析器按标签收文本，比用正则稳得多。

本文用标准库搭一个最小但完整的新闻爬取系统骨架：urllib.request发请求、html.parser抽字段、sqlite3去重入库，全程遵守 robots.txt 并限速。示例适用于 Python 3.8 及以上。

需要先说清楚合规边界：本文的域名是占位示例，实际使用必须换成你自己有权抓取的站点，并先确认其 robots.txt 与服务条款。新闻内容受著作权保护，采集结果只适合做个人研究或趋势统计，不要整篇转载。Python 2 已于 2020 年 1 月 1 日停止维护，本文只给 Python 3 写法。

一、系统的四个部分

一个新闻爬取系统，本质是四段流水线：

分段的好处是：哪一段出问题就换哪一段，不影响其他环节。抓取失败不该导致已入库的数据丢失，抽取规则变了也不该影响链接发现。

二、列表页：发现文章链接

列表页的任务只有一个——把详情页地址找出来。做法和通用爬虫一样：覆写handle_starttag，对a标签取href，再用urljoin补成绝对地址。

# 适用于 Python 3.8+
from html.parser import HTMLParser
from urllib.parse import urljoin, urlsplit


class LinkParser(HTMLParser):
def __init__(self, base):
super().__init__(convert_charrefs=True)
self.base = base
self.links = []

def handle_starttag(self, tag, attrs):
if tag != "a":
return
for name, value in attrs:
if name == "href" and value:
self.links.append(urljoin(self.base, value))


def list_page_links(html, base, host=None):
parser = LinkParser(base)
parser.feed(html)
parser.close()
result = []
seen = set()
for link in parser.links:
parts = urlsplit(link)
if parts.scheme not in ("http", "https"):
continue
if host and parts.hostname != host:
continue
if link in seen:
continue
seen.add(link)
result.append(link)
return result

过滤同站用urlsplit(link).hostname，不要用字符串前缀判断，否则带相同前缀的其他域名会混进来。

三、详情页：抽取标题与正文

详情页抽取的关键是「只收想要的标签里的文本」。新闻正文通常集中在一连串p标签里，标题在h1或title里。用状态标志来决定当前是否处于目标标签内。

# 适用于 Python 3.8+
from html.parser import HTMLParser


class ArticleParser(HTMLParser):
"""从详情页中抽取标题与正文段落。"""

def __init__(self):
super().__init__(convert_charrefs=True)
self.title = ""
self.paragraphs = []
self._in_title = False
self._in_h1 = False
self._in_p = False
self._buf = []

def handle_starttag(self, tag, attrs):
if tag in ("title", "h1"):
self._in_title = tag == "title"
self._in_h1 = tag == "h1"
self._buf = []
elif tag == "p":
self._in_p = True
self._buf = []

def handle_data(self, data):
if self._in_title or self._in_h1 or self._in_p:
self._buf.append(data)

def handle_endtag(self, tag):
if tag == "title" and self._in_title:
self._in_title = False
self._set_title()
elif tag == "h1" and self._in_h1:
self._in_h1 = False
self._set_title()
elif tag == "p" and self._in_p:
self._in_p = False
text = "".join(self._buf).strip()
if text:
self.paragraphs.append(text)

def _set_title(self):
if not self.title:
self.title = "".join(self._buf).strip()

def body(self):
return "\n".join(self.paragraphs)

三个容易忽略的点：convert_charrefs=True（默认值）会把&之类的实体自动还原成字符；属性值可能是None，取href时要判空；handle_data在标签内外都会被调用，靠状态标志过滤。

四、入库与去重

用url做主键，INSERT OR IGNORE天然完成去重：同一个地址抓两次也只留一条。

# 适用于 Python 3.8+
import sqlite3


def init_db(conn):
conn.execute(
"CREATE TABLE IF NOT EXISTS articles ("
" url TEXT PRIMARY KEY,"
" title TEXT,"
" body TEXT,"
" fetched_at TEXT"
")"
)
conn.commit()


def save_article(conn, url, title, body, fetched_at):
conn.execute(
"INSERT OR IGNORE INTO articles (url, title, body, fetched_at) "
"VALUES (?, ?, ?, ?)",
(url, title, body, fetched_at),
)
conn.commit()

用参数占位符?传值，不要用字符串拼接 SQL——那既容易出错，也是 SQL 注入的入口。这里的值虽然来自解析结果而非用户输入，但养成参数化习惯没有坏处。

实战：完整的采集流程

把四段串起来，加上 robots 检查、异常隔离和限速。

# 适用于 Python 3.8+
import datetime
import sqlite3
import time
import urllib.error
import urllib.request
import urllib.robotparser
from urllib.parse import urlsplit

USER_AGENT = "NewsReader/0.1 (contact: me@example.com)"
HEADERS = {"User-Agent": USER_AGENT}
DB = "news.db"


def robots_allows(url, user_agent):
parts = urlsplit(url)
rp = urllib.robotparser.RobotFileParser()
rp.set_url(parts.scheme + "://" + parts.netloc + "/robots.txt")
try:
rp.read()
except urllib.error.URLError:
return False # 拿不到 robots.txt 就保守处理
return rp.can_fetch(user_agent, url)


def fetch_text(url, timeout=10):
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
charset = resp.headers.get_content_charset() or "utf-8"
return raw.decode(charset, errors="replace")


def main():
list_url = "https://www.example.com/news/"
host = urlsplit(list_url).hostname

if not robots_allows(list_url, USER_AGENT):
print("robots.txt 不允许抓取列表页，退出")
return

html = fetch_text(list_url)
links = list_page_links(html, list_url, host=host)
print("发现文章链接:", len(links))

conn = sqlite3.connect(DB)
init_db(conn)

for link in links[:5]: # 演示：只取前 5 条
if not robots_allows(link, USER_AGENT):
print("robots 禁止:", link)
continue
try:
page = fetch_text(link)
except urllib.error.HTTPError as e:
print("HTTP 错误:", e.code, link)
continue
except urllib.error.URLError as e:
print("网络错误:", e.reason, link)
continue

parser = ArticleParser()
parser.feed(page)
parser.close()

now = datetime.datetime.now().isoformat(timespec="seconds")
save_article(conn, link, parser.title, parser.body(), now)
print("入库:", parser.title or link)
time.sleep(1) # 限速

conn.close()


if __name__ == "__main__":
main()

注意main里只保存了links[:5]用于演示。真实系统会分页遍历列表页、把链接推到队列、抽不到正文的页面记录原因以便回头排查。

常见坑点

1. 
   
2. 用正则抽正文
3. 
   

❌re.findall(r"<p>(.*?)</p>", html)—— 换行、嵌套标签、属性变化都会漏。 ✅ 用HTMLParser子类按标签收文本，状态标志控制范围。

1. 
   
2. 标题只认一种来源
3. 
   

❌ 只看h1—— 有些页面标题在title里，或在h1里带站点后缀。 ✅ 两个都收，先到先得，再按需做清洗。

1. 
   
2. 忘记urljoin补全相对地址
3. 
   

❌ 直接把href="2024/01-01.html"入库，后续无法再次访问。 ✅ 用urljoin(base, href)转成绝对地址。

1. 
   
2. 用字符串前缀判断同站
3. 
   

❌link.startswith("https://www.example.com")—— 前缀相同的其他域名会混入。 ✅ 比较urlsplit(link).hostname。

1. 
   
2. 列表页一页到底，不去重
3. 
   

❌ 同一个链接被多个区块重复引用，就抓多遍。 ✅ 入seen集合去重，入库用url主键 +INSERT OR IGNORE。

1. 
   
2. 用字符串拼接 SQL
3. 
   

❌"INSERT INTO articles VALUES ('" + url + "')"—— 地址里有引号就出错。 ✅ 用参数占位符conn.execute(sql, (url, ...))。

1. 
   
2. 一处失败拖垮整批
3. 
   

❌ 不包try，第 3 个链接超时后整个脚本退出。 ✅ 逐个链接包异常处理，失败就跳过并记录。

1. 
   
2. 不查 robots、不限速
3. 
   

❌ 循环里无间隔请求列表页和详情页。 ✅ 先can_fetch判断，请求之间time.sleep。

总结

系统的骨架就是这条流水线，把它跑稳之后，提升点在于分页遍历、增量抓取和抽取规则的维护。先把「发现—抓取—抽取—入库」四段各自隔离好，系统才有长大的空间。