Trafilatura 高级技巧：自定义提取、语言检测与性能优化

自定义提取策略

排除特定元素

Trafilatura 允许在提取前对 HTML 进行预处理，排除不需要的部分：

import trafilatura
from lxml import html

# 下载 HTML
downloaded = trafilatura.fetch_url("https://example.com/article")

# 预处理：移除特定元素后再提取
tree = html.fromstring(downloaded)
for element in tree.xpath("//aside | //nav | //div[@class='sidebar']"):
    element.getparent().remove(element)

extracted_html = html.tostring(tree, encoding="unicode")
result = trafilatura.extract(extracted_html, output_format="markdown")

只提取特定区域

如果你的目标页面有明确的正文容器，可以缩小提取范围：

import trafilatura
from lxml import html

downloaded = trafilatura.fetch_url("https://example.com/article")
tree = html.fromstring(downloaded)

# 只提取 <article> 标签内的内容
article_elem = tree.xpath("//article")
if article_elem:
    article_html = html.tostring(article_elem[0], encoding="unicode")
    result = trafilatura.extract(article_html, output_format="markdown")
else:
    # 回退到全页提取
    result = trafilatura.extract(downloaded, output_format="markdown")

自定义保留标签

# 在 Markdown 输出中保留代码块和行内代码
result = trafilatura.extract(
    downloaded,
    output_format="markdown",
    include_formatting=True,  # 保留加粗/斜体/代码
    include_tables=True,      # 保留表格
)

语言检测

Trafilatura 可以集成语言检测，在处理多语言站点时自动识别内容语言：

pip install trafilatura[langid]

或使用更准确的 fasttext 模型：

pip install trafilatura[fasttext]

提取时检测语言

import trafilatura

downloaded = trafilatura.fetch_url("https://example.com/article")
result = trafilatura.extract(
    downloaded,
    output_format="json",
    with_metadata=True,
    target_language="zh",     # 可选：指定目标语言筛选
)

import json
data = json.loads(result)

# 使用 fasttext 检测语言
from trafilatura import language
lang = language.detect(data.get("text", ""))
print(f"检测到语言: {lang}")

按语言筛选

from trafilatura import language

# 检测并过滤非中文内容
text = "Some English text..."
lang, confidence = language.detect_with_confidence(text)
if lang == "zh" and confidence > 0.8:
    print("高质量中文内容")

检测引擎	准确率	速度	安装
`langid`	中等	快	`trafilatura[langid]`
`fasttext`	高	中等	`trafilatura[fasttext]`

离线 HTML 批量处理

如果你已经有一个 HTML 文件目录，可以批量离线处理——不需要网络：

import os
import trafilatura
from pathlib import Path

html_dir = Path("./downloaded_pages")
output_dir = Path("./extracted_articles")
output_dir.mkdir(exist_ok=True)

for html_file in html_dir.glob("*.html"):
    with open(html_file, "r", encoding="utf-8") as f:
        html_content = f.read()

    result = trafilatura.extract(
        html_content,
        output_format="markdown",
        with_metadata=True,
    )

    if result:
        output_file = output_dir / f"{html_file.stem}.md"
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(result)
        print(f"✅ {html_file.name} → {output_file.name}")

CLI 批量处理本地文件

# 处理 HTML 文件目录
trafilatura --input-dir ./downloaded_pages --output-dir ./articles

# 处理单个文件
trafilatura --input-file ./page.html --output-file ./article.md

提取速度优化

缓存下载结果

import hashlib
import json
import os
import requests
import trafilatura

CACHE_DIR = "./cache"

def cached_fetch(url):
    """带缓存的下载"""
    cache_key = hashlib.md5(url.encode()).hexdigest()
    cache_path = os.path.join(CACHE_DIR, f"{cache_key}.html")

    if os.path.exists(cache_path):
        with open(cache_path, "r", encoding="utf-8") as f:
            return f.read()

    os.makedirs(CACHE_DIR, exist_ok=True)
    resp = requests.get(url, proxies=PROXY, timeout=20)
    resp.encoding = "utf-8"

    with open(cache_path, "w", encoding="utf-8") as f:
        f.write(resp.text)

    return resp.text

配合缓存，增量更新时可以复用已下载的 HTML，只处理新增 URL。

批量初始化

Trafilatura 在第一次导入时会初始化一些资源。如果批量处理大量页面，可以提前预热：

# 预热——导入后做一次小提取，让模块完成初始化
import trafilatura
_ = trafilatura.extract("<html><body><p>warmup</p></body></html>")

限制处理范围

当页面非常大时，可以限制处理长度来提升速度：

# 只提取正文的前 10000 字符
result = trafilatura.extract(
    downloaded,
    output_format="txt",
    max_tree_size=10000,  # 限制解析的 DOM 节点数
)

增量更新策略

对于持续运行的采集任务，增量更新可以避免重复下载相同页面：

import json
import os
from datetime import datetime

STATE_FILE = "crawl_state.json"

def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return json.load(f)
    return {"processed_urls": [], "last_crawl": None}

def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)

# 加载状态
state = load_state()
processed = set(state.get("processed_urls", []))

# 仅处理新 URL
new_urls = [u for u in all_urls if u not in processed]

for url in new_urls:
    # ... 下载和提取 ...
    processed.add(url)

# 保存状态
state["processed_urls"] = list(processed)
state["last_crawl"] = datetime.now().isoformat()
save_state(state)

总结

技巧	适用场景	关键代码
自定义元素排除	页面有固定噪声区域	`lxml` 预处理
限定提取区域	页面结构清晰	`//article` XPath
语言检测	多语言采集	`trafilatura[fasttext]`
离线批量	已有 HTML 文件	`trafilatura --input-dir`
下载缓存	频繁重新处理	MD5 缓存
增量更新	长期运营任务	JSON 状态文件