TikTok Ads MCP Server

"""Performance analytics tools for TikTok Ads MCP server."""

from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional

from ..tiktok_client import TikTokAdsClient


class PerformanceTools:
    """Tools for retrieving TikTok Ads performance data and analytics."""
    
    def __init__(self, client: TikTokAdsClient):
        self.client = client
    
    def _get_date_range(self, date_range: str) -> tuple[str, str]:
        """Convert date range string to start and end dates.
        
        Args:
            date_range: Date range string (today, yesterday, last_7_days, etc.)
            
        Returns:
            Tuple of (start_date, end_date) in YYYY-MM-DD format
        """
        today = datetime.now()
        
        if date_range == "today":
            start_date = end_date = today.strftime("%Y-%m-%d")
        elif date_range == "yesterday":
            yesterday = today - timedelta(days=1)
            start_date = end_date = yesterday.strftime("%Y-%m-%d")
        elif date_range == "last_7_days":
            start_date = (today - timedelta(days=7)).strftime("%Y-%m-%d")
            end_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
        elif date_range == "last_14_days":
            start_date = (today - timedelta(days=14)).strftime("%Y-%m-%d")
            end_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
        elif date_range == "last_30_days":
            start_date = (today - timedelta(days=30)).strftime("%Y-%m-%d")
            end_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
        else:
            # Default to last 7 days
            start_date = (today - timedelta(days=7)).strftime("%Y-%m-%d")
            end_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
        
        return start_date, end_date
    

    async def get_campaign_performance(
        self,
        campaign_ids: List[str],
        date_range: str = "last_7_days",
        metrics: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Get performance metrics for campaigns.
        
        Args:
            campaign_ids: List of campaign IDs to analyze
            date_range: Date range for performance data
            metrics: Specific metrics to include (optional)
            
        Returns:
            Campaign performance data
        """
        try:
            start_date, end_date = self._get_date_range(date_range)
            dimensions = [
                "campaign_id"
            ]
            
            # Default metrics if not specified
            common_metrics = ["campaign_name"]
            if not metrics:
                metrics = [
                    "impressions", "clicks", "conversion", "spend",
                    "ctr", "cpm", "cpc", "conversion_rate_v2"
                ]
            else:
                rewrite_metrics = {
                    "conversions": "conversion"
                }
                metrics = [rewrite_metrics.get(m, m) for m in metrics]
            
            result = await self.client.get_performance_data(
                level="AUCTION_CAMPAIGN",
                entity_ids=campaign_ids,
                metrics=common_metrics+metrics,
                dimensions=dimensions,
                start_date=start_date,
                end_date=end_date,
            )
            
            performance_data = [item.get("metrics", {}) for item in  result.get("data", {}).get("list", [])]
            total_metrics = result.get("data", {}).get("total_metrics", {})
            
            
            return {
                "success": True,
                "date_range": f"{start_date} to {end_date}",
                "campaigns": performance_data,
                "totals": total_metrics,
                "campaign_count": len(performance_data),
                "message": f"Retrieved performance data for {len(performance_data)} campaigns"
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "campaign_ids": campaign_ids,
                "date_range": date_range,
                "message": "Failed to retrieve campaign performance data"
            }
    
    async def get_adgroup_performance(
        self,
        adgroup_ids: List[str],
        date_range: str = "last_7_days",
        metrics: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Get performance metrics for ad groups.
        
        Args:
            adgroup_ids: List of ad group IDs to analyze
            date_range: Date range for performance data
            breakdowns: Data breakdowns (age, gender, country, placement)
            
        Returns:
            Ad group performance data
        """
        try:
            start_date, end_date = self._get_date_range(date_range)
            dimensions = [
                "adgroup_id"
            ]
            # Standard metrics for ad group analysis
            common_metrics = ["campaign_id", "campaign_name", "adgroup_name"]
            if not metrics:
                metrics = [
                    "impressions", "clicks", "conversion", "spend",
                    "ctr", "cpm", "cpc", "conversion_rate_v2"
                ]
            else:
                rewrite_metrics = {
                    "conversions": "conversion"
                }
                metrics = [rewrite_metrics.get(m, m) for m in metrics]
            result = await self.client.get_performance_data(
                level="AUCTION_ADGROUP",
                entity_ids=adgroup_ids,
                metrics=common_metrics+metrics,
                start_date=start_date,
                end_date=end_date,
                dimensions=dimensions,
            )
            
            performance_data = [item.get("metrics", {}) for item in  result.get("data", {}).get("list", [])]
            total_metrics = result.get("data", {}).get("total_metrics", {})
            
            
            return {
                "success": True,
                "date_range": f"{start_date} to {end_date}",
                "adgroups": performance_data,
                "totals": total_metrics,
                "adgroup_count": len(performance_data),
                "message": f"Retrieved performance data for {len(performance_data)} ad groups"
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "adgroup_ids": adgroup_ids,
                "date_range": date_range,
                "message": "Failed to retrieve ad group performance data"
            }