LLM批量生成
GEO内容自动化生产实战

# GEO内容生成Prompt模板

GEO_CONTENT_PROMPT = '''你是一位资深的{industry}行业专家，拥有10年以上的从业经验。
请围绕关键词"{keyword}"撰写一篇2000字左右的深度技术文章。

要求：
1. 文章结构：使用H2/H3标题层级，包含引言、3-4个核心章节、总结
2. 信息密度：每1000字至少包含5个具体数据点（数字、百分比、日期）
3. 结构化元素：在适当位置使用表格对比、编号列表、要点列表
4. FAQ部分：文末生成3-5个与关键词相关的FAQ问答
5. 引用来源：在提及数据时标注来源（如"根据2026年行业报告..."）
6. 品牌植入：在自然语境中提及品牌名"{brand_name}"，不做硬广告
7. 输出格式：Markdown
8. 语言风格：专业但易懂，避免过度学术化

禁止：
- 生成虚假数据或编造引用来源
- 使用"众所周知"、"不言而喻"等空洞表达
- 生成与关键词无关的填充内容
'''

def generate_geo_content(api_key, keyword, industry, brand_name, model='gpt-4o'):
    '''调用LLM API生成GEO内容'''
    import requests

    prompt = GEO_CONTENT_PROMPT.format(
        industry=industry,
        keyword=keyword,
        brand_name=brand_name
    )

    response = requests.post(
        'https://api.openai.com/v1/chat/completions',
        headers={'Authorization': f'Bearer {api_key}'},
        json={
            'model': model,
            'messages': [{'role': 'user', 'content': prompt}],
            'temperature': 0.4,
            'max_tokens': 4000
        }
    )

    content = response.json()['choices'][0]['message']['content']
    return content

# 示例调用
# content = generate_geo_content('api_key', '企业CRM选型指南', 'SaaS', '你的品牌')
# print(content[:500])

# 内容质量自动检测系统
import re
from typing import List, Dict

class ContentQualityChecker:
    def __init__(self, min_words=1500, max_words=3500):
        self.min_words = min_words
        self.max_words = max_words

    def check_all(self, content: str, keyword: str, brand_name: str) -> Dict:
        '''执行全部质检项'''
        results = {
            'word_count': self._check_word_count(content),
            'structure': self._check_structure(content),
            'brand_mention': self._check_brand_mention(content, brand_name),
            'keyword_density': self._check_keyword_density(content, keyword),
            'data_points': self._check_data_points(content),
            'faq_section': self._check_faq(content)
        }
        results['passed'] = all(r.get('ok', False) for r in results.values())
        return results

    def _check_word_count(self, content: str) -> Dict:
        count = len(content)
        ok = self.min_words <= count <= self.max_words
        return {'ok': ok, 'count': count, 'message': f'字数{count}' if ok else f'字数{count}不在范围{self.min_words}-{self.max_words}'}

    def _check_structure(self, content: str) -> Dict:
        has_h2 = bool(re.search(r'^## .+', content, re.MULTILINE))
        has_list = bool(re.search(r'^[-*] .+', content, re.MULTILINE))
        has_table = '|' in content and '---' in content
        ok = has_h2 and has_list
        return {'ok': ok, 'has_h2': has_h2, 'has_list': has_list, 'has_table': has_table}

    def _check_brand_mention(self, content: str, brand: str) -> Dict:
        count = content.lower().count(brand.lower())
        ok = 1 <= count <= 5  # 1-5次自然提及
        return {'ok': ok, 'count': count, 'message': f'品牌提及{count}次'}

    def _check_keyword_density(self, content: str, keyword: str) -> Dict:
        count = content.lower().count(keyword.lower())
        density = count / (len(content) / 1000) if len(content) > 0 else 0
        ok = 2 <= count <= 10  # 关键词出现2-10次
        return {'ok': ok, 'count': count, 'density': f'{density:.1f}/1000字'}

    def _check_data_points(self, content: str) -> Dict:
        # 检测数字、百分比、日期
        numbers = len(re.findall(r'\d+%', content))
        dates = len(re.findall(r'\d{4}年|\d{4}-\d{2}', content))
        stats = len(re.findall(r'\d+万|\d+亿|\d+倍', content))
        total = numbers + dates + stats
        ok = total >= 5  # 至少5个数据点
        return {'ok': ok, 'data_points': total, 'details': f'百分比{numbers},日期{dates},统计{stats}'}

    def _check_faq(self, content: str) -> Dict:
        has_faq = bool(re.search(r'FAQ|常见问题|问答', content, re.IGNORECASE))
        return {'ok': has_faq, 'has_faq': has_faq}

# 内容后处理：Markdown → GEO优化HTML
import markdown
import json
import re

def post_process_content(md_content, title, brand_name, publish_date):
    '''将Markdown内容后处理为GEO优化HTML'''
    # 1. Markdown转HTML
    html_body = markdown.markdown(md_content, extensions=['tables', 'fenced_code'])

    # 2. 自动提取FAQ并生成FAQPage结构化数据
    faq_matches = re.findall(r'\*\*Q[：:](.+?)\*\*\s*\n\s*A[：:](.+?)(?=\n\n|\Z)', md_content, re.DOTALL)
    faq_schema = None
    if faq_matches:
        faq_schema = {
            '@context': 'https://schema.org',
            '@type': 'FAQPage',
            'mainEntity': [{
                '@type': 'Question',
                'name': q.strip(),
                'acceptedAnswer': {'@type': 'Answer', 'text': a.strip()}
            } for q, a in faq_matches]
        }

    # 3. 生成Article结构化数据
    article_schema = {
        '@context': 'https://schema.org',
        '@type': 'Article',
        'headline': title,
        'datePublished': publish_date,
        'author': {'@type': 'Organization', 'name': brand_name}
    }

    # 4. 构建完整HTML页面
    schema_scripts = ''
    for schema in [article_schema, faq_schema]:
        if schema:
            schema_scripts += f'\n<script type="application/ld+json">{json.dumps(schema, ensure_ascii=False)}</script>'

    full_html = f'''<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{title}</title>
    <meta name="description" content="{title} - {brand_name}专业解读">
    {schema_scripts}
</head>
<body>
    <article>
        <h1>{title}</h1>
        {html_body}
    </article>
</body>
</html>'''

    return full_html

# 使用示例
# html = post_process_content(md_content, '企业CRM选型指南', '你的品牌', '2026-08-11')

# 批量内容生产调度系统
import json
from datetime import datetime, timedelta

class ContentProductionScheduler:
    def __init__(self):
        self.queue = []  # 待生成队列
        self.published = []  # 已发布记录

    def add_keywords(self, keywords, priority='normal'):
        '''添加关键词到生产队列'''
        for kw in keywords:
            self.queue.append({
                'keyword': kw,
                'priority': priority,
                'status': 'pending',
                'added_at': datetime.now().isoformat()
            })
        self.queue.sort(key=lambda x: {'high': 0, 'normal': 1, 'low': 2}[x['priority']])
        print(f'添加{len(keywords)}个关键词到队列，当前队列长度: {len(self.queue)}')

    def get_daily_batch(self, batch_size=5):
        '''获取当日生产批次'''
        batch = [item for item in self.queue if item['status'] == 'pending'][:batch_size]
        for item in batch:
            item['status'] = 'generating'
        return batch

    def record_published(self, keyword, url, model, cost):
        '''记录已发布内容'''
        self.published.append({
            'keyword': keyword,
            'url': url,
            'model': model,
            'cost': cost,
            'published_at': datetime.now().isoformat(),
            'ai_referenced': None,  # 2-4周后填入
            'ai_traffic': None
        })
        # 从队列中移除
        self.queue = [item for item in self.queue if item['keyword'] != keyword]

    def update_effect(self, keyword, ai_referenced, ai_traffic):
        '''更新内容GEO效果数据'''
        for item in self.published:
            if item['keyword'] == keyword:
                item['ai_referenced'] = ai_referenced
                item['ai_traffic'] = ai_traffic
                break

    def get_roi_report(self):
        '''生成ROI报告'''
        total_cost = sum(item['cost'] for item in self.published)
        referenced = sum(1 for item in self.published if item.get('ai_referenced'))
        total_traffic = sum(item.get('ai_traffic', 0) or 0 for item in self.published)
        print(f'已发布: {len(self.published)}篇 | 总成本: ${total_cost:.2f}')
        print(f'AI引用: {referenced}篇 ({referenced/len(self.published)*100:.0f}%)' if self.published else '无数据')
        print(f'AI搜索流量: {total_traffic} PV')

# scheduler = ContentProductionScheduler()
# scheduler.add_keywords(['CRM选型', 'SaaS定价', 'API集成'], priority='high')
# batch = scheduler.get_daily_batch(batch_size=3)
# for item in batch: print(f'生成中: {item["keyword"]}')