GEO 生成引擎优化：AI 搜索时代的内容战略与反欺骗技术

# ===== GEO 内容语义覆盖分析与优化 =====
import openai
import numpy as np
from typing import List, Dict

class GEOOptimizer:
    """生成引擎优化分析器：评估和优化内容的 AI 可引用性"""
    
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        # GEO 优化权重配置
        self.weights = {
            'semantic_coverage': 0.25,    # 语义覆盖度
            'structure_quality': 0.20,    # 结构质量
            'authority_signals': 0.20,    # 权威性信号
            'data_density': 0.15,         # 数据密度
            'freshness': 0.10,            # 时效性
            'multimodal': 0.10,           # 多模态支持
        }
    
    def analyze_content(self, content: str, target_topic: str) -> Dict:
        """分析内容的 GEO 分数"""
        # 1. 语义覆盖分析
        semantic_score = self._analyze_semantic_coverage(content, target_topic)
        # 2. 结构质量分析
        structure_score = self._analyze_structure(content)
        # 3. 权威性信号分析
        authority_score = self._analyze_authority(content)
        # 4. 数据密度分析
        data_score = self._analyze_data_density(content)
        
        # 加权总分
        total_score = (
            semantic_score * self.weights['semantic_coverage'] +
            structure_score * self.weights['structure_quality'] +
            authority_score * self.weights['authority_signals'] +
            data_score * self.weights['data_density']
        )
        
        return {
            'total_score': round(total_score, 2),
            'semantic_coverage': round(semantic_score, 2),
            'structure_quality': round(structure_score, 2),
            'authority_signals': round(authority_score, 2),
            'data_density': round(data_score, 2),
            'recommendations': self._generate_recommendations(
                semantic_score, structure_score, authority_score, data_score
            )
        }
    
    def _analyze_semantic_coverage(self, content: str, topic: str) -> float:
        """分析内容对目标主题的语义覆盖度"""
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "system",
                "content": "评估文章对目标主题的语义覆盖度（0-100）。考虑：概念完整性、子领域覆盖、术语丰富度。"
            }, {
                "role": "user",
                "content": f"主题: {topic}\n\n文章:\n{content[:3000]}"
            }]
        )
        return float(response.choices[0].message.content.split()[0])
    
    def _analyze_structure(self, content: str) -> float:
        """分析内容的结构质量"""
        score = 0
        # 检查标题层级
        if '##' in content or '# ' in content:
            score += 25
        # 检查列表使用
        if '\n- ' in content or '\n1. ' in content:
            score += 20
        # 检查段落开头是否有主题句
        paragraphs = content.split('\n\n')
        short_openers = sum(1 for p in paragraphs if len(p.split()[0]) < 100)
        score += min(30, (short_openers / max(len(paragraphs), 1)) * 30)
        # 检查数据/数字密度
        import re
        numbers = re.findall(r'\d+%', content)
        score += min(25, len(numbers) * 5)
        return min(100, score)
    
    def _analyze_authority(self, content: str) -> float:
        """分析权威性信号"""
        score = 0
        # 检查外部引用
        import re
        citations = re.findall(r'(根据|研究显示|据.*报道|来源)', content)
        score += min(40, len(citations) * 10)
        # 检查数据支撑
        data_points = re.findall(r'\d+\.?\d*%|\d+\.?\d*亿|\d+\.?\d*万', content)
        score += min(30, len(data_points) * 5)
        # 检查专业术语密度
        terms = re.findall(r'[A-Z]{2,}|[\u4e00-\u9fa5]{4,}(?:模型|算法|框架|系统|技术)', content)
        score += min(30, len(terms) * 3)
        return min(100, score)
    
    def _analyze_data_density(self, content: str) -> float:
        """分析数据密度"""
        import re
        numbers = re.findall(r'\d+', content)
        words = len(content.split())
        density = len(numbers) / max(words, 1)
        return min(100, density * 500)  # 理想密度约 20%
    
    def _generate_recommendations(self, *scores) -> List[str]:
        """生成优化建议"""
        recs = []
        labels = ['语义覆盖', '结构质量', '权威性信号', '数据密度']
        for label, score in zip(labels, scores):
            if score < 60:
                recs.append(f"⚠️ {label}分数较低（{score}分），建议加强")
        return recs

# ===== AI 内容农场检测系统 =====
import re
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class ContentQualityReport:
    """内容质量检测报告"""
    url: str
    ai_generated_probability: float  # AI 生成概率
    spam_signals: List[str]         # 垃圾信号列表
    quality_score: float            # 质量分数 (0-100)
    recommendation: str             # 处理建议

class ContentFarmDetector:
    """AI 内容农场检测器"""
    
    # AI 生成内容的典型特征
    AI_GENERATION_PATTERNS = [
        # 过度使用模板化表达
        (r'(总的来说|综上所述|总而言之).*?(重要性|意义|价值)', 15),
        # 缺乏具体细节的空泛论述
        (r'在.*时代|随着.*发展|.*越来越', 10),
        # 重复性段落结构
        (r'(首先|其次|再次|最后).*?(首先|其次|再次)', 20),
        # 缺乏具体数据支撑
        (r'^((?!\d).)*$', 5),  # 整段无数字
    ]
    
    # 高质量内容的特征
    QUALITY_SIGNALS = [
        (r'\d{4}年', 10),              # 年份引用
        (r'根据.*研究', 15),            # 研究引用
        (r'(作者|记者|编辑).*报道', 10),  # 署名信息
        (r'https?://', 10),             # 外部链接
        (r'\d+\.?\d*%', 5),          # 百分比数据
    ]
    
    def analyze_content(self, url: str, content: str) -> ContentQualityReport:
        """分析单篇内容的质量"""
        ai_score = 0
        quality_score = 0
        spam_signals = []
        
        # 检测 AI 生成特征
        paragraphs = content.split('\n\n')
        for pattern, weight in self.AI_GENERATION_PATTERNS:
            matches = len(re.findall(pattern, content, re.MULTILINE))
            ai_score += matches * weight
            if matches > 0:
                spam_signals.append(f"发现AI生成特征: {pattern[:30]}")
        
        # 检测高质量信号
        for pattern, weight in self.QUALITY_SIGNALS:
            matches = len(re.findall(pattern, content))
            quality_score += matches * weight
        
        # 分析文本复杂度
        avg_sentence_length = self._avg_sentence_length(content)
        if avg_sentence_length < 10:  # 句子过短，可能是 AI 简化
            ai_score += 20
            spam_signals.append("平均句长过短（AI简化特征）")
        
        # 计算重复度
        repetition_ratio = self._calculate_repetition(content)
        if repetition_ratio > 0.3:
            ai_score += 25
            spam_signals.append(f"文本重复度高: {repetition_ratio:.0%}")
        
        # 最终评分
        ai_probability = min(95, ai_score)
        final_quality = max(0, min(100, quality_score - ai_score * 0.3))
        
        if ai_probability > 70 and final_quality < 40:
            recommendation = "🚨 高概率AI内容农场，建议屏蔽"
        elif ai_probability > 50:
            recommendation = "⚠️ 疑似AI生成内容，建议人工审核"
        else:
            recommendation = "✅ 内容质量可接受"
        
        return ContentQualityReport(
            url=url,
            ai_generated_probability=ai_probability,
            spam_signals=spam_signals,
            quality_score=round(final_quality, 1),
            recommendation=recommendation
        )
    
    def _avg_sentence_length(self, text: str) -> float:
        sentences = re.split(r'[。！？.!?]', text)
        sentences = [s.strip() for s in sentences if s.strip()]
        if not sentences:
            return 0
        return sum(len(s.split()) for s in sentences) / len(sentences)
    
    def _calculate_repetition(self, text: str) -> float:
        words = text.split()
        if len(words) < 10:
            return 0
        unique_ngrams = set()
        total_ngrams = 0
        for i in range(len(words) - 2):
            ngram = ' '.join(words[i:i+3])
            unique_ngrams.add(ngram)
            total_ngrams += 1
        return 1 - (len(unique_ngrams) / max(total_ngrams, 1))

# === 使用示例 ===
detector = ContentFarmDetector()
report = detector.analyze_content(
    url="https://example.com/article",
    content="在AI时代，人工智能正在改变我们的生活..."
)
print(f"AI生成概率: {report.ai_generated_probability}%")
print(f"质量分数: {report.quality_score}")
print(f"建议: {report.recommendation}")