基于大模型的Python智能爬虫：语义识别与数据清洗实践

import requests
import json
import random
import time
from bs4 import BeautifulSoup
from openai import OpenAI

# 模型接口配置
OPENAI_API_KEY = "your-api-key"
OPENAI_BASE_URL = "https://api.openai.com/v1"
MODEL_NAME = "gpt-4o-mini"

# 代理隧道配置
PROXY_CONFIG = {
    "host": "t.16yun.cn",
    "port": "31111",
    "user": "username",
    "pass": "password"
}
proxy_meta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % PROXY_CONFIG
PROXIES = {"http": proxy_meta, "https": proxy_meta}

# 模拟浏览器请求头
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/125.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
}

def fetch_page(url, use_proxy=True):
    """网页采集+异常容错+代理轮换"""
    req_headers = HEADERS.copy()
    if use_proxy:
        tunnel = random.randint(1, 10000)
        req_headers['Proxy-Tunnel'] = str(tunnel)
    try:
        resp = requests.get(url, headers=req_headers, proxies=PROXIES if use_proxy else None, timeout=15)
        resp.raise_for_status()
        return clean_html(resp.text)
    except requests.exceptions.HTTPError as e:
        status = e.response.status_code if e.response else 0
        if status == 429:time.sleep(3)
        print(f"HTTP异常码：{status}")
        return None
    except Exception as e:
        print(f"采集异常：{e}")
        return None

def clean_html(html):
    """HTML语义精炼：剔除噪音标签，压缩上下文"""
    soup = BeautifulSoup(html, 'html.parser')
    # 移除非功能性标签
    for tag in soup(['script','style','nav','footer','aside','iframe','form']):
        tag.decompose()
    # 清除无用属性，减少模型干扰
    for tag in soup.find_all(True):
        tag.attrs = {k:v for k,v in tag.attrs.items() if k in ['href','src','alt','title']}
    return str(soup)

client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)

def extract_with_llm(html_content, extraction_prompt, max_retries=2):
    """LLM通用结构化提取接口"""
    max_chars = 8000
    if len(html_content) > max_chars:
        html_content = html_content[:max_chars] + "\n...(截断)"
    system_prompt = """你是网页提取专家，严格按要求输出JSON；缺省值填null，禁止编造数据。"""
    user_prompt = f"{extraction_prompt}\nHTML内容：{html_content}"
    for _ in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=MODEL_NAME,
                messages=[{"role":"system","content":system_prompt},{"role":"user","content":user_prompt}],
                temperature=0.1,
                response_format={"type":"json_object"}
            )
            return json.loads(response.choices[0].message.content)
        except Exception:
            time.sleep(2)
            continue
    return None

client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)

def extract_with_llm(html_content, extraction_prompt, max_retries=2):
    """LLM通用结构化提取接口"""
    max_chars = 8000
    if len(html_content) > max_chars:
        html_content = html_content[:max_chars] + "\n...(截断)"
    system_prompt = """你是网页提取专家，严格按要求输出JSON；缺省值填null，禁止编造数据。"""
    user_prompt = f"{extraction_prompt}\nHTML内容：{html_content}"
    for _ in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=MODEL_NAME,
                messages=[{"role":"system","content":system_prompt},{"role":"user","content":user_prompt}],
                temperature=0.1,
                response_format={"type":"json_object"}
            )
            return json.loads(response.choices[0].message.content)
        except Exception:
            time.sleep(2)
            continue
    return None

def clean_data_with_llm(raw_data, cleaning_rules):
    """语义化数据标准化清洗"""
    prompt = f"原始数据：{json.dumps(raw_data,ensure_ascii=False)}\n清洗规则：{cleaning_rules}"
    return extract_with_llm("数据清洗", prompt)

# 通用清洗规则
cleaning_rules = """1.金额仅保留纯数字；2.时间统一为YYYY-MM-DD HH:MM:SS；3.中文量词万转为阿拉伯数字；4.无效文本统一置为null"""

def batch_crawl(urls, extract_fn, output_file="output.json"):
 """批量采集、容错存储"""
 results = []
 for idx, url in enumerate(urls):
 print(f"[{idx+1}/{len(urls)}] 采集：{url}")
 data = extract_fn(url)
 if data:results.append(data)
 time.sleep(random.uniform(1,3))
 with open(output_file, 'w', encoding='utf-8') as f:
 json.dump(results, f, ensure_ascii=False, indent=2)
 return results
def main():
 product_urls = ["https://example.com/product/1","https://example.com/product/2","https://example.com/product/3"]
 results = batch_crawl(product_urls, extract_product_info, "products.json")
 if results:
 for item in results:
 clean_data_with_llm(item, cleaning_rules)
if __name__ == '__main__':
 main()