13.2 结构化数据与语义工具（Schema验证、知识图谱API、语义相似度）

# schema_geo_validator.py
import json
from urllib.parse import urlparse

def validate_geo_schema(schema_json):
    """验证Schema是否满足生成式引擎的引用需求"""
    issues = []
    
    # 检查基本结构
    if '@context' not in schema_json:
        issues.append("缺少@context字段")
    if '@type' not in schema_json:
        issues.append("缺少@type字段")
    
    # 检查实体类型
    entity_type = schema_json.get('@type')
    if entity_type in ['Article', 'TechArticle', 'NewsArticle']:
        # 检查作者信息
        if 'author' not in schema_json:
            issues.append("文章类Schema缺少author属性")
        else:
            author = schema_json['author']
            if isinstance(author, dict) and '@type' not in author:
                issues.append("author对象缺少@type")
        
        # 检查时间戳
        if 'datePublished' not in schema_json:
            issues.append("缺少datePublished（影响时效性评分）")
        if 'dateModified' not in schema_json:
            issues.append("缺少dateModified（影响内容新鲜度）")
        
        # 检查引用关系
        if 'citation' in schema_json:
            citations = schema_json['citation']
            if isinstance(citations, list) and len(citations) > 5:
                issues.append("引用过多（>5），可能被视为过度优化")
    
    # 检查URL有效性
    if 'url' in schema_json:
        parsed = urlparse(schema_json['url'])
        if not parsed.scheme or not parsed.netloc:
            issues.append(f"URL格式无效: {schema_json['url']}")
    
    # 检查speakable属性（对语音搜索和摘要生成重要）
    if entity_type in ['WebPage', 'Article'] and 'speakable' not in schema_json:
        issues.append("建议添加speakable属性以优化生成式摘要")
    
    return {
        'valid': len(issues) == 0,
        'issues': issues,
        'geo_score': max(0, 10 - len(issues))  # 10分制
    }

# 使用示例
with open('schema_example.json', 'r') as f:
    schema = json.load(f)
result = validate_geo_schema(schema)
print(f"GEO验证得分: {result['geo_score']}/10")
for issue in result['issues']:
    print(f"  - {issue}")

# knowledge_graph_checker.py
import requests
import json

class KnowledgeGraphChecker:
    def __init__(self, google_api_key, wikidata_endpoint="https://www.wikidata.org/wiki/Special:EntityData"):
        self.google_api_key = google_api_key
        self.wikidata_endpoint = wikidata_endpoint
    
    def check_google_kg(self, entity_name, limit=5):
        """查询Google Knowledge Graph"""
        url = "https://kgsearch.googleapis.com/v1/entities:search"
        params = {
            'query': entity_name,
            'key': self.google_api_key,
            'limit': limit,
            'languages': ['zh', 'en']
        }
        response = requests.get(url, params=params)
        if response.status_code == 200:
            data = response.json()
            results = []
            for item in data.get('itemListElement', []):
                result = item.get('result', {})
                results.append({
                    'name': result.get('name'),
                    'description': result.get('description'),
                    'score': item.get('resultScore', 0),
                    'types': result.get('@type', []),
                    'detailed_description': result.get('detailedDescription', {}).get('articleBody', '')
                })
            return results
        return []
    
    def check_wikidata(self, entity_name):
        """查询Wikidata"""
        # 先搜索实体ID
        search_url = "https://www.wikidata.org/w/api.php"
        params = {
            'action': 'wbsearchentities',
            'search': entity_name,
            'language': 'zh',
            'format': 'json'
        }
        response = requests.get(search_url, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get('search'):
                entity_id = data['search'][0]['id']
                # 获取实体详细信息
                entity_url = f"{self.wikidata_endpoint}/{entity_id}.json"
                entity_response = requests.get(entity_url)
                if entity_response.status_code == 200:
                    return entity_response.json()
        return None
    
    def analyze_content_entities(self, text):
        """分析文本中的实体及其知识图谱关联度"""
        # 使用简单的NER提取实体（生产环境建议使用spaCy或StanfordNLP）
        # 这里仅作示例
        entities = self._extract_entities_simple(text)
        results = []
        for entity in entities:
            kg_data = self.check_google_kg(entity)
            wikidata = self.check_wikidata(entity)
            results.append({
                'entity': entity,
                'in_google_kg': len(kg_data) > 0,
                'google_score': kg_data[0]['score'] if kg_data else 0,
                'in_wikidata': wikidata is not None,
                'types': kg_data[0]['types'] if kg_data else []
            })
        return results
    
    def _extract_entities_simple(self, text):
        """简单实体提取（仅用于演示）"""
        # 实际项目应使用NLP库
        import re
        # 假设实体是中文专有名词（2-6个汉字）
        pattern = r'[\u4e00-\u9fa5]{2,6}'
        candidates = re.findall(pattern, text)
        # 去重并过滤常见词
        stop_words = {'我们', '他们', '这个', '那个', '什么', '如何', '为什么'}
        return list(set([c for c in candidates if c not in stop_words]))

# 使用示例
checker = KnowledgeGraphChecker(google_api_key='YOUR_API_KEY')
content = "TensorFlow是Google开发的开源机器学习框架，广泛应用于深度学习和人工智能领域。"
results = checker.analyze_content_entities(content)
for r in results:
    print(f"实体: {r['entity']}")
    print(f"  Google KG: {'✓' if r['in_google_kg'] else '✗'} (得分: {r['google_score']})")
    print(f"  Wikidata: {'✓' if r['in_wikidata'] else '✗'}")

# internal_knowledge_graph.py
from neo4j import GraphDatabase

class InternalKnowledgeGraph:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def add_entity(self, entity_id, name, type, properties=None):
        with self.driver.session() as session:
            session.run(
                "MERGE (e:Entity {id: $id}) "
                "SET e.name = $name, e.type = $type, e.properties = $props",
                id=entity_id, name=name, type=type, props=properties or {}
            )
    
    def add_relationship(self, from_id, to_id, relation_type, properties=None):
        with self.driver.session() as session:
            session.run(
                "MATCH (a:Entity {id: $from_id}), (b:Entity {id: $to_id}) "
                "MERGE (a)-[r:RELATES {type: $relation_type}]->(b) "
                "SET r.properties = $props",
                from_id=from_id, to_id=to_id,
                relation_type=relation_type, props=properties or {}
            )
    
    def query_entity_network(self, entity_id, depth=2):
        """查询实体的关联网络"""
        with self.driver.session() as session:
            result = session.run(
                "MATCH (e:Entity {id: $id})-[*1..$depth]-(connected) "
                "RETURN e, connected, relationships(e)",
                id=entity_id, depth=depth
            )
            return [record.data() for record in result]

# 使用示例
kg = InternalKnowledgeGraph("bolt://localhost:7687", "neo4j", "password")
kg.add_entity("tensorflow", "TensorFlow", "Software", {"version": "2.15", "license": "Apache 2.0"})
kg.add_entity("google", "Google", "Organization", {"headquarters": "Mountain View"})
kg.add_relationship("tensorflow", "google", "developed_by", {"year": 2015})

# semantic_similarity.py
import numpy as np
from openai import OpenAI
from sklearn.metrics.pairwise import cosine_similarity

class SemanticSimilarityChecker:
    def __init__(self, api_key, model="text-embedding-3-small"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    def get_embedding(self, text):
        """获取文本的向量嵌入"""
        response = self.client.embeddings.create(
            model=self.model,
            input=text
        )
        return response.data[0].embedding
    
    def calculate_similarity(self, query, content):
        """计算查询与内容的语义相似度"""
        query_embedding = self.get_embedding(query)
        content_embedding = self.get_embedding(content)
        
        similarity = cosine_similarity(
            [query_embedding],
            [content_embedding]
        )[0][0]
        
        return similarity
    
    def batch_analyze(self, queries, contents):
        """批量分析多个查询与内容的匹配度"""
        # 缓存嵌入以减少API调用
        query_embeddings = [self.get_embedding(q) for q in queries]
        content_embeddings = [self.get_embedding(c) for c in contents]
        
        results = []
        for q_idx, q_emb in enumerate(query_embeddings):
            for c_idx, c_emb in enumerate(content_embeddings):
                similarity = cosine_similarity([q_emb], [c_emb])[0][0]
                results.append({
                    'query': queries[q_idx],
                    'content': contents[c_idx][:50] + '...',
                    'similarity': similarity,
                    'geo_potential': self._classify_potential(similarity)
                })
        
        # 按相似度排序
        results.sort(key=lambda x: x['similarity'], reverse=True)
        return results
    
    def _classify_potential(self, similarity):
        """根据相似度分类GEO潜力"""
        if similarity > 0.85:
            return "高（极可能被引用）"
        elif similarity > 0.70:
            return "中（可能被引用）"
        elif similarity > 0.55:
            return "低（需要优化）"
        else:
            return "极低（内容不相关）"

# 使用示例
checker = SemanticSimilarityChecker(api_key='YOUR_OPENAI_API_KEY')

# 模拟用户查询
queries = [
    "什么是TensorFlow？",
    "TensorFlow和PyTorch有什么区别？",
    "如何安装TensorFlow？"
]

# 网站内容片段
contents = [
    "TensorFlow是一个端到端的开源机器学习平台，由Google开发。",
    "PyTorch是Facebook开发的深度学习框架，而TensorFlow由Google开发。",
    "安装TensorFlow可以使用pip install tensorflow命令。"
]

results = checker.batch_analyze(queries, contents)
for r in results[:5]:
    print(f"查询: {r['query']}")
    print(f"内容: {r['content']}")
    print(f"相似度: {r['similarity']:.3f}")
    print(f"GEO潜力: {r['geo_potential']}")
    print("---")

# local_semantic_similarity.py
from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class LocalSemanticSimilarity:
    def __init__(self, model_name="paraphrase-multilingual-MiniLM-L12-v2"):
        """初始化本地语义模型"""
        self.model = SentenceTransformer(model_name)
    
    def encode(self, texts):
        """批量编码文本"""
        return self.model.encode(texts, convert_to_numpy=True)
    
    def find_best_match(self, query, candidates, top_k=3):
        """找到与查询最匹配的候选内容"""
        query_emb = self.encode([query])
        candidate_embs = self.encode(candidates)
        
        similarities = cosine_similarity(query_emb, candidate_embs)[0]
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        
        results = []
        for idx in top_indices:
            results.append({
                'content': candidates[idx][:100],
                'similarity': similarities[idx],
                'index': idx
            })
        return results
    
    def content_clustering(self, contents, threshold=0.75):
        """对内容进行语义聚类（发现重复或高度相似内容）"""
        embs = self.encode(contents)
        n = len(contents)
        duplicates = []
        
        for i in range(n):
            for j in range(i+1, n):
                sim = cosine_similarity([embs[i]], [embs[j]])[0][0]
                if sim > threshold:
                    duplicates.append({
                        'content_a': contents[i][:50],
                        'content_b': contents[j][:50],
                        'similarity': sim,
                        'action': '合并或删除' if sim > 0.9 else '考虑优化'
                    })
        
        return duplicates

# 使用示例
local_checker = LocalSemanticSimilarity()

# 检查内容是否重复
contents = [
    "TensorFlow是Google开发的机器学习框架。",
    "TensorFlow是一个由Google开发的端到端机器学习平台。",
    "PyTorch是Facebook开发的深度学习框架。"
]

duplicates = local_checker.content_clustering(contents)
for d in duplicates:
    print(f"重复内容: {d['content_a']} <-> {d['content_b']}")
    print(f"相似度: {d['similarity']:.2f}")
    print(f"建议: {d['action']}")

# geo_semantic_pipeline.py
import schedule
import time
import json
from datetime import datetime

class GEOSemanticPipeline:
    def __init__(self, config):
        self.schema_validator = SchemaGeoValidator()
        self.kg_checker = KnowledgeGraphChecker(config['google_api_key'])
        self.semantic_checker = LocalSemanticSimilarity()
        self.config = config
    
    def daily_audit(self):
        """每日审计：检查所有核心页面的语义健康度"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'pages_audited': 0,
            'issues_found': 0,
            'geo_score': 0
        }
        
        # 读取核心页面列表
        with open(self.config['core_pages_file'], 'r') as f:
            pages = json.load(f)
        
        total_score = 0
        for page in pages:
            page_report = self.audit_single_page(page)
            report['pages_audited'] += 1
            total_score += page_report['score']
            
            if page_report['issues']:
                report['issues_found'] += len(page_report['issues'])
                self._log_issue(page['url'], page_report['issues'])
        
        report['geo_score'] = total_score / len(pages) if pages else 0
        
        # 生成报告
        self._generate_report(report)
        return report
    
    def audit_single_page(self, page):
        """审计单个页面"""
        issues = []
        score = 10
        
        # 1. Schema验证
        schema_result = self.schema_validator.validate(page['schema'])
        if not schema_result['valid']:
            issues.extend([f"Schema: {i}" for i in schema_result['issues']])
            score -= len(schema_result['issues'])
        
        # 2. 知识图谱验证
        kg_result = self.kg_checker.analyze_content_entities(page['content'])
        missing_entities = [e['entity'] for e in kg_result if not e['in_google_kg']]
        if missing_entities:
            issues.append(f"知识图谱缺失实体: {missing_entities[:3]}")
            score -= 1
        
        # 3. 语义相似度检查
        # 检查内容是否与目标查询匹配
        for query in page['target_queries']:
            sim = self.semantic_checker.find_best_match(query, [page['content']])
            if sim[0]['similarity'] < 0.6:
                issues.append(f"查询'{query}'与内容语义匹配度低 ({sim[0]['similarity']:.2f})")
                score -= 1
        
        return {
            'url': page['url'],
            'score': max(0, score),
            'issues': issues
        }
    
    def _log_issue(self, url, issues):
        """记录问题到日志系统"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'url': url,
            'issues': issues
        }
        with open(self.config['issue_log_file'], 'a') as f:
            f.write(json.dumps(log_entry) + '\n')
    
    def _generate_report(self, report):
        """生成审计报告"""
        report_file = f"reports/geo_audit_{datetime.now().strftime('%Y%m%d')}.json"
        with open(report_file, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"审计报告已生成: {report_file}")

# 定时任务
config = {
    'google_api_key': 'YOUR_API_KEY',
    'core_pages_file': 'core_pages.json',
    'issue_log_file': 'geo_issues.log'
}

pipeline = GEOSemanticPipeline(config)

# 每天凌晨2点执行审计
schedule.every().day.at("02:00").do(pipeline.daily_audit)

# 也可以手动执行
if __name__ == "__main__":
    pipeline.daily_audit()