Python爬虫实战：基于最新技术的智能关键词排名监控系统-洪萨配资

引言：SEO监控的重要性与挑战

在当今数字化营销时代，搜索引擎排名直接影响着网站的流量、转化率和商业价值。关键词排名监控是SEO优化的核心环节，它帮助企业了解自身在搜索引擎中的表现，评估竞争对手策略，并及时调整优化方案。然而，传统的手动查询方式效率低下，而市面上的商业监控工具往往价格昂贵且定制性差。本文将介绍如何利用Python爬虫技术构建一个高效、智能的关键词排名监控系统。

技术栈亮点：为什么选择这些最新技术？

Playwright- 微软开源的现代化浏览器自动化框架，支持无头浏览器操作，完美处理JavaScript渲染页面
Asyncio- Python原生异步I/O框架，实现高并发监控
Pandas & SQLAlchemy- 数据处理与存储
FastAPI- 构建监控API接口
Docker- 容器化部署，确保环境一致性

系统架构设计

text

关键词排名监控系统架构： 1. 数据采集层：Playwright异步爬虫 2. 数据处理层：Pandas数据清洗与分析 3. 数据存储层：PostgreSQL/SQLite数据库 4. API服务层：FastAPI RESTful接口 5. 可视化层：Dash/Streamlit仪表板（可选）

完整代码实现

1. 环境配置与依赖安装

python

# requirements.txt playwright>=1.40.0 asyncio pandas>=2.0.0 sqlalchemy>=2.0.0 fastapi>=0.104.0 uvicorn>=0.24.0 beautifulsoup4>=4.12.0 lxml>=4.9.0 pydantic>=2.0.0 aiohttp>=3.9.0 python-dotenv>=1.0.0 schedule>=1.2.0

bash

# 安装依赖 pip install -r requirements.txt # 安装Playwright浏览器 playwright install chromium

2. 核心爬虫类实现

python

import asyncio import json from datetime import datetime from typing import List, Dict, Optional, Tuple import pandas as pd from pydantic import BaseModel, Field from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from playwright.async_api import async_playwright, Browser, Page import aiohttp from urllib.parse import quote_plus import logging from contextlib import asynccontextmanager import hashlib # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # 数据模型定义 Base = declarative_base() class KeywordRanking(Base): """关键词排名数据表""" __tablename__ = 'keyword_rankings' id = Column(Integer, primary_key=True) keyword_id = Column(String(100), index=True) keyword = Column(String(500), nullable=False) domain = Column(String(500), nullable=False) url = Column(Text, nullable=False) rank = Column(Integer, nullable=False) search_engine = Column(String(50), default='google') search_location = Column(String(100), default='us') device_type = Column(String(20), default='desktop') search_date = Column(DateTime, default=datetime.utcnow) page_title = Column(String(1000)) created_at = Column(DateTime, default=datetime.utcnow) def __repr__(self): return f"<KeywordRanking(keyword='{self.keyword}', rank={self.rank})>" # Pydantic模型 class KeywordConfig(BaseModel): """关键词配置模型""" keyword: str domain: str search_engine: str = "google" location: str = "us" device: str = "desktop" max_pages: int = 10 check_frequency: str = "daily" class RankingResult(BaseModel): """排名结果模型""" keyword: str domain: str url: str rank: int page_title: str search_engine: str search_date: datetime = Field(default_factory=datetime.utcnow) class AdvancedRankingMonitor: """高级关键词排名监控器""" def __init__(self, db_url: str = "sqlite:///rankings.db"): """ 初始化监控器 Args: db_url: 数据库连接URL """ self.db_url = db_url self.engine = create_engine(db_url) self.SessionLocal = sessionmaker(bind=self.engine) # 初始化数据库 self._init_database() # 浏览器配置 self.browser_config = { 'headless': True, 'args': [ '--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox', ] } # 搜索引擎配置 self.search_engines = { 'google': { 'desktop': 'https://www.google.com/search?q={query}&hl={lang}&num={num}', 'mobile': 'https://www.google.com/search?q={query}&hl={lang}&num={num}&ua=mobile' }, 'bing': { 'desktop': 'https://www.bing.com/search?q={query}&count={num}', 'mobile': 'https://www.bing.com/search?q={query}&count={num}' }, 'baidu': { 'desktop': 'https://www.baidu.com/s?wd={query}&rn={num}', 'mobile': 'https://m.baidu.com/s?wd={query}&rn={num}' } } # 用户代理池 self.user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' ] def _init_database(self): """初始化数据库表""" Base.metadata.create_all(bind=self.engine) logger.info("数据库初始化完成") @asynccontextmanager async def get_browser_context(self): """获取浏览器上下文管理器""" playwright = await async_playwright().start() browser = await playwright.chromium.launch(**self.browser_config) try: # 创建上下文，添加随机用户代理 import random context = await browser.new_context( user_agent=random.choice(self.user_agents), viewport={'width': 1920, 'height': 1080}, locale='en-US', timezone_id='America/New_York' ) yield context finally: await browser.close() await playwright.stop() async def search_keyword( self, keyword: str, search_engine: str = "google", device: str = "desktop", location: str = "us", max_results: int = 100 ) -> List[Dict]: """ 搜索关键词并获取结果 Args: keyword: 搜索关键词 search_engine: 搜索引擎 device: 设备类型 location: 搜索位置 max_results: 最大结果数 Returns: 搜索结果列表 """ logger.info(f"开始搜索关键词: {keyword} ({search_engine}, {device})") # 构建搜索URL if search_engine not in self.search_engines: raise ValueError(f"不支持的搜索引擎: {search_engine}") search_url = self.search_engines[search_engine][device] search_params = { 'query': quote_plus(keyword), 'lang': location, 'num': min(max_results, 100) } # 设置位置Cookie（针对Google） cookies = [] if search_engine == 'google': cookies.append({ 'name': 'NID', 'value': f'location={location}', 'domain': '.google.com', 'path': '/' }) results = [] async with self.get_browser_context() as context: # 添加Cookies if cookies: await context.add_cookies(cookies) page = await context.new_page() try: # 导航到搜索页面 url = search_url.format(**search_params) await page.goto(url, wait_until='networkidle', timeout=60000) # 等待搜索结果加载 await page.wait_for_selector('div.g', timeout=10000) # 解析搜索结果 if search_engine == 'google': results = await self._parse_google_results(page, max_results) elif search_engine == 'bing': results = await self._parse_bing_results(page, max_results) elif search_engine == 'baidu': results = await self._parse_baidu_results(page, max_results) logger.info(f"关键词 '{keyword}' 找到 {len(results)} 个结果") except Exception as e: logger.error(f"搜索关键词 '{keyword}' 时出错: {str(e)}") # 截图以便调试 await page.screenshot(path=f"error_{keyword.replace(' ', '_')}.png") finally: await page.close() return results async def _parse_google_results(self, page: Page, max_results: int) -> List[Dict]: """解析Google搜索结果""" results = [] # 获取所有搜索结果 search_items = await page.query_selector_all('div.g') for rank, item in enumerate(search_items[:max_results], 1): try: # 获取标题 title_element = await item.query_selector('h3') title = await title_element.inner_text() if title_element else "无标题" # 获取链接 link_element = await item.query_selector('a') url = await link_element.get_attribute('href') if link_element else "" # 获取描述 desc_element = await item.query_selector('div[data-sncf="1"]') description = await desc_element.inner_text() if desc_element else "" if url and url.startswith('http'): results.append({ 'rank': rank, 'title': title.strip(), 'url': url, 'description': description.strip()[:200] }) except Exception as e: logger.debug(f"解析搜索结果时出错: {str(e)}") continue return results async def _parse_bing_results(self, page: Page, max_results: int) -> List[Dict]: """解析Bing搜索结果""" results = [] search_items = await page.query_selector_all('li.b_algo') for rank, item in enumerate(search_items[:max_results], 1): try: title_element = await item.query_selector('h2 a') title = await title_element.inner_text() if title_element else "无标题" url = await title_element.get_attribute('href') if title_element else "" desc_element = await item.query_selector('div.b_caption p') description = await desc_element.inner_text() if desc_element else "" if url: results.append({ 'rank': rank, 'title': title.strip(), 'url': url, 'description': description.strip()[:200] }) except Exception as e: logger.debug(f"解析Bing结果时出错: {str(e)}") continue return results async def _parse_baidu_results(self, page: Page, max_results: int) -> List[Dict]: """解析百度搜索结果""" results = [] search_items = await page.query_selector_all('div.result') for rank, item in enumerate(search_items[:max_results], 1): try: title_element = await item.query_selector('h3 a') title = await title_element.inner_text() if title_element else "无标题" url = await title_element.get_attribute('href') if title_element else "" desc_element = await item.query_selector('div.c-abstract') description = await desc_element.inner_text() if desc_element else "" if url: results.append({ 'rank': rank, 'title': title.strip(), 'url': url, 'description': description.strip()[:200] }) except Exception as e: logger.debug(f"解析百度结果时出错: {str(e)}") continue return results async def check_ranking( self, keyword_config: KeywordConfig ) -> List[RankingResult]: """ 检查指定关键词的排名 Args: keyword_config: 关键词配置 Returns: 排名结果列表 """ # 执行搜索 search_results = await self.search_keyword( keyword=keyword_config.keyword, search_engine=keyword_config.search_engine, device=keyword_config.device, location=keyword_config.location, max_results=keyword_config.max_pages * 10 ) ranking_results = [] # 分析结果，找到目标域名的排名 for result in search_results: if keyword_config.domain in result['url']: ranking_result = RankingResult( keyword=keyword_config.keyword, domain=keyword_config.domain, url=result['url'], rank=result['rank'], page_title=result['title'], search_engine=keyword_config.search_engine ) ranking_results.append(ranking_result) # 保存到数据库 self.save_ranking_to_db(ranking_result) return ranking_results def save_ranking_to_db(self, ranking_result: RankingResult): """保存排名结果到数据库""" session = self.SessionLocal() try: # 生成唯一ID keyword_hash = hashlib.md5( f"{ranking_result.keyword}_{ranking_result.domain}".encode() ).hexdigest() ranking_record = KeywordRanking( keyword_id=keyword_hash, keyword=ranking_result.keyword, domain=ranking_result.domain, url=ranking_result.url, rank=ranking_result.rank, search_engine=ranking_result.search_engine, page_title=ranking_result.page_title, search_date=ranking_result.search_date ) session.add(ranking_record) session.commit() logger.info(f"已保存排名结果: {ranking_result.keyword} - 排名 {ranking_result.rank}") except Exception as e: session.rollback() logger.error(f"保存排名结果时出错: {str(e)}") finally: session.close() async def batch_monitor( self, keyword_configs: List[KeywordConfig], concurrent_tasks: int = 5 ) -> Dict[str, List[RankingResult]]: """ 批量监控多个关键词 Args: keyword_configs: 关键词配置列表 concurrent_tasks: 并发任务数 Returns: 监控结果字典 """ from concurrent.futures import ThreadPoolExecutor results = {} # 使用信号量控制并发数 semaphore = asyncio.Semaphore(concurrent_tasks) async def monitor_with_semaphore(config): async with semaphore: return await self.check_ranking(config) # 创建监控任务 tasks = [monitor_with_semaphore(config) for config in keyword_configs] # 并发执行所有任务 monitor_results = await asyncio.gather(*tasks, return_exceptions=True) # 整理结果 for config, result in zip(keyword_configs, monitor_results): if isinstance(result, Exception): logger.error(f"监控关键词 '{config.keyword}' 时出错: {str(result)}") results[config.keyword] = [] else: results[config.keyword] = result return results def get_ranking_history( self, keyword: str = None, domain: str = None, start_date: datetime = None, end_date: datetime = None ) -> pd.DataFrame: """ 获取排名历史数据 Args: keyword: 关键词筛选 domain: 域名筛选 start_date: 开始日期 end_date: 结束日期 Returns: DataFrame格式的历史数据 """ session = self.SessionLocal() try: query = session.query(KeywordRanking) if keyword: query = query.filter(KeywordRanking.keyword == keyword) if domain: query = query.filter(KeywordRanking.domain == domain) if start_date: query = query.filter(KeywordRanking.search_date >= start_date) if end_date: query = query.filter(KeywordRanking.search_date <= end_date) query = query.order_by(KeywordRanking.search_date.desc()) results = query.all() # 转换为DataFrame data = [{ 'keyword': r.keyword, 'domain': r.domain, 'rank': r.rank, 'url': r.url, 'search_date': r.search_date, 'search_engine': r.search_engine, 'device_type': r.device_type } for r in results] df = pd.DataFrame(data) return df finally: session.close() class KeywordRankingAPI: """关键词排名监控API""" def __init__(self, monitor: AdvancedRankingMonitor): self.monitor = monitor self.app = FastAPI(title="关键词排名监控API") self.setup_routes() def setup_routes(self): """设置API路由""" @self.app.post("/monitor/single/") async def monitor_single(keyword_config: KeywordConfig): """监控单个关键词""" results = await self.monitor.check_ranking(keyword_config) return { "status": "success", "keyword": keyword_config.keyword, "results": [result.dict() for result in results] } @self.app.post("/monitor/batch/") async def monitor_batch(keyword_configs: List[KeywordConfig]): """批量监控关键词""" results = await self.monitor.batch_monitor(keyword_configs) return { "status": "success", "total_keywords": len(keyword_configs), "results": { keyword: [r.dict() for r in result] for keyword, result in results.items() } } @self.app.get("/history/") async def get_history( keyword: Optional[str] = None, domain: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None ): """获取排名历史""" # 转换日期参数 start_dt = datetime.fromisoformat(start_date) if start_date else None end_dt = datetime.fromisoformat(end_date) if end_date else None df = self.monitor.get_ranking_history( keyword=keyword, domain=domain, start_date=start_dt, end_date=end_dt ) return { "status": "success", "data": df.to_dict(orient='records'), "count": len(df) } @self.app.get("/keywords/summary/") async def get_keywords_summary(days: int = 30): """获取关键词摘要统计""" end_date = datetime.utcnow() start_date = end_date - timedelta(days=days) df = self.monitor.get_ranking_history( start_date=start_date, end_date=end_date ) if df.empty: return {"status": "success", "summary": {}} # 计算统计信息 summary = { "total_keywords": df['keyword'].nunique(), "total_domains": df['domain'].nunique(), "average_rank": df['rank'].mean(), "top_keywords": df.groupby('keyword')['rank'].mean().sort_values().head(10).to_dict(), "rank_trend": df.groupby(df['search_date'].dt.date)['rank'].mean().to_dict() } return { "status": "success", "summary": summary } async def main(): """主函数示例""" # 初始化监控器 monitor = AdvancedRankingMonitor("sqlite:///seo_rankings.db") # 定义要监控的关键词 keyword_configs = [ KeywordConfig( keyword="Python爬虫教程", domain="example.com", search_engine="google", location="us", device="desktop" ), KeywordConfig( keyword="SEO优化技巧", domain="example.com", search_engine="baidu", location="zh-CN", device="mobile" ), KeywordConfig( keyword="机器学习算法", domain="example.org", search_engine="google", location="uk", device="desktop", max_pages=5 ) ] # 执行批量监控 logger.info("开始批量监控关键词...") results = await monitor.batch_monitor(keyword_configs, concurrent_tasks=3) # 输出结果 for keyword, rankings in results.items(): if rankings: print(f"\n关键词: {keyword}") for rank in rankings: print(f" 排名 {rank.rank}: {rank.url}") else: print(f"\n关键词: {keyword} - 未找到排名") # 获取历史数据并保存为CSV history_df = monitor.get_ranking_history() if not history_df.empty: history_df.to_csv('ranking_history.csv', index=False, encoding='utf-8-sig') print(f"\n历史数据已保存到 ranking_history.csv，共 {len(history_df)} 条记录") # 启动API服务（可选） # api = KeywordRankingAPI(monitor) # uvicorn.run(api.app, host="0.0.0.0", port=8000) if __name__ == "__main__": # 运行异步主函数 asyncio.run(main())

3. Docker部署配置

dockerfile

# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ wget \ gnupg \ unzip \ && rm -rf /var/lib/apt/lists/* # 安装Chrome依赖 RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - \ && echo "deb http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list \ && apt-get update \ && apt-get install -y google-chrome-stable fonts-ipafont-gothic fonts-wqy-zenhei fonts-thai-tlwg fonts-kacst fonts-freefont-ttf \ && rm -rf /var/lib/apt/lists/* # 复制项目文件 COPY requirements.txt . COPY . . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt \ && playwright install chromium \ && playwright install-deps # 创建非root用户 RUN useradd -m -u 1000 seomonitor USER seomonitor # 启动命令 CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

yaml

# docker-compose.yml version: '3.8' services: seo-monitor: build: . container_name: seo-rank-monitor ports: - "8000:8000" volumes: - ./data:/app/data - ./logs:/app/logs environment: - DATABASE_URL=sqlite:///data/rankings.db - LOG_LEVEL=INFO - MAX_CONCURRENT_TASKS=5 restart: unless-stopped networks: - seo-network postgresql: # 可选：使用PostgreSQL代替SQLite image: postgres:15 container_name: seo-db environment: - POSTGRES_DB=seo_rankings - POSTGRES_USER=seo_user - POSTGRES_PASSWORD=secure_password volumes: - postgres-data:/var/lib/postgresql/data ports: - "5432:5432" networks: - seo-network redis: # 可选：用于缓存和任务队列 image: redis:7-alpine container_name: seo-redis ports: - "6379:6379" networks: - seo-network volumes: postgres-data: networks: seo-network: driver: bridge

4. 定时监控任务配置

python

# scheduler.py import schedule import time import asyncio from datetime import datetime from typing import List import yaml class RankingMonitorScheduler: """排名监控调度器""" def __init__(self, monitor: AdvancedRankingMonitor, config_file: str = "monitor_config.yaml"): self.monitor = monitor self.config_file = config_file self.keyword_groups = self.load_config() def load_config(self) -> dict: """加载监控配置""" try: with open(self.config_file, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config.get('keyword_groups', {}) except FileNotFoundError: logger.warning(f"配置文件 {self.config_file} 不存在，使用默认配置") return {} async def run_monitoring_group(self, group_name: str): """运行监控组""" if group_name not in self.keyword_groups: logger.error(f"监控组 '{group_name}' 不存在") return group_config = self.keyword_groups[group_name] keyword_configs = [] for item in group_config.get('keywords', []): config = KeywordConfig( keyword=item['keyword'], domain=item['domain'], search_engine=item.get('search_engine', 'google'), location=item.get('location', 'us'), device=item.get('device', 'desktop'), max_pages=item.get('max_pages', 10) ) keyword_configs.append(config) logger.info(f"开始监控组 '{group_name}'，共 {len(keyword_configs)} 个关键词") results = await self.monitor.batch_monitor( keyword_configs, concurrent_tasks=group_config.get('concurrent_tasks', 3) ) # 发送通知（可选） self.send_notifications(group_name, results) logger.info(f"监控组 '{group_name}' 完成") def send_notifications(self, group_name: str, results: dict): """发送监控结果通知""" # 这里可以集成邮件、Slack、钉钉等通知方式 pass def setup_schedule(self): """设置定时任务""" # 每日监控 schedule.every().day.at("02:00").do( lambda: asyncio.create_task(self.run_monitoring_group("daily")) ) # 每周监控 schedule.every().monday.at("03:00").do( lambda: asyncio.create_task(self.run_monitoring_group("weekly")) ) # 每月监控 schedule.every().day.at("04:00").do( lambda: asyncio.create_task(self.run_monitoring_group("monthly")) ).tag("first_of_month") logger.info("定时任务设置完成") def run(self): """运行调度器""" self.setup_schedule() logger.info("监控调度器启动...") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务 # 配置示例文件 monitor_config.yaml """ keyword_groups: daily: frequency: "daily" concurrent_tasks: 5 keywords: - keyword: "Python编程" domain: "python.org" search_engine: "google" location: "us" device: "desktop" - keyword: "人工智能" domain: "openai.com" search_engine: "bing" location: "global" device: "desktop" weekly: frequency: "weekly" concurrent_tasks: 3 keywords: - keyword: "机器学习" domain: "scikit-learn.org" search_engine: "google" location: "us" device: "desktop" max_pages: 5 monthly: frequency: "monthly" concurrent_tasks: 2 keywords: - keyword: "深度学习框架" domain: "pytorch.org" search_engine: "google" location: "us" device: "desktop" max_pages: 3 """

高级功能扩展

1. 反爬虫规避策略增强

python

class AdvancedAntiDetection: """高级反检测策略""" @staticmethod async def mimic_human_behavior(page: Page): """模拟人类行为""" # 随机移动鼠标 await page.mouse.move( random.randint(0, 500), random.randint(0, 500) ) # 随机滚动页面 await page.evaluate(f""" window.scrollTo({{ top: {random.randint(0, 1000)}, behavior: 'smooth' }}); """) # 随机延迟 await asyncio.sleep(random.uniform(1.0, 3.0)) @staticmethod def rotate_proxies(): """代理IP轮换""" # 实现代理池轮换逻辑 pass @staticmethod def generate_fingerprint(): """生成浏览器指纹""" # 生成随机浏览器指纹 pass

2. 数据可视化分析

python

import plotly.graph_objects as go import plotly.express as px from plotly.subplots import make_subplots class RankingVisualizer: """排名数据可视化""" @staticmethod def create_ranking_timeline(df: pd.DataFrame): """创建排名时间线图""" fig = px.line( df, x='search_date', y='rank', color='keyword', title='关键词排名变化趋势', labels={'rank': '排名', 'search_date': '日期'}, hover_data=['url'] ) fig.update_yaxes(autorange="reversed") # 排名越小越好，所以反转Y轴 return fig @staticmethod def create_competitor_analysis(df: pd.DataFrame): """创建竞争对手分析图""" # 分析竞争对手的排名分布 competitor_df = df[df['domain'] != df['domain'].iloc[0]] # 排除自己的域名 fig = px.box( competitor_df, x='domain', y='rank', title='竞争对手排名分布', points='all' ) fig.update_yaxes(autorange="reversed") return fig