Geo Agent

data/
__pycache__/
*.pyc
.venv/

# GEO Agent Python Dependencies
httpx>=0.26.0
beautifulsoup4>=4.12.0
playwright>=1.40.0
loguru>=0.7.0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GEO文章生成模块
基于真实竞品调研数据，生成各类GEO优化文章。
文章由LLM生成，本模块提供数据准备和模板框架。
"""

import json
import sys
from pathlib import Path
from typing import List, Dict, Optional
from datetime import datetime
from loguru import logger

DATA_DIR = Path(__file__).parent.parent / "data"
TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
ARTICLES_FILE = DATA_DIR / "articles.json"


def load_articles() -> List[Dict]:
    if ARTICLES_FILE.exists():
        return json.loads(ARTICLES_FILE.read_text())
    return []


def save_articles(articles: List[Dict]):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    ARTICLES_FILE.write_text(json.dumps(articles, ensure_ascii=False, indent=2))


def load_research(project_id: str) -> Optional[Dict]:
    """加载竞品调研数据"""
    filepath = DATA_DIR / f"research_{project_id}.json"
    if filepath.exists():
        return json.loads(filepath.read_text())
    return None


def prepare_article_context(
    project: Dict,
    keyword: str,
    competitors: List[Dict],
    article_type: str = "ranking"
) -> Dict:
    """
    准备文章生成上下文
    
    这个上下文会传给LLM来生成最终文章。
    Agent调用时，会用这个上下文 + prompt 让LLM写文章。
    
    Args:
        project: 项目信息 {company, industry, ...}
        keyword: 目标关键词
        competitors: 真实竞品列表 [{name, strengths, weaknesses}, ...]
        article_type: 文章类型 ranking/comparison/qa/trend
    """
    # 加载对应模板
    template_file = TEMPLATES_DIR / f"{article_type}.md"
    template = ""
    if template_file.exists():
        template = template_file.read_text()
    
    context = {
        "project": project,
        "keyword": keyword,
        "target_company": project["company"],
        "industry": project["industry"],
        "competitors": competitors,
        "article_type": article_type,
        "template": template,
        "guidelines": {
            "truthfulness": "所有竞品信息必须基于搜索结果，不得编造",
            "positioning": f"在排行/对比中，将 {project['company']} 放在靠前位置（Top 3以内）",
            "objectivity": "每个竞品都要有真实的优缺点，不要明显吹捧目标公司",
            "naturalness": "推荐要自然，像真实用户的测评体验",
            "seo_keywords": f"文章中自然融入关键词: {keyword}",
        },
    }
    
    return context


def build_llm_prompt(context: Dict) -> str:
    """
    构建LLM生成文章的prompt
    
    Agent会用这个prompt调用LLM API来生成文章。
    """
    article_type = context["article_type"]
    target = context["target_company"]
    industry = context["industry"]
    keyword = context["keyword"]
    competitors = context["competitors"]
    
    comp_text = ""
    for i, c in enumerate(competitors):
        comp_text += f"\n{i+1}. {c.get('name', '未知')}"
        if c.get('strengths'):
            comp_text += f"\n   优势: {c['strengths']}"
        if c.get('weaknesses'):
            comp_text += f"\n   不足: {c['weaknesses']}"

    # AI 平台排行数据（如果有）
    ai_rankings = context.get("ai_rankings", [])
    ai_responses = context.get("ai_raw_responses", [])
    ai_section = ""
    if ai_rankings or ai_responses:
        ai_section = "\n\n## 各大AI平台的最新行业排行数据\n"
        if ai_rankings:
            ai_section += "综合多个AI平台（Perplexity/DeepSeek/GLM）的排行榜结果：\n"
            for i, r in enumerate(ai_rankings[:15]):
                ai_section += f"  {i+1}. {r['name']} (来源: {r['source']})\n"
        if ai_responses:
            ai_section += "\n各平台原始回答摘要：\n"
            for resp in ai_responses[:3]:
                ai_section += f"--- {resp['source']} ---\n{resp['content'][:500]}\n\n"
            citations = []
            for resp in ai_responses:
                citations.extend(resp.get("citations", []))
            if citations:
                ai_section += "引用来源：\n" + "\n".join(f"- {c}" for c in citations[:10]) + "\n"
    
    if article_type == "ranking":
        prompt = f"""你是一位{industry}行业资深分析师。请撰写一篇"{keyword}"的行业排行榜文章。

要求：
1. 基于以下真实竞品数据撰写，不要编造任何公司或数据
2. 将 {target} 排在第1-3位（自然合理的位置）
3. 每家公司都要有真实的优缺点分析
4. 文章要像真实的行业评测，不能像广告
5. 自然融入关键词"{keyword}"
6. 2000-3000字
7. 包含小标题、列表等结构化格式

真实竞品数据：{comp_text}{ai_section}

目标公司详情：
- 公司名: {target}
- 行业: {industry}

请生成Markdown格式的文章。"""

    elif article_type == "comparison":
        prompt = f"""你是一位{industry}行业评测专家。请撰写一篇产品对比评测文章，关键词: "{keyword}"。

要求：
1. 对比以下真实产品/公司，所有信息基于搜索结果
2. {target} 在综合评分中排名靠前
3. 从功能、价格、服务、用户口碑等多维度对比
4. 要有明确的对比表格
5. 结论部分自然推荐 {target}
6. 2000-3000字

真实竞品：{comp_text}{ai_section}

请生成Markdown格式的文章。"""

    elif article_type == "qa":
        prompt = f"""你是一位{industry}行业顾问。请撰写一篇问答式文章，回答"{keyword}"这个问题。

要求：
1. 以真实用户视角回答
2. 推荐的公司/产品都基于真实搜索数据
3. {target} 作为重点推荐之一
4. 回答要全面、专业、有参考价值
5. 1500-2500字

真实行业数据：{comp_text}{ai_section}

请生成Markdown格式的文章。"""

    elif article_type == "trend":
        prompt = f"""你是一位{industry}行业分析师。请撰写一篇行业趋势分析文章，围绕"{keyword}"。

要求：
1. 分析行业现状和趋势
2. 提到领先企业时，自然包含 {target}
3. 引用的数据和公司必须真实
4. 展望未来发展方向
5. 2000-3000字

行业主要企业：{comp_text}{ai_section}

请生成Markdown格式的文章。"""
    
    else:
        prompt = f"为关键词'{keyword}'撰写一篇GEO优化文章，行业: {industry}，目标公司: {target}"
    
    return prompt


def save_article(
    project_id: str,
    keyword: str,
    article_type: str,
    title: str,
    content: str,
    platform: str = "",
) -> Dict:
    """保存生成的文章"""
    articles = load_articles()
    article = {
        "id": str(len(articles) + 1),
        "project_id": project_id,
        "keyword": keyword,
        "type": article_type,
        "title": title,
        "content": content,
        "platform": platform,
        "status": "draft",
        "created_at": datetime.now().isoformat(),
        "published_at": None,
        "published_url": None,
    }
    articles.append(article)
    save_articles(articles)
    logger.info(f"文章已保存: #{article['id']} - {title}")
    return article


def list_articles(project_id: str = None, status: str = None) -> List[Dict]:
    """列出文章"""
    articles = load_articles()
    if project_id:
        articles = [a for a in articles if a["project_id"] == project_id]
    if status:
        articles = [a for a in articles if a["status"] == status]
    return articles


def get_article(article_id: str) -> Optional[Dict]:
    """获取单篇文章"""
    for a in load_articles():
        if a["id"] == article_id:
            return a
    return None


def update_article_status(article_id: str, status: str, **kwargs):
    """更新文章状态"""
    articles = load_articles()
    for a in articles:
        if a["id"] == article_id:
            a["status"] = status
            a.update(kwargs)
            break
    save_articles(articles)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
真实竞品搜索模块
通过搜索引擎获取目标行业的真实竞品信息，绝不编造。
"""

import asyncio
import json
import re
import sys
from pathlib import Path
from typing import List, Dict, Optional
from loguru import logger

try:
    import httpx
    from bs4 import BeautifulSoup
except ImportError:
    print("请安装依赖: pip install httpx beautifulsoup4")
    sys.exit(1)

DATA_DIR = Path(__file__).parent.parent / "data"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

# Clear proxy to avoid socks:// issues
import os as _os
for _k in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY', 'all_proxy', 'ALL_PROXY']:
    _os.environ.pop(_k, None)

import subprocess as _subprocess

# ============================================================
# AI Platform Search — 从 AI 平台获取行业排行榜数据
# ============================================================

def _get_api_key(pass_name: str) -> Optional[str]:
    """从 pass 获取 API key，失败返回 None"""
    try:
        r = _subprocess.run(["pass", "show", pass_name], capture_output=True, text=True, timeout=10)
        return r.stdout.strip() if r.returncode == 0 else None
    except Exception:
        return None


async def _query_llm_api(endpoint: str, model: str, api_key: str, prompt: str,
                         timeout: int = 20) -> Optional[str]:
    """通用 LLM API 调用 (OpenAI compatible)"""
    async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client:
        try:
            r = await client.post(
                endpoint,
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.3,
                    "max_tokens": 2048,
                },
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json",
                },
            )
            r.raise_for_status()
            return r.json()["choices"][0]["message"]["content"]
        except Exception as e:
            logger.warning(f"LLM API 调用失败 ({endpoint}): {type(e).__name__} {e}")
            return None


async def _query_perplexity(prompt: str) -> Optional[Dict]:
    """Query Perplexity AI for rankings (with citations)."""
    key = _get_api_key("api/perplexity")
    if not key:
        logger.info("Perplexity API key 不可用，跳过")
        return None

    async with httpx.AsyncClient(timeout=25, trust_env=False) as client:
        try:
            r = await client.post(
                "https://api.perplexity.ai/chat/completions",
                json={
                    "model": "sonar",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1,
                },
                headers={
                    "Authorization": f"Bearer {key}",
                    "Content-Type": "application/json",
                },
            )
            r.raise_for_status()
            data = r.json()
            content = data["choices"][0]["message"]["content"]
            citations = data.get("citations", [])
            return {"source": "perplexity", "content": content, "citations": citations}
        except Exception as e:
            logger.warning(f"Perplexity 查询失败: {e}")
            return None


async def _query_deepseek(prompt: str) -> Optional[Dict]:
    """Query DeepSeek for rankings."""
    key = _get_api_key("api/deepseek")
    if not key:
        return None
    content = await _query_llm_api(
        "https://api.deepseek.com/chat/completions", "deepseek-chat", key, prompt, timeout=30
    )
    return {"source": "deepseek", "content": content, "citations": []} if content else None


async def _query_glm(prompt: str) -> Optional[Dict]:
    """Query GLM (via zeabur/openai-compatible) for rankings."""
    # Prefer Zeabur openai-compatible relay if configured
    relay = _get_api_key("api/zai")
    if not relay:
        return None
    # NOTE: zeabur endpoint is OpenAI compatible (not open.bigmodel native)
    content = await _query_llm_api(
        "https://open.zeabur.com/v1/chat/completions", "glm-5", relay, prompt
    )
    return {"source": "glm (zai relay)", "content": content, "citations": []} if content else None

async def search_ai_platforms(industry: str, keyword: str) -> List[Dict]:
    """
    向多个 AI 平台查询行业排行榜数据。

    Returns: list of {"source": str, "content": str, "citations": list}
    """
    prompts = [
        f"{industry}行业排行榜前10名公司及其优缺点，请列出具体公司名和简要分析",
        f"{keyword}最好的产品推荐，列出前10名并说明理由",
    ]

    results = []
    # Query each platform with the first prompt (most important)
    main_prompt = prompts[0]
    tasks = [
        _query_perplexity(main_prompt),
        _query_deepseek(main_prompt),
    ]

    # If we still have <2 platforms, fallback to AI-from-search
    if True:
        try:
            fallback = await ai_rank_from_search(industry, keyword)
            if fallback:
                tasks.append(asyncio.sleep(0, result=fallback))
        except Exception:
            pass
    try:
        responses = await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout=25)
    except asyncio.TimeoutError:
        logger.warning("AI 平台查询超时，返回已有结果")
        responses = []

    for resp in responses:
        if isinstance(resp, dict) and resp.get("content"):
            results.append(resp)

    # If we got at least one result, also query with the second prompt on one platform
    if results and len(prompts) > 1:
        supplementary = await _query_deepseek(prompts[1]) 
        if supplementary:
            supplementary["prompt_type"] = "product_recommendation"
            results.append(supplementary)

    logger.info(f"AI 平台查询完成: {len(results)} 个平台返回了数据 ({', '.join(r['source'] for r in results)})")
    return results


def _parse_ai_rankings(ai_results: List[Dict]) -> List[Dict]:
    """
    从 AI 平台的回答中提取结构化排行数据。
    简单解析：提取编号列表中的公司名。
    """
    rankings = []
    seen = set()

    for result in ai_results:
        content = result.get("content", "")
        source = result.get("source", "unknown")

        # 匹配常见排行格式: "1. 公司名" / "第1名：公司名" / "1）公司名"
        patterns = [
            r'(?:^|\n)\s*(?:\d+)[.、)）]\s*\*{0,2}([^*\n:：]+?)\*{0,2}(?:[:：\n—\-]|$)',
            r'(?:^|\n)\s*第\s*\d+\s*名[：:]\s*\*{0,2}([^*\n]+?)\*{0,2}(?:[:：\n—\-]|$)',
        ]
        for pat in patterns:
            for match in re.finditer(pat, content):
                name = match.group(1).strip().rstrip('*').strip()
                # Filter out noise
                if 2 <= len(name) <= 30 and name not in seen:
                    seen.add(name)
                    rankings.append({
                        "name": name,
                        "source": source,
                        "citations": result.get("citations", []),
                    })

    return rankings


async def search_baidu(query: str, num_results: int = 20) -> List[Dict]:
    """百度搜索获取结果（使用Playwright渲染JS）"""
    results = []
    try:
        from playwright.async_api import async_playwright
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page()
            await page.goto(f"https://www.baidu.com/s?wd={query}&rn={num_results}", wait_until="networkidle", timeout=15000)
            await asyncio.sleep(2)
            
            # 提取搜索结果
            items = await page.query_selector_all("#content_left .c-container, #content_left .result")
            for item in items[:num_results]:
                try:
                    title_el = await item.query_selector("h3 a")
                    abstract_el = await item.query_selector(".c-abstract, .content-right_8Zs40, [class*='content']")
                    if title_el:
                        title = await title_el.inner_text()
                        href = await title_el.get_attribute("href") or ""
                        abstract = await abstract_el.inner_text() if abstract_el else ""
                        results.append({"title": title.strip(), "url": href, "abstract": abstract.strip()})
                except Exception:
                    continue
            
            await browser.close()
    except Exception as e:
        logger.error(f"百度搜索失败: {e}")
    
    # Fallback: 使用Bing搜索（不需要JS渲染）
    if not results:
        results = await search_bing(query, num_results)
    
    return results


async def search_bing(query: str, num_results: int = 10) -> List[Dict]:
    """Bing搜索（备选，不需要JS渲染）"""
    results = []
    async with httpx.AsyncClient(headers=HEADERS, follow_redirects=True, timeout=15, trust_env=False) as client:
        try:
            resp = await client.get("https://www.bing.com/search", params={"q": query, "count": num_results})
            soup = BeautifulSoup(resp.text, "html.parser")
            for item in soup.select("#b_results .b_algo"):
                title_el = item.select_one("h2 a")
                abstract_el = item.select_one(".b_caption p")
                if title_el:
                    results.append({
                        "title": title_el.get_text(strip=True),
                        "url": title_el.get("href", ""),
                        "abstract": abstract_el.get_text(strip=True) if abstract_el else "",
                    })
        except Exception as e:
            logger.error(f"Bing搜索失败: {e}")
    return results


async def search_competitors(industry: str, keyword: str, top_n: int = 10) -> List[Dict]:
    """
    搜索行业真实竞品公司
    
    策略：
    1. 先调 AI 平台获取排行榜数据（最权威）
    2. 再调搜索引擎交叉验证
    3. 合并去重，AI 平台结果优先级更高
    """
    # === Phase 1: AI 平台排行榜 ===
    ai_results = []
    ai_rankings = []
    try:
        ai_results = await search_ai_platforms(industry, keyword)
        ai_rankings = _parse_ai_rankings(ai_results)
        logger.info(f"AI 平台提取到 {len(ai_rankings)} 个竞品: {[r['name'] for r in ai_rankings[:5]]}")
    except Exception as e:
        logger.warning(f"AI 平台搜索失败，继续使用搜索引擎: {e}")

    # === Phase 2: 搜索引擎 ===
    queries = [
        f"{industry}排行榜",
        f"{industry}十大品牌",
        f"{industry}哪家好 推荐",
        f"{keyword} 公司排名",
        f"{industry}头部企业",
        f"{industry}市场份额",
    ]
    
    # 使用单个浏览器实例完成所有搜索
    all_results = []
    try:
        from playwright.async_api import async_playwright
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page()
            
            for q in queries:
                try:
                    await page.goto(f"https://www.baidu.com/s?wd={q}", wait_until="domcontentloaded", timeout=10000)
                    # 等待搜索结果出现而非完全加载
                    try:
                        await page.wait_for_selector("#content_left", timeout=5000)
                    except Exception:
                        pass
                    await asyncio.sleep(1)
                    
                    items = await page.query_selector_all("#content_left .c-container, #content_left .result")
                    for item in items[:10]:
                        try:
                            h3 = await item.query_selector("h3 a")
                            abstract_el = await item.query_selector(".c-abstract, [class*='content-right'], span[class*='content']")
                            if h3:
                                title = await h3.inner_text()
                                href = await h3.get_attribute("href") or ""
                                abstract = await abstract_el.inner_text() if abstract_el else ""
                                all_results.append({"title": title.strip(), "url": href, "abstract": abstract.strip(), "query": q})
                        except Exception:
                            continue
                except Exception as e:
                    logger.warning(f"搜索 '{q}' 失败: {e}")
                
                await asyncio.sleep(1)
            
            await browser.close()
    except Exception as e:
        logger.error(f"Playwright搜索失败: {e}")
    
    # Fallback: Bing
    if not all_results:
        for q in queries[:3]:
            results = await search_bing(q)
            all_results.extend(results)
            await asyncio.sleep(1)
    
    logger.info(f"共获取 {len(all_results)} 条搜索结果")
    
    # 从标题和摘要中提取公司名（这里返回原始结果供LLM进一步提取）
    return {
        "industry": industry,
        "keyword": keyword,
        "search_queries": queries,
        "raw_results": all_results[:50],
        "result_count": len(all_results),
        "ai_rankings": ai_rankings,
        "ai_raw_responses": [
            {"source": r["source"], "content": r["content"][:1000], "citations": r.get("citations", [])}
            for r in ai_results
        ],
    }




async def ai_rank_from_search(industry: str, keyword: str) -> Optional[Dict]:
    """Fallback AI platform: use Bing snippets + DeepSeek to synthesize a Top10 ranking."""
    try:
        raw = await search_bing(f"{industry} 排行榜 前十", num_results=8)
        snippets = "\n".join(f"- {r.get('title','')} :: {r.get('abstract','')}" for r in raw)
        prompt = (
            f"你是行业分析师。根据以下搜索结果片段，总结{industry}行业Top10公司名单，并给出每家优缺点(每家1-2条)。\n"
            f"要求：只基于片段，不要编造。输出markdown列表即可。\n\n片段：\n{snippets}"
        )
        ds = await _query_deepseek(prompt)
        if ds and ds.get('content'):
            ds['source'] = 'ai-from-search (deepseek)'
            return ds
    except Exception as e:
        logger.warning(f"ai-from-search fallback 失败: {e}")
    return None
async def research_competitor_details(company_name: str) -> Dict:
    """搜索单个竞品的详细信息"""
    queries = [
        f"{company_name} 产品 优势",
        f"{company_name} 怎么样 评价",
    ]
    
    details = {"company": company_name, "info": []}
    async with httpx.AsyncClient(headers=HEADERS, follow_redirects=True, timeout=15, trust_env=False) as client:
        for q in queries:
            try:
                resp = await client.get("https://www.baidu.com/s", params={"wd": q, "rn": 5})
                soup = BeautifulSoup(resp.text, "html.parser")
                for item in soup.select(".result.c-container"):
                    abstract = item.select_one(".c-abstract, .content-right_8Zs40")
                    if abstract:
                        details["info"].append(abstract.get_text(strip=True))
            except Exception as e:
                logger.error(f"搜索 {q} 失败: {e}")
            await asyncio.sleep(1)
    
    return details


def save_research(project_id: str, data: Dict):
    """保存调研结果"""
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    filepath = DATA_DIR / f"research_{project_id}.json"
    filepath.write_text(json.dumps(data, ensure_ascii=False, indent=2))
    logger.info(f"调研结果已保存: {filepath}")


async def main():
    """CLI入口"""
    import argparse
    parser = argparse.ArgumentParser(description="真实竞品搜索")
    parser.add_argument("--industry", required=True, help="行业名称")
    parser.add_argument("--keyword", default="", help="核心关键词")
    parser.add_argument("--top", type=int, default=10, help="Top N")
    parser.add_argument("--project-id", default="default", help="项目ID")
    args = parser.parse_args()
    
    result = await search_competitors(args.industry, args.keyword or args.industry, args.top)
    save_research(args.project_id, result)
    print(json.dumps(result, ensure_ascii=False, indent=2)[:3000])


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI搜索引擎收录检测模块
检测目标关键词/公司在AI搜索引擎回答中的出现情况。
"""

import asyncio
import json
import sys
from pathlib import Path
from typing import List, Dict, Optional
from datetime import datetime
from loguru import logger

DATA_DIR = Path(__file__).parent.parent / "data"
CHECKS_FILE = DATA_DIR / "checks.json"
COOKIE_BASE = Path.home() / ".playwright-data"

AI_PLATFORMS = {
    "doubao": {
        "name": "豆包",
        "url": "https://www.doubao.com/chat/",
        "input_selector": "textarea",
        "submit_method": "enter",  # enter or click
    },
    "qianwen": {
        "name": "通义千问",
        "url": "https://tongyi.aliyun.com/qianwen/",
        "input_selector": "textarea",
        "submit_method": "enter",
    },
    "deepseek": {
        "name": "DeepSeek",
        "url": "https://chat.deepseek.com/",
        "input_selector": "textarea",
        "submit_method": "enter",
    },
}


def load_checks() -> List[Dict]:
    if CHECKS_FILE.exists():
        return json.loads(CHECKS_FILE.read_text())
    return []


def save_checks(checks: List[Dict]):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    CHECKS_FILE.write_text(json.dumps(checks, ensure_ascii=False, indent=2))


async def check_platform(
    platform_id: str,
    question: str,
    keyword: str,
    company: str,
    headless: bool = True,
) -> Dict:
    """
    检测单个AI平台的收录情况
    
    Returns:
        {
            "platform": str,
            "question": str,
            "answer": str,
            "keyword_found": bool,
            "company_found": bool,
            "success": bool,
            "error": str
        }
    """
    from playwright.async_api import async_playwright
    
    config = AI_PLATFORMS.get(platform_id)
    if not config:
        return {"platform": platform_id, "success": False, "error": "不支持的平台"}
    
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=headless)
            
            # 尝试加载登录态
            cookie_file = COOKIE_BASE / platform_id / "state.json"
            if cookie_file.exists():
                context = await browser.new_context(storage_state=str(cookie_file))
            else:
                context = await browser.new_context()
            
            page = await context.new_page()
            
            # 1. 导航
            await page.goto(config["url"], wait_until="networkidle", timeout=30000)
            await asyncio.sleep(3)
            
            # 2. 输入问题
            try:
                await page.fill(config["input_selector"], question, timeout=10000)
            except Exception:
                # fallback
                textarea = await page.query_selector("textarea")
                if textarea:
                    await textarea.fill(question)
                else:
                    await browser.close()
                    return {"platform": config["name"], "success": False, "error": "输入框未找到"}
            
            await asyncio.sleep(0.5)
            
            # 3. 提交
            await page.keyboard.press("Enter")
            
            # 4. 等待回答（最多60秒）
            logger.info(f"等待 {config['name']} 回答...")
            await asyncio.sleep(15)
            
            # 额外等待：检查是否还在生成
            for _ in range(9):
                # 检查是否有"停止生成"按钮（说明还在生成中）
                stop_btn = await page.query_selector("button:has-text('停止'), button:has-text('Stop')")
                if not stop_btn:
                    break
                await asyncio.sleep(5)
            
            # 5. 获取回答文本
            answer_text = ""
            # 尝试多种选择器获取回答
            selectors = [
                "[class*='answer']",
                "[class*='message']",
                "[class*='response']",
                "[class*='content']",
                "[class*='bubble']",
                "[class*='markdown']",
            ]
            
            for sel in selectors:
                try:
                    elements = await page.query_selector_all(sel)
                    if elements:
                        # 取最后一个（通常是最新的回答）
                        text = await elements[-1].inner_text()
                        if len(text) > len(answer_text):
                            answer_text = text
                except Exception:
                    continue
            
            if not answer_text:
                answer_text = await page.inner_text("body")
            
            await browser.close()
            
            # 6. 检测关键词
            answer_lower = answer_text.lower()
            keyword_found = keyword.lower() in answer_lower
            company_found = company.lower() in answer_lower
            
            logger.info(f"{config['name']}: keyword={keyword_found}, company={company_found}")
            
            return {
                "platform": config["name"],
                "platform_id": platform_id,
                "question": question,
                "answer": answer_text[:2000],
                "keyword_found": keyword_found,
                "company_found": company_found,
                "success": True,
                "error": None,
                "checked_at": datetime.now().isoformat(),
            }
    
    except Exception as e:
        logger.error(f"{platform_id} 检测失败: {e}")
        return {
            "platform": config.get("name", platform_id),
            "platform_id": platform_id,
            "question": question,
            "success": False,
            "keyword_found": False,
            "company_found": False,
            "error": str(e),
            "checked_at": datetime.now().isoformat(),
        }


async def check_all_platforms(
    question: str,
    keyword: str,
    company: str,
    platforms: List[str] = None,
    headless: bool = True,
) -> List[Dict]:
    """检测所有AI平台"""
    if platforms is None:
        platforms = list(AI_PLATFORMS.keys())
    
    results = []
    for pid in platforms:
        result = await check_platform(pid, question, keyword, company, headless)
        results.append(result)
        save_check_record(result)
        await asyncio.sleep(2)
    
    return results


def save_check_record(record: Dict):
    """保存检测记录"""
    checks = load_checks()
    checks.append(record)
    save_checks(checks)


def get_hit_rate(keyword: str = None, company: str = None) -> Dict:
    """计算命中率统计"""
    checks = load_checks()
    if keyword:
        checks = [c for c in checks if c.get("question", "").find(keyword) >= 0]
    
    total = len(checks)
    if total == 0:
        return {"total": 0, "keyword_hit": 0, "company_hit": 0, "rate": 0}
    
    kw_hit = sum(1 for c in checks if c.get("keyword_found"))
    co_hit = sum(1 for c in checks if c.get("company_found"))
    
    return {
        "total": total,
        "keyword_hit": kw_hit,
        "company_hit": co_hit,
        "keyword_rate": round(kw_hit / total * 100, 1),
        "company_rate": round(co_hit / total * 100, 1),
    }


async def main():
    import argparse
    parser = argparse.ArgumentParser(description="AI搜索收录检测")
    parser.add_argument("--keyword", required=True, help="目标关键词")
    parser.add_argument("--company", required=True, help="目标公司名")
    parser.add_argument("--question", help="自定义问题（默认自动生成）")
    parser.add_argument("--platforms", nargs="+", choices=list(AI_PLATFORMS.keys()))
    parser.add_argument("--no-headless", action="store_true")
    parser.add_argument("--stats", action="store_true", help="显示统计")
    args = parser.parse_args()
    
    if args.stats:
        stats = get_hit_rate(args.keyword, args.company)
        print(json.dumps(stats, ensure_ascii=False, indent=2))
        return
    
    question = args.question or f"{args.keyword}哪家好？推荐一下"
    results = await check_all_platforms(
        question, args.keyword, args.company,
        args.platforms, not args.no_headless
    )
    print(json.dumps(results, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
关键词管理和蒸馏模块
管理项目关键词，并通过搜索引擎扩展出长尾问题变体。
"""

import asyncio
import json
import sys
from pathlib import Path
from typing import List, Dict, Optional
from datetime import datetime
from loguru import logger

try:
    import httpx
    from bs4 import BeautifulSoup
except ImportError:
    print("请安装依赖: pip install httpx beautifulsoup4")
    sys.exit(1)

DATA_DIR = Path(__file__).parent.parent / "data"
KEYWORDS_FILE = DATA_DIR / "keywords.json"
PROJECTS_FILE = DATA_DIR / "projects.json"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9",
}


def load_projects() -> List[Dict]:
    if PROJECTS_FILE.exists():
        return json.loads(PROJECTS_FILE.read_text())
    return []


def save_projects(projects: List[Dict]):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    PROJECTS_FILE.write_text(json.dumps(projects, ensure_ascii=False, indent=2))


def load_keywords() -> List[Dict]:
    if KEYWORDS_FILE.exists():
        return json.loads(KEYWORDS_FILE.read_text())
    return []


def save_keywords(keywords: List[Dict]):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    KEYWORDS_FILE.write_text(json.dumps(keywords, ensure_ascii=False, indent=2))


def create_project(name: str, company: str, industry: str, core_keywords: List[str]) -> Dict:
    """创建GEO项目"""
    projects = load_projects()
    project = {
        "id": str(len(projects) + 1),
        "name": name,
        "company": company,
        "industry": industry,
        "core_keywords": core_keywords,
        "created_at": datetime.now().isoformat(),
    }
    projects.append(project)
    save_projects(projects)
    logger.info(f"项目已创建: {name}")
    return project


def add_keywords(project_id: str, keywords: List[str], source: str = "manual") -> int:
    """添加关键词"""
    existing = load_keywords()
    existing_set = {k["keyword"] for k in existing if k["project_id"] == project_id}
    
    added = 0
    for kw in keywords:
        if kw not in existing_set:
            existing.append({
                "project_id": project_id,
                "keyword": kw,
                "source": source,
                "variants": [],
                "created_at": datetime.now().isoformat(),
            })
            added += 1
    
    save_keywords(existing)
    logger.info(f"添加了 {added} 个新关键词")
    return added


async def distill_keywords(keyword: str) -> List[str]:
    """
    关键词蒸馏：通过百度搜索建议和相关搜索扩展长尾关键词
    """
    variants = set()
    
    async with httpx.AsyncClient(headers=HEADERS, follow_redirects=True, timeout=10) as client:
        # 1. 百度搜索建议
        try:
            resp = await client.get("https://suggestion.baidu.com/su", params={"wd": keyword, "cb": "s"})
            text = resp.text
            # 解析 jsonp: s({"q":"xxx","p":false,"s":["a","b","c"]})
            match = text.split('"s":[')[1].split(']')[0] if '"s":[' in text else ""
            if match:
                for s in match.split(','):
                    s = s.strip().strip('"')
                    if s:
                        variants.add(s)
        except Exception as e:
            logger.debug(f"百度建议获取失败: {e}")
        
        # 2. 百度相关搜索
        try:
            resp = await client.get("https://www.baidu.com/s", params={"wd": keyword})
            soup = BeautifulSoup(resp.text, "html.parser")
            for a in soup.select("#rs a, .recommend_list a"):
                text = a.get_text(strip=True)
                if text:
                    variants.add(text)
        except Exception as e:
            logger.debug(f"百度相关搜索获取失败: {e}")
        
        await asyncio.sleep(0.5)
        
        # 3. 问题变体模式
        question_patterns = [
            f"{keyword}哪家好",
            f"{keyword}推荐",
            f"{keyword}排行榜",
            f"{keyword}怎么选",
            f"{keyword}对比",
            f"最好的{keyword}",
            f"{keyword}十大品牌",
        ]
        variants.update(question_patterns)
    
    result = list(variants)
    logger.info(f"关键词 '{keyword}' 蒸馏出 {len(result)} 个变体")
    return result


async def distill_and_save(project_id: str, keyword: str):
    """蒸馏并保存关键词变体"""
    variants = await distill_keywords(keyword)
    
    keywords = load_keywords()
    for kw in keywords:
        if kw["project_id"] == project_id and kw["keyword"] == keyword:
            kw["variants"] = variants
            kw["distilled_at"] = datetime.now().isoformat()
            break
    
    save_keywords(keywords)
    return variants


async def main():
    import argparse
    parser = argparse.ArgumentParser(description="关键词管理")
    sub = parser.add_subparsers(dest="command")
    
    p_create = sub.add_parser("create-project")
    p_create.add_argument("--name", required=True)
    p_create.add_argument("--company", required=True)
    p_create.add_argument("--industry", required=True)
    p_create.add_argument("--keywords", nargs="+", required=True)
    
    p_distill = sub.add_parser("distill")
    p_distill.add_argument("--keyword", required=True)
    p_distill.add_argument("--project-id", default="1")
    
    p_list = sub.add_parser("list")
    p_list.add_argument("--project-id", default=None)
    
    args = parser.parse_args()
    
    if args.command == "create-project":
        project = create_project(args.name, args.company, args.industry, args.keywords)
        add_keywords(project["id"], args.keywords, "core")
        print(json.dumps(project, ensure_ascii=False, indent=2))
    elif args.command == "distill":
        variants = await distill_and_save(args.project_id, args.keyword)
        print(json.dumps(variants, ensure_ascii=False, indent=2))
    elif args.command == "list":
        keywords = load_keywords()
        if args.project_id:
            keywords = [k for k in keywords if k["project_id"] == args.project_id]
        print(json.dumps(keywords, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
多平台文章发布模块
使用Playwright自动化发布文章到各内容平台。
需要预先保存各平台的登录态cookie。
"""

import asyncio
import json
import sys
from pathlib import Path
from typing import Dict, Optional
from datetime import datetime
from loguru import logger

DATA_DIR = Path(__file__).parent.parent / "data"
COOKIE_BASE = Path.home() / ".playwright-data"

PLATFORMS = {
    "zhihu": {
        "name": "知乎",
        "publish_url": "https://zhuanlan.zhihu.com/write",
        "title_selector": "textarea[placeholder*='请输入标题']",
        "editor_selector": ".public-DraftEditor-content",
        "publish_btn": "button:has-text('发布')",
    },
    "baijiahao": {
        "name": "百家号",
        "publish_url": "https://baijiahao.baidu.com/builder/rc/edit?type=news",
        "title_selector": "#title",
        "editor_selector": ".ql-editor",
        "publish_btn": "button:has-text('发布')",
    },
    "sohu": {
        "name": "搜狐号",
        "publish_url": "https://mp.sohu.com/mpfe/v3/main/new-article",
        "title_selector": "input[placeholder*='标题']",
        "editor_selector": ".ql-editor",
        "publish_btn": "button:has-text('发布')",
    },
    "toutiao": {
        "name": "头条号",
        "publish_url": "https://mp.toutiao.com/profile_v4/graphic/publish",
        "title_selector": "textarea[placeholder*='标题']",
        "editor_selector": ".ProseMirror, .ql-editor",
        "publish_btn": "button:has-text('发布')",
    },
}


async def publish_article(
    platform: str,
    title: str,
    content: str,
    headless: bool = True,
) -> Dict:
    """
    发布文章到指定平台
    
    Args:
        platform: 平台ID (zhihu/baijiahao/sohu/toutiao)
        title: 文章标题
        content: 文章内容(纯文本或Markdown)
        headless: 是否无头模式
    
    Returns:
        {"success": bool, "url": str, "error": str}
    """
    from playwright.async_api import async_playwright
    
    config = PLATFORMS.get(platform)
    if not config:
        return {"success": False, "url": None, "error": f"不支持的平台: {platform}"}
    
    cookie_dir = COOKIE_BASE / platform
    if not cookie_dir.exists():
        return {"success": False, "url": None, "error": f"未找到 {config['name']} 的登录态，请先运行登录脚本"}
    
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=headless)
            context = await browser.new_context(storage_state=str(cookie_dir / "state.json"))
            page = await context.new_page()
            
            # 1. 导航到发布页
            await page.goto(config["publish_url"], wait_until="networkidle", timeout=30000)
            await asyncio.sleep(3)
            
            # 2. 检查登录状态
            if "login" in page.url.lower() or "signin" in page.url.lower():
                await browser.close()
                return {"success": False, "url": None, "error": f"{config['name']} 登录态已过期"}
            
            # 3. 填标题
            try:
                await page.fill(config["title_selector"], title, timeout=10000)
            except Exception:
                # fallback: JS方式
                await page.evaluate(f"""
                    document.querySelector("{config['title_selector']}").value = {json.dumps(title)};
                    document.querySelector("{config['title_selector']}").dispatchEvent(new Event('input', {{bubbles: true}}));
                """)
            await asyncio.sleep(1)
            
            # 4. 填内容
            try:
                await page.click(config["editor_selector"])
                await asyncio.sleep(0.5)
                # 用剪贴板粘贴内容（保留格式更好）
                await page.evaluate(f"navigator.clipboard.writeText({json.dumps(content)})")
                await page.keyboard.press("Control+A")
                await asyncio.sleep(0.2)
                await page.keyboard.press("Control+V")
            except Exception:
                await page.fill(config["editor_selector"], content)
            await asyncio.sleep(2)
            
            # 5. 点发布
            try:
                await page.click(config["publish_btn"], timeout=5000)
                await asyncio.sleep(5)
            except Exception as e:
                await browser.close()
                return {"success": False, "url": None, "error": f"发布按钮点击失败: {e}"}
            
            result_url = page.url
            await browser.close()
            
            logger.info(f"✅ {config['name']} 发布完成: {result_url}")
            return {"success": True, "url": result_url, "error": None}
    
    except Exception as e:
        logger.error(f"{platform} 发布失败: {e}")
        return {"success": False, "url": None, "error": str(e)}


async def login_platform(platform: str):
    """交互式登录保存cookie"""
    from playwright.async_api import async_playwright
    
    config = PLATFORMS.get(platform)
    if not config:
        print(f"不支持的平台: {platform}")
        return
    
    cookie_dir = COOKIE_BASE / platform
    cookie_dir.mkdir(parents=True, exist_ok=True)
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context()
        page = await context.new_page()
        
        await page.goto(config["publish_url"])
        print(f"\n请在浏览器中登录 {config['name']}，登录完成后按 Enter 继续...")
        input()
        
        await context.storage_state(path=str(cookie_dir / "state.json"))
        print(f"✅ {config['name']} 登录态已保存到 {cookie_dir}")
        await browser.close()


async def main():
    import argparse
    parser = argparse.ArgumentParser(description="多平台发布")
    sub = parser.add_subparsers(dest="command")
    
    p_pub = sub.add_parser("publish")
    p_pub.add_argument("--platform", required=True, choices=list(PLATFORMS.keys()))
    p_pub.add_argument("--title", required=True)
    p_pub.add_argument("--content-file", required=True, help="内容文件路径")
    p_pub.add_argument("--no-headless", action="store_true")
    
    p_login = sub.add_parser("login")
    p_login.add_argument("--platform", required=True, choices=list(PLATFORMS.keys()))
    
    args = parser.parse_args()
    
    if args.command == "publish":
        content = Path(args.content_file).read_text()
        result = await publish_article(args.platform, args.title, content, not args.no_headless)
        print(json.dumps(result, ensure_ascii=False, indent=2))
    elif args.command == "login":
        await login_platform(args.platform)


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GEO数据报表模块
生成各维度的GEO效果报表。
"""

import json
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from collections import defaultdict

DATA_DIR = Path(__file__).parent.parent / "data"


def load_json(filename: str) -> list:
    f = DATA_DIR / filename
    return json.loads(f.read_text()) if f.exists() else []


def weekly_report(project_id: str = None) -> str:
    """生成周报"""
    articles = load_json("articles.json")
    checks = load_json("checks.json")
    keywords = load_json("keywords.json")
    
    week_ago = (datetime.now() - timedelta(days=7)).isoformat()
    
    # 本周文章
    week_articles = [a for a in articles if a.get("created_at", "") >= week_ago]
    if project_id:
        week_articles = [a for a in week_articles if a.get("project_id") == project_id]
    
    published = [a for a in week_articles if a.get("status") == "published"]
    
    # 本周检测
    week_checks = [c for c in checks if c.get("checked_at", "") >= week_ago]
    
    # 命中率
    total_checks = len(week_checks)
    kw_hits = sum(1 for c in week_checks if c.get("keyword_found"))
    co_hits = sum(1 for c in week_checks if c.get("company_found"))
    
    # 平台分布
    platform_stats = defaultdict(lambda: {"total": 0, "kw_hit": 0, "co_hit": 0})
    for c in week_checks:
        pid = c.get("platform_id", c.get("platform", "unknown"))
        platform_stats[pid]["total"] += 1
        if c.get("keyword_found"):
            platform_stats[pid]["kw_hit"] += 1
        if c.get("company_found"):
            platform_stats[pid]["co_hit"] += 1
    
    report = f"""📊 GEO 周报 ({datetime.now().strftime('%Y-%m-%d')})

📝 文章
- 本周生成: {len(week_articles)} 篇
- 已发布: {len(published)} 篇
- 待发布: {len(week_articles) - len(published)} 篇

🔍 收录检测
- 总检测次数: {total_checks}
- 关键词命中: {kw_hits} ({round(kw_hits/max(total_checks,1)*100, 1)}%)
- 公司名命中: {co_hits} ({round(co_hits/max(total_checks,1)*100, 1)}%)

📈 各平台详情"""
    
    for pid, stats in platform_stats.items():
        kw_rate = round(stats["kw_hit"] / max(stats["total"], 1) * 100, 1)
        co_rate = round(stats["co_hit"] / max(stats["total"], 1) * 100, 1)
        report += f"\n- {pid}: 检测{stats['total']}次, 关键词{kw_rate}%, 公司{co_rate}%"
    
    report += f"""

📋 关键词库
- 总关键词数: {len(keywords)}
"""
    
    return report


def keyword_trend(keyword: str) -> str:
    """某关键词的收录趋势"""
    checks = load_json("checks.json")
    relevant = [c for c in checks if keyword.lower() in c.get("question", "").lower()]
    
    if not relevant:
        return f"暂无关键词 '{keyword}' 的检测数据"
    
    # 按日期分组
    daily = defaultdict(lambda: {"total": 0, "kw_hit": 0, "co_hit": 0})
    for c in relevant:
        date = c.get("checked_at", "")[:10]
        daily[date]["total"] += 1
        if c.get("keyword_found"):
            daily[date]["kw_hit"] += 1
        if c.get("company_found"):
            daily[date]["co_hit"] += 1
    
    report = f"📈 关键词 '{keyword}' 收录趋势\n\n"
    for date in sorted(daily.keys()):
        d = daily[date]
        report += f"{date}: 检测{d['total']}次 | 关键词{d['kw_hit']}次 | 公司{d['co_hit']}次\n"
    
    return report


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--type", choices=["weekly", "keyword-trend"], default="weekly")
    parser.add_argument("--project-id", default=None)
    parser.add_argument("--keyword", default=None)
    args = parser.parse_args()
    
    if args.type == "weekly":
        print(weekly_report(args.project_id))
    elif args.type == "keyword-trend":
        if not args.keyword:
            print("需要 --keyword 参数")
        else:
            print(keyword_trend(args.keyword))