""" AI Visibility Tracker - Brand Radar Monitoring ================================================ Purpose: Track brand visibility in AI-generated search answers using Ahrefs Brand Radar APIs. Python: 3.10+ Usage: python ai_visibility_tracker.py --target example.com --json python ai_visibility_tracker.py --target example.com --competitor comp1.com --json python ai_visibility_tracker.py --target example.com --history --json python ai_visibility_tracker.py --target example.com --sov --json """ import argparse import asyncio import json import logging import subprocess import sys from dataclasses import dataclass, field, asdict from datetime import datetime from pathlib import Path from typing import Any # Add parent to path for base_client import sys.path.insert(0, str(Path(__file__).parent)) from base_client import BaseAsyncClient, config logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class ImpressionMetrics: """AI search impression metrics for a brand.""" total: int = 0 trend: str = "stable" # increasing, decreasing, stable change_pct: float = 0.0 period: str = "" breakdown: dict = field(default_factory=dict) @dataclass class MentionMetrics: """AI search mention metrics for a brand.""" total: int = 0 trend: str = "stable" change_pct: float = 0.0 period: str = "" breakdown: dict = field(default_factory=dict) @dataclass class SovMetric: """Share of Voice metric for a single domain.""" domain: str = "" sov_pct: float = 0.0 change_pct: float = 0.0 @dataclass class HistoryPoint: """Single data point in a time series.""" date: str = "" value: float = 0.0 @dataclass class CompetitorVisibility: """Aggregated AI visibility metrics for a competitor domain.""" domain: str = "" impressions: int = 0 mentions: int = 0 sov: float = 0.0 @dataclass class AiVisibilityResult: """Complete AI visibility tracking result.""" target: str = "" impressions: ImpressionMetrics = field(default_factory=ImpressionMetrics) mentions: MentionMetrics = field(default_factory=MentionMetrics) share_of_voice: dict = field(default_factory=dict) impressions_history: list[HistoryPoint] = field(default_factory=list) mentions_history: list[HistoryPoint] = field(default_factory=list) sov_history: list[HistoryPoint] = field(default_factory=list) competitors: list[CompetitorVisibility] = field(default_factory=list) recommendations: list[str] = field(default_factory=list) timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) def to_dict(self) -> dict: """Convert result to dictionary.""" return { "target": self.target, "impressions": asdict(self.impressions), "mentions": asdict(self.mentions), "share_of_voice": self.share_of_voice, "impressions_history": [asdict(h) for h in self.impressions_history], "mentions_history": [asdict(h) for h in self.mentions_history], "sov_history": [asdict(h) for h in self.sov_history], "competitors": [asdict(c) for c in self.competitors], "recommendations": self.recommendations, "timestamp": self.timestamp, } # --------------------------------------------------------------------------- # MCP tool caller helper # --------------------------------------------------------------------------- def call_mcp_tool(tool_name: str, params: dict) -> dict: """ Call an Ahrefs MCP tool and return the parsed JSON response. In Claude Desktop / Claude Code environments the MCP tools are invoked directly by the AI agent. This helper exists so that the script can also be executed standalone via subprocess for testing purposes. """ logger.info(f"Calling MCP tool: {tool_name} with params: {params}") try: cmd = ["claude", "mcp", "call", "ahrefs", tool_name, json.dumps(params)] result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode == 0 and result.stdout.strip(): return json.loads(result.stdout.strip()) logger.warning(f"MCP tool {tool_name} returned non-zero or empty: {result.stderr}") return {} except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as exc: logger.warning(f"MCP call failed ({exc}). Returning empty dict.") return {} # --------------------------------------------------------------------------- # AI Visibility Tracker # --------------------------------------------------------------------------- class AiVisibilityTracker(BaseAsyncClient): """Track brand visibility across AI-generated search results.""" def __init__(self, **kwargs): super().__init__(**kwargs) self.logger = logging.getLogger(self.__class__.__name__) # ---- Impressions ---- async def get_impressions_overview(self, target: str) -> ImpressionMetrics: """Fetch current AI impression metrics via brand-radar-impressions-overview.""" self.logger.info(f"Fetching impressions overview for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-impressions-overview", {"target": target}, ) metrics = ImpressionMetrics() if not data: return metrics metrics.total = data.get("total_impressions", data.get("impressions", 0)) metrics.change_pct = data.get("change_pct", data.get("change", 0.0)) metrics.period = data.get("period", "") metrics.breakdown = data.get("breakdown", {}) if metrics.change_pct > 5: metrics.trend = "increasing" elif metrics.change_pct < -5: metrics.trend = "decreasing" else: metrics.trend = "stable" return metrics # ---- Mentions ---- async def get_mentions_overview(self, target: str) -> MentionMetrics: """Fetch current AI mention metrics via brand-radar-mentions-overview.""" self.logger.info(f"Fetching mentions overview for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-mentions-overview", {"target": target}, ) metrics = MentionMetrics() if not data: return metrics metrics.total = data.get("total_mentions", data.get("mentions", 0)) metrics.change_pct = data.get("change_pct", data.get("change", 0.0)) metrics.period = data.get("period", "") metrics.breakdown = data.get("breakdown", {}) if metrics.change_pct > 5: metrics.trend = "increasing" elif metrics.change_pct < -5: metrics.trend = "decreasing" else: metrics.trend = "stable" return metrics # ---- Share of Voice ---- async def get_sov_overview(self, target: str) -> dict: """Fetch Share of Voice overview via brand-radar-sov-overview.""" self.logger.info(f"Fetching SOV overview for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-sov-overview", {"target": target}, ) if not data: return {"brand_sov": 0.0, "competitors": []} brand_sov = data.get("sov", data.get("share_of_voice", 0.0)) competitors_raw = data.get("competitors", []) competitors = [] for comp in competitors_raw: competitors.append(SovMetric( domain=comp.get("domain", ""), sov_pct=comp.get("sov", comp.get("share_of_voice", 0.0)), change_pct=comp.get("change_pct", 0.0), )) return { "brand_sov": brand_sov, "competitors": [asdict(c) for c in competitors], } # ---- History ---- async def get_impressions_history(self, target: str) -> list[HistoryPoint]: """Fetch impressions history via brand-radar-impressions-history.""" self.logger.info(f"Fetching impressions history for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-impressions-history", {"target": target}, ) return self._parse_history(data) async def get_mentions_history(self, target: str) -> list[HistoryPoint]: """Fetch mentions history via brand-radar-mentions-history.""" self.logger.info(f"Fetching mentions history for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-mentions-history", {"target": target}, ) return self._parse_history(data) async def get_sov_history(self, target: str) -> list[HistoryPoint]: """Fetch SOV history via brand-radar-sov-history.""" self.logger.info(f"Fetching SOV history for {target}") data = await asyncio.to_thread( call_mcp_tool, "brand-radar-sov-history", {"target": target}, ) return self._parse_history(data) def _parse_history(self, data: dict | list) -> list[HistoryPoint]: """Parse history data from MCP response into HistoryPoint list.""" points: list[HistoryPoint] = [] if not data: return points items = data if isinstance(data, list) else data.get("history", data.get("data", [])) for item in items: if isinstance(item, dict): points.append(HistoryPoint( date=item.get("date", item.get("period", "")), value=item.get("value", item.get("impressions", item.get("mentions", item.get("sov", 0.0)))), )) return points # ---- Competitor Comparison ---- async def compare_competitors( self, target: str, competitors: list[str] ) -> list[CompetitorVisibility]: """Aggregate AI visibility metrics for target and competitors.""" self.logger.info(f"Comparing competitors: {competitors}") results: list[CompetitorVisibility] = [] all_domains = [target] + competitors for domain in all_domains: imp = await self.get_impressions_overview(domain) men = await self.get_mentions_overview(domain) sov_data = await self.get_sov_overview(domain) results.append(CompetitorVisibility( domain=domain, impressions=imp.total, mentions=men.total, sov=sov_data.get("brand_sov", 0.0), )) # Sort by SOV descending results.sort(key=lambda x: x.sov, reverse=True) return results # ---- Trend Calculation ---- @staticmethod def calculate_trends(history: list[HistoryPoint]) -> dict: """Determine trend direction and statistics from history data.""" if not history or len(history) < 2: return { "direction": "insufficient_data", "avg_value": 0.0, "min_value": 0.0, "max_value": 0.0, "change_pct": 0.0, "data_points": len(history) if history else 0, } values = [h.value for h in history] first_value = values[0] last_value = values[-1] avg_value = sum(values) / len(values) min_value = min(values) max_value = max(values) if first_value > 0: change_pct = ((last_value - first_value) / first_value) * 100 else: change_pct = 0.0 if change_pct > 10: direction = "strongly_increasing" elif change_pct > 3: direction = "increasing" elif change_pct < -10: direction = "strongly_decreasing" elif change_pct < -3: direction = "decreasing" else: direction = "stable" return { "direction": direction, "avg_value": round(avg_value, 2), "min_value": round(min_value, 2), "max_value": round(max_value, 2), "change_pct": round(change_pct, 2), "data_points": len(values), } # ---- Recommendations ---- @staticmethod def generate_recommendations(result: AiVisibilityResult) -> list[str]: """Generate actionable recommendations for improving AI visibility.""" recs: list[str] = [] # Impression-based recommendations if result.impressions.total == 0: recs.append( "AI 검색에서 브랜드 노출이 감지되지 않았습니다. " "E-E-A-T 시그널(경험, 전문성, 권위성, 신뢰성)을 강화하여 " "AI 엔진이 콘텐츠를 참조할 수 있도록 하세요." ) elif result.impressions.trend == "decreasing": recs.append( "AI 검색 노출이 감소 추세입니다. 최신 콘텐츠 업데이트 및 " "구조화된 데이터(Schema Markup) 추가를 검토하세요." ) elif result.impressions.trend == "increasing": recs.append( "AI 검색 노출이 증가 추세입니다. 현재 콘텐츠 전략을 " "유지하면서 추가 키워드 확장을 고려하세요." ) # Mention-based recommendations if result.mentions.total == 0: recs.append( "AI 응답에서 브랜드 언급이 없습니다. " "브랜드명이 포함된 고품질 콘텐츠를 제작하고, " "외부 사이트에서의 브랜드 언급(Citations)을 늘리세요." ) elif result.mentions.trend == "decreasing": recs.append( "AI 응답 내 브랜드 언급이 줄어들고 있습니다. " "콘텐츠 신선도(Freshness)와 업계 권위 신호를 점검하세요." ) # SOV recommendations sov_value = result.share_of_voice.get("brand_sov", 0.0) if sov_value < 10: recs.append( f"AI 검색 Share of Voice가 {sov_value}%로 낮습니다. " "핵심 키워드에 대한 종합 가이드, FAQ 콘텐츠, " "원본 데이터/연구 자료를 발행하여 인용 가능성을 높이세요." ) elif sov_value < 25: recs.append( f"AI 검색 Share of Voice가 {sov_value}%입니다. " "경쟁사 대비 차별화된 전문 콘텐츠와 " "독점 데이터 기반 인사이트를 강화하세요." ) # Competitor-based recommendations if result.competitors: top_competitor = result.competitors[0] if top_competitor.domain != result.target and top_competitor.sov > sov_value: recs.append( f"최대 경쟁사 {top_competitor.domain}의 SOV가 " f"{top_competitor.sov}%로 앞서고 있습니다. " "해당 경쟁사의 콘텐츠 전략과 인용 패턴을 분석하세요." ) # General best practices recs.append( "AI 검색 최적화를 위해 다음 사항을 지속적으로 점검하세요: " "(1) 구조화된 데이터(JSON-LD) 적용, " "(2) FAQ 및 How-to 콘텐츠 발행, " "(3) 신뢰할 수 있는 외부 사이트에서의 백링크 확보, " "(4) 콘텐츠 정기 업데이트 및 정확성 검증." ) return recs # ---- Main Orchestrator ---- async def track( self, target: str, competitors: list[str] | None = None, include_history: bool = False, include_sov: bool = False, ) -> AiVisibilityResult: """ Orchestrate full AI visibility tracking. Args: target: Domain to track competitors: Optional list of competitor domains include_history: Whether to fetch historical trends include_sov: Whether to include SOV analysis """ self.logger.info(f"Starting AI visibility tracking for {target}") result = AiVisibilityResult(target=target) # Core metrics (always fetched) result.impressions = await self.get_impressions_overview(target) result.mentions = await self.get_mentions_overview(target) # Share of Voice if include_sov or competitors: result.share_of_voice = await self.get_sov_overview(target) # History if include_history: result.impressions_history = await self.get_impressions_history(target) result.mentions_history = await self.get_mentions_history(target) if include_sov: result.sov_history = await self.get_sov_history(target) # Competitor comparison if competitors: result.competitors = await self.compare_competitors(target, competitors) # Generate recommendations result.recommendations = self.generate_recommendations(result) self.print_stats() return result # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: """Build argument parser for CLI usage.""" parser = argparse.ArgumentParser( description="AI Visibility Tracker - Monitor brand visibility in AI search results", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --target example.com --json %(prog)s --target example.com --competitor comp1.com --competitor comp2.com --json %(prog)s --target example.com --history --sov --json %(prog)s --target example.com --output report.json """, ) parser.add_argument( "--target", required=True, help="Target domain to track (e.g., example.com)", ) parser.add_argument( "--competitor", action="append", default=[], help="Competitor domain (repeatable). e.g., --competitor a.com --competitor b.com", ) parser.add_argument( "--history", action="store_true", help="Include historical trend data (impressions, mentions, SOV over time)", ) parser.add_argument( "--sov", action="store_true", help="Include Share of Voice analysis", ) parser.add_argument( "--json", action="store_true", help="Output result as JSON to stdout", ) parser.add_argument( "--output", type=str, default=None, help="Save JSON output to file path", ) return parser def print_summary(result: AiVisibilityResult) -> None: """Print a human-readable summary of AI visibility results.""" print("\n" + "=" * 60) print(f" AI Visibility Report: {result.target}") print("=" * 60) print(f"\n Impressions: {result.impressions.total:,}") print(f" Trend: {result.impressions.trend} ({result.impressions.change_pct:+.1f}%)") print(f"\n Mentions: {result.mentions.total:,}") print(f" Trend: {result.mentions.trend} ({result.mentions.change_pct:+.1f}%)") if result.share_of_voice: sov = result.share_of_voice.get("brand_sov", 0.0) print(f"\n Share of Voice: {sov:.1f}%") comp_list = result.share_of_voice.get("competitors", []) if comp_list: print(" Competitors:") for c in comp_list: print(f" {c.get('domain', '?')}: {c.get('sov_pct', 0):.1f}%") if result.impressions_history: trend_info = AiVisibilityTracker.calculate_trends(result.impressions_history) print(f"\n Impressions Trend: {trend_info['direction']}") print(f" Range: {trend_info['min_value']:,.0f} - {trend_info['max_value']:,.0f}") print(f" Change: {trend_info['change_pct']:+.1f}%") if result.competitors: print("\n Competitor Comparison:") for cv in result.competitors: marker = " <-- target" if cv.domain == result.target else "" print(f" {cv.domain}: SOV={cv.sov:.1f}%, Imp={cv.impressions:,}, Men={cv.mentions:,}{marker}") if result.recommendations: print("\n Recommendations:") for i, rec in enumerate(result.recommendations, 1): print(f" {i}. {rec}") print("\n" + "=" * 60) print(f" Generated: {result.timestamp}") print("=" * 60 + "\n") async def main() -> None: """CLI entry point.""" parser = build_parser() args = parser.parse_args() tracker = AiVisibilityTracker( max_concurrent=5, requests_per_second=2.0, ) result = await tracker.track( target=args.target, competitors=args.competitor if args.competitor else None, include_history=args.history, include_sov=args.sov, ) # Output if args.json or args.output: output_data = result.to_dict() json_str = json.dumps(output_data, ensure_ascii=False, indent=2) if args.json: print(json_str) if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json_str, encoding="utf-8") logger.info(f"Report saved to {args.output}") else: print_summary(result) if __name__ == "__main__": asyncio.run(main())