网站开发公司取名网站虚拟服务器

网站开发公司取名,网站虚拟服务器,郴州网站建设哪里比较好,网站建设上海网站建设引言#xff1a;SEO监控的重要性与挑战在当今数字化营销时代#xff0c;搜索引擎排名直接影响着网站的流量、转化率和商业价值。关键词排名监控是SEO优化的核心环节#xff0c;它帮助企业了解自身在搜索引擎中的表现#xff0c;评估竞争对手策略#xff0c;并及时调整优化…引言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 playwright1.40.0 asyncio pandas2.0.0 sqlalchemy2.0.0 fastapi0.104.0 uvicorn0.24.0 beautifulsoup44.12.0 lxml4.9.0 pydantic2.0.0 aiohttp3.9.0 python-dotenv1.0.0 schedule1.2.0bash# 安装依赖 pip install -r requirements.txt # 安装Playwright浏览器 playwright install chromium2. 核心爬虫类实现pythonimport 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( levellogging.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_keyTrue) keyword_id Column(String(100), indexTrue) keyword Column(String(500), nullableFalse) domain Column(String(500), nullableFalse) url Column(Text, nullableFalse) rank Column(Integer, nullableFalse) search_engine Column(String(50), defaultgoogle) search_location Column(String(100), defaultus) device_type Column(String(20), defaultdesktop) search_date Column(DateTime, defaultdatetime.utcnow) page_title Column(String(1000)) created_at Column(DateTime, defaultdatetime.utcnow) def __repr__(self): return fKeywordRanking(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_factorydatetime.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(bindself.engine) # 初始化数据库 self._init_database() # 浏览器配置 self.browser_config { headless: True, args: [ --disable-blink-featuresAutomationControlled, --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}uamobile }, 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(bindself.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_agentrandom.choice(self.user_agents), viewport{width: 1920, height: 1080}, localeen-US, timezone_idAmerica/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: flocation{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_untilnetworkidle, timeout60000) # 等待搜索结果加载 await page.wait_for_selector(div.g, timeout10000) # 解析搜索结果 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(pathferror_{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-sncf1]) 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( keywordkeyword_config.keyword, search_enginekeyword_config.search_engine, devicekeyword_config.device, locationkeyword_config.location, max_resultskeyword_config.max_pages * 10 ) ranking_results [] # 分析结果找到目标域名的排名 for result in search_results: if keyword_config.domain in result[url]: ranking_result RankingResult( keywordkeyword_config.keyword, domainkeyword_config.domain, urlresult[url], rankresult[rank], page_titleresult[title], search_enginekeyword_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_idkeyword_hash, keywordranking_result.keyword, domainranking_result.domain, urlranking_result.url, rankranking_result.rank, search_engineranking_result.search_engine, page_titleranking_result.page_title, search_dateranking_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_exceptionsTrue) # 整理结果 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( keywordkeyword, domaindomain, start_datestart_dt, end_dateend_dt ) return { status: success, data: df.to_dict(orientrecords), 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(daysdays) df self.monitor.get_ranking_history( start_datestart_date, end_dateend_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( keywordPython爬虫教程, domainexample.com, search_enginegoogle, locationus, devicedesktop ), KeywordConfig( keywordSEO优化技巧, domainexample.com, search_enginebaidu, locationzh-CN, devicemobile ), KeywordConfig( keyword机器学习算法, domainexample.org, search_enginegoogle, locationuk, devicedesktop, max_pages5 ) ] # 执行批量监控 logger.info(开始批量监控关键词...) results await monitor.batch_monitor(keyword_configs, concurrent_tasks3) # 输出结果 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, indexFalse, encodingutf-8-sig) print(f\n历史数据已保存到 ranking_history.csv共 {len(history_df)} 条记录) # 启动API服务可选 # api KeywordRankingAPI(monitor) # uvicorn.run(api.app, host0.0.0.0, port8000) 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_URLsqlite:///data/rankings.db - LOG_LEVELINFO - MAX_CONCURRENT_TASKS5 restart: unless-stopped networks: - seo-network postgresql: # 可选使用PostgreSQL代替SQLite image: postgres:15 container_name: seo-db environment: - POSTGRES_DBseo_rankings - POSTGRES_USERseo_user - POSTGRES_PASSWORDsecure_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: bridge4. 定时监控任务配置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, encodingutf-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( keyworditem[keyword], domainitem[domain], search_engineitem.get(search_engine, google), locationitem.get(location, us), deviceitem.get(device, desktop), max_pagesitem.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_tasksgroup_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. 反爬虫规避策略增强pythonclass 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(): 生成浏览器指纹 # 生成随机浏览器指纹 pass2. 数据可视化分析pythonimport 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, xsearch_date, yrank, colorkeyword, title关键词排名变化趋势, labels{rank: 排名, search_date: 日期}, hover_data[url] ) fig.update_yaxes(autorangereversed) # 排名越小越好所以反转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, xdomain, yrank, title竞争对手排名分布, pointsall ) fig.update_yaxes(autorangereversed) return fig最佳实践与注意事项1. 合规性考虑遵守robots.txt协议控制请求频率避免对目标网站造成负担仅用于合法、合规的SEO监控目的2. 性能优化使用连接池和会话复用实现结果缓存机制分布式爬虫架构扩展3. 错误处理与监控实现完整的异常处理机制添加日志记录和监控报警数据备份和恢复策略结论本文详细介绍了如何利用Python最新技术构建一个完整的关键词排名监控系统。通过结合Playwright、Asyncio、FastAPI等现代技术我们创建了一个高效、可扩展、功能丰富的监控解决方案。这个系统不仅可以帮助SEO专业人员实时跟踪关键词排名变化还能为网站优化决策提供数据支持。该系统具有以下核心优势高效准确支持JavaScript渲染页面获取准确的搜索结果灵活扩展模块化设计易于添加新的搜索引擎支持智能调度支持定时任务和批量监控数据丰富提供历史数据分析和趋势可视化易于部署支持Docker容器化部署