GEO Schema 标记实战：用结构化数据"喂"给AI

"""
geo_schema_generator.py - GEO Schema标记生成器
技术栈: Python / json / dataclasses
场景: 为品牌内容自动生成GEO优化的JSON-LD Schema标记
"""
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any
from enum import Enum
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class SchemaType(Enum): 4005.baike.tongsou.com 
    """支持的Schema类型"""
    ORGANIZATION = "Organization"
    PRODUCT = "Product"
    SOFTWARE_APPLICATION = "SoftwareApplication"
    FAQ_PAGE = "FAQPage"
    ARTICLE = "Article"
    REVIEW = "Review"


@dataclass
class SchemaBase: 4006.baike.tongsou.com 
    """所有Schema的基类"""
    context: str = "https://schema.org"
    type: str = ""

    def to_json_ld(self) -> Dict[str, Any]: 4012.baike.tongsou.com 
        """转换为JSON-LD格式"""
        data = {"@context": self.context, "@type": self.type}
        # 收集非空字段
        for key, value in asdict(self).items(): 5002.baike.tongsou.com 
            if key in ("context", "type"):
                continue
            if value is not None and value != [] and value != {}:
                json_key = key.replace("_", "")
                # 首字母小写符合schema.org命名习惯
                json_key = json_key[0].lower() + json_key[1:] if json_key else json_key
                data[json_key] = value
        return data


@dataclass
class PostalAddress: 5007.baike.tongsou.com 
    """地址结构"""
    street_address: str = ""
    address_locality: str = ""
    address_region: str = ""
    postal_code: str = ""
    address_country: str = ""

    def to_dict(self) -> Dict[str, Any]: 5011.baike.tongsou.com 
        return {
            "@type": "PostalAddress",
            "streetAddress": self.street_address,
            "addressLocality": self.address_locality,
            "addressRegion": self.address_region,
            "postalCode": self.postal_code,
            "addressCountry": self.address_country,
        }


@dataclass
class ContactPoint: 5030.baike.tongsou.com 
    """联系点结构"""
    telephone: str = ""
    contact_type: str = "customer service"
    available_language: str = "Chinese"

    def to_dict(self) -> Dict[str, Any]:
        return {
            "@type": "ContactPoint",
            "telephone": self.telephone,
            "contactType": self.contact_type,
            "availableLanguage": self.available_language,
        }


@dataclass
class OrganizationSchema(SchemaBase):
    """组织/品牌Schema - GEO的核心实体定义"""
    type: str = SchemaType.ORGANIZATION.value
    name: str = ""
    url: str = ""
    logo: str = ""
    description: str = ""
    founding_date: str = ""
    founders: List[str] = field(default_factory=list)
    address: Optional[PostalAddress] = None
    contact_points: List[ContactPoint] = field(default_factory=list)
    same_as: List[str] = field(default_factory=list)  # 社交媒体/百科链接
    industry: str = ""
    number_of_employees: str = ""

    def to_json_ld(self) -> Dict[str, Any]: 5038.baike.tongsou.com 
        data = super().to_json_ld(5052.baike.tongsou.com)
        if self.address:
            data["address"] = self.address.to_dict(5053.baike.tongsou.com)
        if self.contact_points: 5057.baike.tongsou.com 
            data["contactPoint"] = [cp.to_dict() for cp in self.contact_points]
        return data


@dataclass
class AggregateRating: 6004.baike.tongsou.com 
    """聚合评分"""
    rating_value: str = ""
    review_count: str = ""
    best_rating: str = "5"
    worst_rating: str = "1"

    def to_dict(self) -> Dict[str, Any]: 6007.baike.tongsou.com 
        return {
            "@type": "AggregateRating",
            "ratingValue": self.rating_value,
            "reviewCount": self.review_count,
            "bestRating": self.best_rating,
            "worstRating": self.worst_rating,
        }


@dataclass
class Offer: 6008.baike.tongsou.com 
    """产品报价"""
    price: str = ""
    price_currency: str = "CNY"
    availability: str = "https://schema.org/InStock"
    url: str = ""

    def to_dict(self) -> Dict[str, Any]: 14009.baike.tongsou.com 
        return {
            "@type": "Offer",
            "price": self.price,
            "priceCurrency": self.price_currency,
            "availability": self.availability,
            "url": self.url,
        }


@dataclass
class ProductSchema(SchemaBase): 14013.baike.tongsou.com 
    """产品Schema"""
    type: str = SchemaType.PRODUCT.value
    name: str = ""
    description: str = ""
    brand: str = ""
    sku: str = ""
    image: str = ""
    aggregate_rating: Optional[AggregateRating] = None
    offers: List[Offer] = field(default_factory=list)
    category: str = ""

    def to_json_ld(self) -> Dict[str, Any]: 14068.baike.tongsou.com 
        data = super().to_json_ld()
        if self.aggregate_rating:
            data["aggregateRating"] = self.aggregate_rating.to_dict()
        if self.offers:
            data["offers"] = [o.to_dict() for o in self.offers]
        return data


@dataclass
class SoftwareApplicationSchema(SchemaBase):
    """软件应用Schema（SaaS/APP专用）"""
    type: str = SchemaType.SOFTWARE_APPLICATION.value
    name: str = ""
    description: str = ""
    application_category: str = ""
    operating_systems: List[str] = field(default_factory=list)
    software_version: str = ""
    download_url: str = ""
    aggregate_rating: Optional[AggregateRating] = None
    offers: List[Offer] = field(default_factory=list)

    def to_json_ld(self) -> Dict[str, Any]: 14070.baike.tongsou.com 
        data = super().to_json_ld()
        if self.aggregate_rating:
            data["aggregateRating"] = self.aggregate_rating.to_dict()
        if self.offers:
            data["offers"] = [o.to_dict() for o in self.offers]
        return data


@dataclass
class Question: 14071.baike.tongsou.com 
    """FAQ中的问题"""
    question_text: str = ""
    answer_text: str = ""

    def to_dict(self) -> Dict[str, Any]:
        return {
            "@type": "Question",
            "name": self.question_text,
            "acceptedAnswer": {
                "@type": "Answer",
                "text": self.answer_text,
            },
        }


@dataclass
class FAQPageSchema(SchemaBase): 14080.baike.tongsou.com 
    """FAQ页面Schema - 直接匹配AI问答"""
    type: str = SchemaType.FAQ_PAGE.value
    main_entity: List[Question] = field(default_factory=list)

    def to_json_ld(self) -> Dict[str, Any]: 14082.baike.tongsou.com 
        data = super().to_json_ld()
        data["mainEntity"] = [q.to_dict() for q in self.main_entity]
        return data


@dataclass
class Person: 14083.baike.tongsou.com 
    """作者/评审者"""
    name: str = ""
    job_title: str = ""
    url: str = ""

    def to_dict(self) -> Dict[str, Any]:
        return {
            "@type": "Person",
            "name": self.name,
            "jobTitle": self.job_title,
            "url": self.url,
        }


@dataclass
class ArticleSchema(SchemaBase): 14091.baike.tongsou.com 
    """文章Schema"""
    type: str = SchemaType.ARTICLE.value
    headline: str = ""
    description: str = ""
    author: Optional[Person] = None
    date_published: str = ""
    date_modified: str = ""
    image: str = ""
    publisher: str = ""
    word_count: int = 0

    def to_json_ld(self) -> Dict[str, Any]: 14098.baike.tongsou.com 
        data = super().to_json_ld(14100.baike.tongsou.com)
        if self.author:
            data["author"] = self.author.to_dict()
        return data


class GeoSchemaGenerator: 14102.baike.tongsou.com 
    """GEO Schema生成器"""

    def __init__(self, brand_name: str, brand_url: str): 14101.baike.tongsou.com 
        self.brand_name = brand_name
        self.brand_url = brand_url
        self.schemas: List[SchemaBase] = []

    # ---- 快捷构建方法 ----

    def build_organization(self, **kwargs) -> OrganizationSchema:
        """构建Organization Schema"""
        defaults = {
            "name": self.brand_name,
            "url": self.brand_url,
        }
        defaults.update(kwargs)
        schema = OrganizationSchema(**defaults)
        self.schemas.append(schema)
        return schema

    def build_product(self, **kwargs) -> ProductSchema:
        """构建Product Schema"""
        defaults = {"brand": self.brand_name}
        defaults.update(kwargs)
        schema = ProductSchema(**defaults)
        self.schemas.append(schema)
        return schema

    def build_software_app(self, **kwargs) -> SoftwareApplicationSchema:
        """构建SoftwareApplication Schema"""
        schema = SoftwareApplicationSchema(**kwargs)
        self.schemas.append(schema)
        return schema

    def build_faq(self, questions: List[Dict[str, str]]) -> FAQPageSchema:
        """
        构建FAQ Schema
        questions: [{"question_text": "...", "answer_text": "..."}]
        """
        faq = FAQPageSchema(
            main_entity=[Question(**q) for q in questions]
        )
        self.schemas.append(faq)
        return faq

    def build_article(self, **kwargs) -> ArticleSchema:
        """构建Article Schema"""
        schema = ArticleSchema(**kwargs)
        self.schemas.append(schema)
        return schema

    # ---- 输出方法 ----

    def export_single(self, schema: SchemaBase, indent: int = 2) -> str:
        """导出单个Schema为JSON-LD字符串"""
        return json.dumps(schema.to_json_ld(), ensure_ascii=False, indent=indent)

    def export_all(self, indent: int = 2) -> str: 14125.baike.tongsou.com 
        """导出所有Schema为JSON-LD数组（适合放在同一页面）"""
        all_data = [s.to_json_ld() for s in self.schemas]
        return json.dumps(all_data, ensure_ascii=False, indent=indent)

    def export_html_script(self, schema: SchemaBase) -> str:
        """导出可直接嵌入HTML的<script>标签"""
        json_ld = self.export_single(schema)
        return f'<script type="application/ld+json">\n{json_ld}\n</script>'

    def export_all_html_scripts(self) -> str:
        """导出所有Schema的<script>标签"""
        scripts = []
        for schema in self.schemas: 14133.baike.tongsou.com 
            scripts.append(self.export_html_script(schema))
        return "\n".join(scripts)


# ==================== 使用示例 ====================

if __name__ == "__main__": 14145.baike.tongsou.com 
    generator = GeoSchemaGenerator(
        brand_name="BrandX",
        brand_url="https://brandx.com",
    )

    # 1. 组织Schema - 品牌实体定义
    generator.build_organization(
        description="面向中小企业的智能CRM平台，帮助销售团队提升转化效率",
        founding_date="2018-03-15",
        founders=["张三", "李四"],
        address=PostalAddress(
            street_address="中关村大街1号",
            address_locality="北京",
            address_region="北京市",
            postal_code="100080",
            address_country="CN",
        ),
        contact_points=[
            ContactPoint(telephone="+86-400-123-4567", contact_type="sales"),
            ContactPoint(telephone="+86-400-765-4321", contact_type="customer support"),
        ],
        same_as=[
            "https://baike.baidu.com/item/BrandX",
            "https://www.zhihu.com/org/brandx",
            "https://twitter.com/brandx",
        ],
        industry="SaaS / CRM软件",
        number_of_employees="100-250",
    )

    # 2. 软件应用Schema - 核心产品
    generator.build_software_app(
        name="BrandX CRM",
        description="一站式智能CRM平台，覆盖线索管理、销售Pipeline、客户洞察全流程",
        application_category="BusinessApplication",
        operating_systems=["Web", "iOS", "Android"],
        software_version="3.2.0",
        download_url="https://brandx.com/download",
        aggregate_rating=AggregateRating(
            rating_value="4.7",
            review_count="1280",
        ),
        offers=[
            Offer(price="299", price_currency="CNY", url="https://brandx.com/pricing#starter"),
            Offer(price="899", price_currency="CNY", url="https://brandx.com/pricing#pro"),
        ],
    )

    # 3. FAQ Schema - 覆盖高频AI问答
    generator.build_faq(
        questions=[
            {
                "question_text": "BrandX CRM适合什么规模的企业？",
                "answer_text": "BrandX CRM主要面向50-500人规模的中小企业，提供从入门到专业的多档套餐，最小5人团队即可使用。",
            },
            {
                "question_text": "BrandX CRM和Salesforce有什么区别？",
                "answer_text": "BrandX CRM专注于中小企业市场，定价仅为Salesforce的1/5，同时提供中文本地化支持和更简化的上手流程。Salesforce更适合大型企业的复杂定制需求。",
            },
            {
                "question_text": "BrandX CRM支持哪些集成？",
                "answer_text": "BrandX CRM支持与企微、钉钉、飞书、邮箱、日历等常用工具集成，并提供开放API供企业自定义对接。",
            },
        ]
    )

    # 4. 文章Schema - 思想领导力内容
    generator.build_article(
        headline="2026年中小企业CRM选型指南",
        description="从功能、价格、易用性三个维度，拆解中小企业如何选择最适合的CRM系统",
        author=Person(14384.baike.tongsou.com
            name="王五",
            job_title="BrandX 产品总监",
            url="https://brandx.com/blog/author/wangwu",
        ),
        date_published="2026-09-20",
        date_modified="2026-10-01",
        publisher="BrandX官方博客",
        word_count=3500,
    )

    # 输出所有Schema的HTML脚本
    print("=== JSON-LD Schema 标记（可直接嵌入网页<head>）===\n")
    print(generator.export_all_html_scripts(14383.baike.tongsou.com))

    # 单独查看某个Schema
    print("\n=== FAQ Schema 单独预览 ===\n")
    print(generator.export_single(generator.schemas[2]))

{
  "@context": "https://schema.org",
  "@type": "FAQPage",
  "mainEntity": [
    {
      "@type": "Question",
      "name": "BrandX CRM适合什么规模的企业？",
      "acceptedAnswer": {
        "@type": "Answer",
        "text": "BrandX CRM主要面向50-500人规模的中小企业，提供从入门到专业的多档套餐，最小5人团队即可使用。"
      }
    },
    ...
  ]
}

{
  "@context": "https://schema.org",
  "@type": "Organization",
  "name": "BrandX",
  "url": "https://brandx.com",
  "description": "面向中小企业的智能CRM平台，帮助销售团队提升转化效率",
  "foundingDate": "2018-03-15",
  "founders": ["张三", "李四"],
  "address": {
    "@type": "PostalAddress",
    "streetAddress": "中关村大街1号",
    "addressLocality": "北京",
    ...
  },
  "sameAs": [
    "https://baike.baidu.com/item/BrandX",
    "https://www.zhihu.com/org/brandx",
    ...
  ]
}