Python实战之SEO优化自动化工具开发指南

# 网络请求和数据抓取
import requests
from bs4 import BeautifulSoup
import selenium
from selenium import webdriver

# 数据处理和分析
import pandas as pd
import numpy as np
from textstat import flesch_reading_ease

# SEO专用库
importwww.chinasem.cn advertools as adv
from googlesearch import search

# 数据可视化
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px

# 自动化和调度
import schedule
import time
from datetime import datetime

# 外链建设相关
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import json
import random

# 配置管理
import configparser
import os
from dotenv import load_dotenv

class KeywordResearcher:
    def __init__(self, api_key=None):
        self.api_key = api_key
        
    def extract_keywords_from_content(self, content, language='zh'):
        """从内容中提取关键词"""
        # 使用jieba进行中文分词
        import jieba
        import jieba.analyse
        
        keywords = jieba.analyse.extract_tags(
            content, 
            topK=20, 
            withWeight=True
        )
        return keywords
    
    def get_search_suggestions(self, seed_keyword):
        """获取搜索建议"""
        suggestions = adv.serp_goog(
            q=seed_keyword,
            cx=self.api_key,
            num=10
        )
        return suggestions
    
    def analyze_keyword_difficulty(self, keyword):
        """分析关键词竞争难度"""
        # 模拟竞争度分析逻辑
        search_results = list(search(keyword, num=10, stop=10))
        
        difficulty_score = {
            'keyword': keyword,
            'competition_level': len(search_results),
            'estimated_difficulty': 'Medium'  # 可以基于更复杂的算法
        }
        return difficulty_score

class TechnicalSEOAnalyzer:
    def __init__(self):
        self.session = requests.Session()
        
    def check_page_speed(self, url):
        """检查页面加载速度"""
        start_time = time.time()
        try:
            response = self.session.get(url, timeout=10)
            load_time = time.time() - start_time
            
            return {
                'url': url,
                'load_time': round(load_time, 2),
                'status_code': response.status_code,
                'content_size': len(response.content)
            }
        except Exception as e:
            return {'url': url, 'error': str(e)}
    
    def analyze_meta_tags(self, url):
        """分析Meta标签"""
        try:
            response = self.session.get(url)
            soup = BeautifulSoup(response.content, 'html.parser')
            
            meta_analysis = {
                'title': soup.find('title').text if soup.find('title') else None,
                'meta_description': None,
                'meta_keywords': None,
                'h1_tags': [h1.text for h1 in soup.find_all('h1')],
                'h2_tags': [h2.text for h2 in soup.find_all('h2')],
                'image_alt_missing': len([img for img in soup.find_all('img') if not img.get('alt')])
            }
            
            # 获取meta description
            meta_desc = soup.find('meta', attrs={'name': 'description'})
            if meta_desc:
                meta_analysis['meta_description'] = meta_desc.get('content')
                
            return meta_analysis
            
        except Exception as e:
            return {'url': url, 'error': str(e)}
    
    def check_internal_links(self, url, domain):
        """检查内链结构"""
        try:
            response = self.session.get(url)
            soup = BeautifulSoup(response.content, 'html.parser')
            
            all_links = soup.find_all('a', href=True)
            internal_links = [
  android              link['href'] for link in all_links 
                if domain in link['href'] or link['href'].startswith('/')
            ]
            
            return {
                'total_links': len(all_links),
                'internal_links': len(internal_links),
                'external_links': len(all_links) - len(internal_links),
                'internal_link_ratio': len(internal_links) / len(all_links) if all_links else 0
            }
            
        except Exception as e:
            return {'url': url, 'error': str(e)}

class ContentOptimizer:
    def __init__(self):
        pass
    
    def analyze_keyword_density(self, content, target_keywords):
        """分析关键词密度"""
        import re
        
        # 清理文本
        clean_content = re.sub(r'<[^>]+>', '', content.lower())
        word_count http://www.chinasem.cn= len(clean_content.split())
        
        keyword_analysis = {}
        for keyword in target_keywords:
            keyword_count = clean_content.count(keyword.lower())
            density = (keyword_count / word_count) * 100 if word_count > 0 else 0
            
            keyword_analysis[keyword] = {
                'count': keyword_count,
                'density': round(density, 2),
                'recommendation': self._get_density_recommendation(density)
            }
            
        return keyword_analysis
    
    def _get_density_recommendation(self, density):
        """获取关键词密度建议"""
        if density < 1:
            return "密度过低，建议增加关键词使用"
        elif density > 3:
            return "密度过高，可能被视为关键词堆砌"
        else:
            return "密度适中"
    
    def analyze_content_quality(self, content):
        """分析内容质量"""
        word_count = len(content.split())
        
        # 使用textstat库分析可读性
        readability_score = flesch_reading_ease(content)
        
        quality_metrics = {
            'word_count': word_count,
            'readability_score': readability_score,
            'readability_level': self._get_readability_level(readability_score),
            'recommendations': self._get_content_recommendations(word_count, readability_score)
        }
        
        return quality_metrics
    
    def _get_readability_level(self, score):
        """获取可读性等级"""
        if score >= 90:
            return "非常容易阅读"
        elif score >= 80:
            return "容易阅读"
        elif score >= 70:
            return "较容易阅读"
        elif score >= 60:
            return "标准阅读难度"
        else:
            return "较难阅读"
    
    def _get_content_recommendations(self, word_count, readability_score):
        """获取内容优化建议"""
        recommendations = []
        
        if word_count < 300:
            recommendations.append("内容长度偏短，建议增加到至少300字")
        elif word_count > 2000:
            recommendations.append("内容较长，考虑分段或分页")
            
        if readability_score < 60:
            recommendations.append("内容可读性较低，建议使用更简单的句式")
            
        return recommendations

class BacklinkBuilder:
    def __init__(self, email_config=None):
        self.email_config = email_config or {}
        self.prospects_db = []

    def find_link_opportunities(self, target_keywords, competitor_urls=None):
        """发现外链机会"""
        opportunities = []

        # 1. 基于关键词搜索相关网站
        for keyword in target_keywords:
            search_queries = [
                f"{keyword} 资源页面",
                f"{keyword} 链接",
                f"{keyword} 目录",
                f"最佳 {keyword} 网站",
                f"{keyword} 工具推荐"
            ]

            for query in search_queries:
                try:
                    search_results = list(search(query, num=10, stop=10))
                    for url in search_results:
                        opportunity = self._analyze_link_opportunity(url, keyword)
                        if opportunity['score'] > 50:  # 只保留高质量机会
                            opportunities.append(opportunity)
                except Exception as e:
                    print(f"搜索错误: {e}")

        # 2. 分析竞争对手外链
        if competitor_urls:
            for competitor_url in competitor_urls:
                competitor_backlinks = self._get_competitor_backlinks(competitor_url)
                opportunities.extend(competitor_backlinks)

        return self._deduplicate_opportunities(opportunities)

    def _analyze_link_opportunity(self, url, keyword):
        """分析单个外链机会"""
        try:
            response = requests.get(url, timeout=10)
            soup = BeautifulSoup(response.content, 'html.parser')

            # 基础信息提取
            title = soup.find('title').text if soup.find('title') else ""
            meta_desc = soup.find('meta', attrs={'name': 'description'})
            meta_desc = meta_desc.get('content') if meta_desc else ""

            # 计算相关性得分
            relevance_score = self._calculate_relevance_score(
                title + " " + meta_desc, keyword
            )

            # 检查是否有联系方式
            contact_info = self._extract_contact_info(soup)

            # 检查页面权威性指标
            authority_score = self._estimate_authority(soup, url)

            opportunity = {
                'url': url,
                'title': title,
                'keyword': keyword,
                'relevance_score': relevance_score,
                'authority_score': authority_score,
                'contact_info': contact_info,
                'score': (relevance_score + authority_score) / 2,
                'status': 'discovered',
                'discovered_date': datetime.now().isoformat()
            }

            return opportunity

        except Exception as e:
            return {
                'url': url,
                'keyword': keyword,
                'error': str(e),
                'score': 0,
                'status': 'error'
            }

    def _calculate_relevance_score(self, content, keyword):
        """计算内容相关性得分"""
        content_lower = content.lower()
        keyword_lower = keyword.lower()

        # 简单的相关性计算
        keyword_count = content_lower.count(keyword_lower)
        content_length = len(content.split())

        if content_length == 0:
            return 0

        # 基于关键词密度和出现次数计算得分
        density = (keyword_count / content_length) * 100
        base_score = min(keyword_count * 10, 50)  # 最多50分
        density_bonus = min(density * 5, 30)  # 最多30分

        return min(base_score + density_bonus, 100)

    def _extract_contact_info(self, soup):
        """提取联系信息"""
        contact_info = {
            'email': None,
            'contact_page': None,
            'social_media': []
        }

        # 查找邮箱
        import re
        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
        page_text = soup.get_text()
        emails = re.findall(email_pattern, page_text)
        if emails:
            contact_info['email'] = emails[0]

        # 查找联系页面链接
        contact_links = soup.find_all('a', href=True)
        for link in contact_links:
            href = link['href'].lower()
            text = link.get_text().lower()
            if any(word in href or word in text for word in ['contact', '联系', 'about', '关于']):
                contact_info['contact_page'] = link['href']
                break

        # 查找社交媒体链接
        social_patterns = {
            'twitter': r'twitter\.com',
            'facebook': r'facebook\.com',
            'linkedin': r'linkedin\.com',
            'weibo': r'weibo\.com'
        }

        for link in contact_links:
            href = link.get('href', '')
            for platform, pattern in social_patterns.items():
                if re.search(pattern, href):
                    contact_info['social_media'].append({
                        'platform': platform,
                        'url': href
                    })

        return contact_info

    def _estimate_authority(self, soup, url):
        """估算网站权威性"""
        authority_score = 0

        # 基于域名年龄（简化版）
        domain = url.split('/')[2]
        if len(domain.split('.')) >= 2:
            authority_score += 20

        # 基于内容质量指标
        text_content = soup.get_text()
        word_count = len(text_content.split())

        if word_count > 500:
            authority_score += 20
        if word_count > 1000:
            authority_score += 10

        # 基于页面结构
        if soup.find_all('h1'):
            authority_score += 10
        if soup.find_all('h2'):
            authority_score += 10
        if soup.find_all('img'):
            authority_score += 10

        # 基于外链数量（页面中的外链）
        external_links = len([
            link for link in soup.find_all('a', href=True)
            if 'http' in link['href'] and domain not in link['href']
        ])

        if external_links > 5:
            authority_score += 10
        if external_links > 20:
            authority_score += 10

        return min(authority_score, 100)

    def _get_competitor_backlinks(self, competitor_url):
        """获取竞争对手的外链（简化版）"""
        # 这里应该集成专业的外链分析API
        # 如Ahrefs、SEMrush等，这里提供一个模拟实现

        mock_backlinks = [
            {
                'url': 'https://example-blog.com',
                'title': '相关行业博客',
                'authority_score': 75,
                'relevance_score': 80,
                'score': 77.5,
                'source': f'competitor_analysis_{competitor_url}',
                'status': 'discovered',
                'discovered_date': datetime.now().isoformat()
            }
        ]

        return mock_backlinks

    def _deduplicate_opportunities(self, opportunities):
        """去重外链机会"""
        seen_urls = set()
        unique_opportunities = []

        for opp in opportunities:
            if opp.get('url') not in seen_urls:
                seen_urls.add(opp.get('url'))
                unique_opportunities.append(opp)

        # 按得分排序
        return sorted(unique_opportunities, key=lambda x: x.get('score', 0), reverse=True)

    def generate_outreach_email(self, opportunity, your_website, your_content_url):
        """生成外链申请邮件"""
        templates = [
            {
                'subject': f"关于{opportunity['title']}的资源推荐",
                'body': f"""
您好，

我是{your_website}的内容编辑。我刚刚阅读了您的文章"{opportunity['title']}"，内容非常有价值。

我们最近发布了一篇关于{opportunity['keyword']}的深度文章：{your_content_url}

这篇文章提供了独特的见解和实用的建议，我认为它会为您的读者带来额外的价值。如果您觉得合适，是否可以考虑在您的文章中添加这个链接？

感谢您的时间和考虑。

最好的祝愿，
[您的姓名]
                """
            },
            {
                'subject': f"为您的{opportunity['keyword']}资源页面推荐优质内容",
                'body': f"""
您好，

我在搜索{opportunity['keyword']}相关资源时发现了您的网站{opportunity['url']}。您整理的资源列表非常全面！

我想向您推荐我们最近发布的一篇文章：{your_content_url}

这篇文章深入探讨了{opportunity['keyword']}的最新趋势和最佳实践，包含了原创研究和案例分析。我相信它会是您资源列表的有价值补充。

如果您有任何问题或需要更多信息，请随时联系我。

谢谢！
[您的姓名]
                """
            }
        ]

        template = random.choice(templates)
        return {
            'to_email': opportunity['contact_info'].get('email'),
            'subject': template['subject'],
            'body': template['body'],
            'opportunity_id': opportunity.get('url'),
            'created_date': datetime.now().isoformat()
        }

    def send_outreach_email(self, email_data):
        """发送外链申请邮件"""
        if not self.email_config or not email_data.get('to_email'):
            return {'status': 'error', 'message': '邮件配置或收件人邮箱缺失'}

        try:
            msg = MIMEMultipart()
            msg['From'] = self.email_config['from_email']
            msg['To'] = email_data['to_email']
            msg['Subject'] = email_data['subject']

            msg.attach(MIMEText(email_data['body'], 'plain', 'utf-8'))

            server = smtplib.SMTP(self.email_config['smtp_server'], self.email_config['smtp_port'])
            server.starttls()
            server.login(self.email_config['username'], self.email_config['password'])

            text = msg.as_string()
            server.sendmail(self.email_config['from_email'], email_data['to_email'], text)
            server.quit()

            return {
                'status': 'sent',
                'message': '邮件发送成功',
                'sent_date': datetime.now().isoformat()
            }

        except Exception as e:
            return {
                'status': 'error',
                'message': f'邮件发送失败: {str(e)}'
            }

    def track_backlink_status(self, target_url, backlink_urls):
        """监控外链状态"""
        backlink_status = []

        for backlink_url in backlink_urls:
            try:
                response = requests.get(backlink_url, timeout=10)
                soup = BeautifulSoup(response.content, 'html.parser')

                # 检查是否包含目标链接
                links = soup.find_all('a', href=True)
                has_backlink = any(target_url in link['href'] for link in links)

                status = {
                    'backlink_url': backlink_url,
                    'target_url': target_url,
                    'has_backlink': has_backlink,
                    'checked_date': datetime.now().isoformat(),
                    'status_code': response.status_code
                }

                backlink_status.append(status)

            except Exception as e:
                backlink_status.append({
                    'backlink_url': backlink_url,
                    'target_url': target_url,
                    'error': str(e),
                    'checked_date': datetime.now().isoformat()
                })

        return backlink_status

    def save_prospects_to_file(self, opportunities, filename='backlink_prospects.json'):
        """保存外链机会到文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(opportunities, f, ensure_ascii=False, indent=2)

        return filename

    def load_prospects_from_file(self, filename='backlink_prospects.json'):
        """从文件加载外链机会"""
        try:
            with open(filename, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            return []

class SEOReportGenerator:
    def __init__(self, output_dir='reports'):
        selphpf.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        
    def generate_comprehensive_report(self, analysis_data):
        """生成综合SEO报告"""
        report_date = datetime.now().strftime('%Y-%m-%d')
        
        # 创建HTML报告
        html_content = self._create_html_report(analysis_data, report_date)
        
        # 保存报告
        report_path = os.path.join(self.output_dir, f'seo_report_{report_date}.html')
        with open(report_path, 'w', encoding='utf-8') as f:
            f.write(html_content)
            
        return report_path
    
    def _create_html_report(self, data, date):
        """创建HTML格式报告"""
        html_template = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>SEO分析报告 - {date}</title>
            <meta charset="utf-8">
            <style>
                body {{ font-family: Arial, sans-serif; margin: 40px; }}
                .header {{ background-color: #f4f4f4; padding: 20px; }}
                .section {{ margin: 20px 0; }}
                .metric {{ background-color: #e9e9e9; padding: 10px; margin: 5px 0; }}
                .recommendation {{ background-color: #fff3cd; padding: 10px; margin: 5px 0; }}
            </style>
        </head>
        <body>
            <div class="header">
                <h1>SEO自动化分析报告</h1>
                <p>生成日期: {date}</p>
            </div>
            
            <div class="section">
                <h2>技术SEO检测结果</h2>
                {self._format_technical_seo_data(data.get('technical_seo', {}))}
            </div>
            
            <div class="section">
                <h2>内容优化建议</h2>
                {self._format_content_optimization_data(data.get('content_optimization', {}))}
            </div>
            
            <div class="section">
                <h2>关键词分析</h2>
                {self._format_keyword_data(data.get('keyword_analysis', {}))}
            </div>
        </body>
        </html>
        """
        return html_template
    
    def _format_technical_seo_data(self, data):
        """格式化技术SEO数据"""
        if not data:
            return "<p>暂无技术SEO数据</p>"
            
        html = ""
        for url, metrics in data.items():
            html += f"""
            <div class="metric">
       SYdctdI         <h3>{url}</h3>
                <p>加载时间: {metrics.get('load_time', 'N/A')}秒</p>
                <p>状态码: {metrics.get('status_code', 'N/A')}</p>
                <p>内容大小: {metrics.get('content_size', 'N/A')} bytes</p>
            </div>
            """
        return html
    
    def _format_content_optimization_data(self, data):
        """格式化内容优化数据"""
        if not data:
            return "<p>暂无内容优化数据</p>"
            
        html = ""
        for page, analysis in data.items():
            html += f"""
            <div class="metric">
                <h3>{page}</h3>
                <p>字数: {analysis.get('word_count', 'N/A')}</p>
                <p>可读性评分: {analysis.get('readability_score', 'N/A')}</p>
                <p>可读性等级: {analysis.get('readability_level', 'N/A')}</p>
            </div>
            """
            
            recommendations = analysis.get('recommendations', [])
            if recommendations:
                html += '<div class="recommendation"><h4>优化建议:</h4><ul>'
                for rec in recommendations:
                    html += f'<li>{rec}</li>'
                html += '</ul></div>'
                
        return html
    
    def _format_keyword_data(self, data):
        """格式化关键词数据"""
        if not data:
            return "<p>暂无关键词数据</p>"
            
        html = ""
        for keyword, metrics in data.items():
            html += f"""
            <div class="metric">
                <h3>{keyword}</h3>
                <p>出现次数: {metrics.get('count', 'N/A')}</p>
                <p>密度: {metrics.get('density', 'N/A')}%</p>
                <p>建议: {metrics.get('recommendation', 'N/A')}</p>
            </div>
            """
        return html

def main():
    # 初始化各个模块
    keyword_researcher = KeywordResearcher()
    technical_analyzer = TechnicalSEOAnalyzer()
    content_optimizer = ContentOptimizer()

    # 邮件配置（用于外链建设）
    email_config = {
        'from_email': 'your-email@example.com',
        'smtp_server': 'smtp.gmail.com',
        'smtp_port': 587,
        'username': 'your-email@example.com',
        'password': 'your-app-password'
    }
    backlink_builder = BacklinkBuilder(email_config)
    report_generator = SEOReportGenerator()
    
    # 目标网站和关键词
    target_url = "https://example.com"
    target_keywords = ["SEO优化", "搜索引擎优化", "网站优化"]
    
    # 执行分析
    analysis_results = {}
    
    # 1. 技术SEO检测
    print("正在进行技术SEO检测...")
    technical_results = technical_analyzer.check_page_speed(target_url)
    meta_results = technical_analyzer.analyze_meta_tags(target_url)
    analysis_results['technical_seo'] = {
        target_url: {**technical_results, **meta_results}
    }
    
    # 2. 内容优化分析
    print("正在进行内容优化分析...")
    # 这里需要获取页面内容
    response = requests.get(target_url)
    content = response.text
    
    keyword_density = content_optimizer.analyze_keyword_density(content, target_keywords)
    content_quality = content_optimizer.analyze_content_quality(content)
    
    analysis_results['content_optimization'] = {
        target_url: {**content_quality}
    }
    analysis_results['keyword_analysis'] = keyword_density

    # 3. 外链建设分析
    print("正在进行外链机会发现...")
    competitor_urls = ["https://competitor1.com", "https://competitor2.com"]
    link_opportunities = backlink_builder.find_link_opportunities(
        target_keywords,
        competitor_urls
    )

    # 保存外链机会
    prospects_file = backlink_builder.save_prospects_to_file(link_opportunities)
    print(f"发现 {len(link_opportunities)} 个外链机会，已保存到 {prospects_file}")

    # 生成外链申请邮件（示例）
    if link_opportunities:
        sample_opportunity = link_opportunities[0]
        if sample_opportunity.get('contact_info', {}).get('email'):
            email_content = backlink_builder.generate_outreach_email(
                sample_opportunity,
                target_url,
                f"{target_url}/your-content-page"
            )
            print("示例外链申请邮件已生成")

    analysis_results['backlink_opportunities'] = {
        'total_found': len(link_opportunities),
        'high_quality': len([opp for opp in link_opportunities if opp.get('score', 0) > 75]),
        'with_contact_info': len([opp for opp in link_opportunities if opp.get('contact_info', {}).get('email')])
    }

    # 4. 生成报告
    print("正在生成报告...")
    report_path = report_generator.generate_comprehensive_report(analysis_results)
    print(f"报告已生成: {report_path}")

if __name__ == "__main__":
    main()

def schedule_seo_analysis():
    """设置定期SEO分析任务"""
    
    # 每天早上9点执行
    schedule.every().day.at("09:00").do(main)
    
    # 每周一执行完整分析
    schedule.every().monday.at("10:00").do(comprehensive_analysis)
    
    print("SEO自动化任务已启动...")
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次

def comprehensive_analysis():
    """执行全面的SEO分析"""
    # 包含更多深度分析的逻辑
    pass

[DEFAULT]
target_urls = https://example1.com,https://example2.com
target_keywords = SEO优化,搜索引擎优化,网站优化

[API_KEYS]
google_api_key = your_google_api_key
google_cx = your_custom_search_engine_id

[SETTINGS]
report_output_dir = reports
analysis_frequency = daily
email_notifications = true

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "main.py"]