深度学习在DOM解析中的应用：自动识别页面关键内容区块

import time
import requests
from bs4 import BeautifulSoup
import sqlite3
import tensorflow as tf
import numpy as np
from concurrent.futures import ThreadPoolExecutor

# -----配置代理 IP（亿牛云爬虫代理示例 www.16yun.cn）-----
PROXY_HOST = "proxy.16yun.cn"
PROXY_PORT = 8100
PROXY_USER = "16YUN"
PROXY_PASS = "16IP"
proxy_meta = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
proxies = {
    "http": proxy_meta,
    "https": proxy_meta,
}

# --------- HTTP 会话与头部设置 ---------
session = requests.Session()
session.proxies.update(proxies)
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36",
})
# 示例 Cookie 设置
session.cookies.set("st_si", "123456789", domain="eastmoney.com")
session.cookies.set("st_asi", "abcdefg", domain="eastmoney.com")


# --------- 数据库初始化 ---------
conn = sqlite3.connect("eastmoney_news.db", check_same_thread=False)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS news (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT,
    pub_time TEXT,
    content TEXT
)
""")
conn.commit()


# --------- 加载并量化模型（TFLite） ---------
# 假设已有 Keras 模型 'content_block_model.h5'，先转换为 TFLite
def convert_to_tflite(h5_path, tflite_path):
    model = tf.keras.models.load_model(h5_path)
    converter = tf.lite.TFLiteConverter.from_keras_model(model)
    converter.optimizations = [tf.lite.Optimize.DEFAULT]
    tflite_model = converter.convert()
    with open(tflite_path, "wb") as f:
        f.write(tflite_model)

# convert_to_tflite("content_block_model.h5", "model_quant.tflite")  # 一次性执行

# 加载 TFLite 模型
interpreter = tf.lite.Interpreter(model_path="model_quant.tflite")
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# 模型预测函数（批量）
def predict_blocks(text_list):
    # 文本预处理示例：截断/填充到固定长度
    seqs = [[ord(c) for c in text[:200]] + [0] * (200 - len(text[:200])) for text in text_list]
    inp = np.array(seqs, dtype=np.float32)
    interpreter.set_tensor(input_details[0]['index'], inp)
    interpreter.invoke()
    preds = interpreter.get_tensor(output_details[0]['index'])
    # 返回布尔列表，True 表示为“关键区块”
    return [bool(p[1] > 0.5) for p in preds]


# --------- 爬取解析函数 ---------
def fetch_and_parse(url):
    start = time.time()

    # 1. 获取页面
    resp = session.get(url, timeout=10)
    resp.raise_for_status()
    # 2. 解析 DOM
    soup = BeautifulSoup(resp.text, "lxml")
    # 3. 提取候选区块（示例：所有 <div>）
    divs = soup.find_all("div")
    texts = [div.get_text(strip=True) for div in divs]
    # 4. 批量模型预测
    is_key = predict_blocks(texts)
    # 5. 筛选关键区块并抽取新闻字段
    news_items = []
    for div, flag in zip(divs, is_key):
        if not flag:
            continue
        title_tag = div.find("h1") or div.find("h2") or div.find("h3")
        time_tag = div.find("span", class_="time") or div.find("em")
        content = div.get_text(strip=True)
        if title_tag and time_tag:
            news_items.append({
                "title": title_tag.get_text(strip=True),
                "pub_time": time_tag.get_text(strip=True),
                "content": content
            })
    # 6. 存储到 SQLite
    for item in news_items:
        cursor.execute(
            "INSERT INTO news (title, pub_time, content) VALUES (?, ?, ?)",
            (item["title"], item["pub_time"], item["content"])
        )
    conn.commit()

    end = time.time()
    return end - start  # 返回耗时


# --------- 多线程压测 ---------
if __name__ == "__main__":
    urls = [
        "https://www.eastmoney.com/a/20250422XYZ.html",
        # ... 更多新闻页 URL 列表 ...
    ]
    with ThreadPoolExecutor(max_workers=5) as executor:
        times = list(executor.map(fetch_and_parse, urls))
    print(f"平均单页耗时：{sum(times)/len(times):.2f} 秒")