SERP抓取做排名追踪：requests加BeautifulSoup够不够用

import requests
from requests.auth import HTTPProxyAuth

# 亿牛云代理配置
proxy_host = "proxy.16yun.cn"
proxy_port = "端口"
proxy_user = "用户名"
proxy_pass = "密码"

proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"

proxies = {
    "http": proxy_url,
    "https": proxy_url,
}

# 带代理请求百度SERP
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}

resp = requests.get(
    "https://www.baidu.com/s",
    params={"wd": "Python爬虫教程"},
    headers=headers,
    proxies=proxies,
    timeout=15,
)

import sqlite3

def init_db(db_path="serp_tracker.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS keywords (
            id INTEGER PRIMARY KEY,
            keyword TEXT NOT NULL,
            domain TEXT NOT NULL,
            engine TEXT DEFAULT 'baidu',
            region TEXT DEFAULT '',
            last_rank INTEGER,
            last_checked TEXT
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rank_history (
            id INTEGER PRIMARY KEY,
            keyword_id INTEGER,
            rank INTEGER,
            url TEXT,
            checked_at TEXT,
            FOREIGN KEY (keyword_id) REFERENCES keywords(id)
        )
    """)
    conn.commit()
    return conn

from urllib.parse import urlparse

def normalize_url(url):
    parsed = urlparse(url)
    host = parsed.netloc.lower()
    if host.startswith("www."):
        host = host[4:]
    return host

def find_rank(results, target_domain, max_results=100):
    target = normalize_url(target_domain)
    for rank, item in enumerate(results[:max_results], 1):
        real_url = resolve_real_url(item["url"])
        if not real_url:
            continue
        if normalize_url(real_url) == target:
            return rank, real_url
    return None, None

import sqlite3
 return None
 def parse_results(self, html, engine="baidu"):
 soup = BeautifulSoup(html, "html.parser")
 results = []
 if engine == "baidu":
 for item in soup.select("div.result, div.c-container"):
 a = item.select_one("h3 a") or item.select_one("a")
 if a:
 results.append({"url": a.get("href", ""), "title": a.get_text(strip=True)})
 else:
 for item in soup.select("div.g"):
 a = item.select_one("a")
 if a:
 results.append({"url": a.get("href", ""), "title": a.get_text(strip=True)})
 return results
 def resolve_url(self, redirect_url, engine="baidu"):
 if engine == "google":
 return redirect_url
 try:
 resp = requests.get(redirect_url, allow_redirects=False, timeout=10, proxies={"http": self.proxy, "https": self.proxy} if self.proxy else None)
 return resp.headers.get("Location", redirect_url)
 except:
 return redirect_url
 def track(self):
 rows = self.conn.execute("SELECT id, keyword, domain, engine FROM keywords").fetchall()
 for kw_id, keyword, domain, engine in rows:
 html = self.fetch_serp(keyword, engine)
 if not html:
 continue
 results = self.parse_results(html, engine)
 target = domain.replace("www.", "").replace("https://", "").replace("http://", "").rstrip("/")
 found_rank = None
 found_url = None
 for rank, item in enumerate(results, 1):
 real_url = self.resolve_url(item["url"], engine)
 url_host = urlparse(real_url).netloc.lower().replace("www.", "")
 if url_host == target:
 found_rank = rank
 found_url = real_url
 break
 now = datetime.now().isoformat()
 self.conn.execute(
 "INSERT INTO ranks (keyword_id, rank, url, checked_at) VALUES (?, ?, ?, ?)",
 (kw_id, found_rank, found_url, now)
 )
 self.conn.commit()
 status = f"第{found_rank}名" if found_rank else "未进入前50"
 print(f"[{keyword}] {status}")
 time.sleep(random.uniform(5, 12))
 def get_history(self, keyword_id):
 return self.conn.execute(
 "SELECT rank, checked_at FROM ranks WHERE keyword_id = ? ORDER BY checked_at",
 (keyword_id,)
 ).fetchall()
if __name__ == "__main__":
 tracker = RankTracker()
 tracker.add_keyword("Python教程", "docs.python.org")
 tracker.add_keyword("爬虫框架", "scrapy.org")
 tracker.track()