Meta Ads MCP

"""Insights and Reporting functionality for Meta Ads API."""

import json
from typing import Optional, Union, Dict, List
from .api import meta_api_tool, make_api_request
from .utils import download_image, try_multiple_download_methods, ad_creative_images, create_resource_from_image
from .server import mcp_server
import base64
import datetime


# Prefixes of action_type values that are always redundant duplicates of other
# action types already present in the response.  For every canonical event
# (e.g. "purchase"), the Meta API returns 5-8 variants that carry the exact
# same numeric value:
#   omni_purchase, onsite_web_purchase, onsite_web_app_purchase,
#   web_in_store_purchase, web_app_in_store_purchase,
#   offsite_conversion.fb_pixel_purchase  …
# Removing these cuts each insight row from ~4 KB to ~1 KB without any
# information loss.
_REDUNDANT_ACTION_PREFIXES = (
    "omni_",                       # omnichannel roll-up  (== onsite_web_app_*)
    "onsite_web_app_",             # web+app combined     (== onsite_web_*)
    "onsite_web_",                 # web-only subset      (== canonical + onsite)
    "onsite_app_",                 # app-only subset      (== onsite_conversion.*)
    "web_app_in_store_",           # web+app in-store     (== web_in_store_*)
    "offsite_conversion.fb_pixel_",  # pixel attribution  (== canonical type)
)


def _strip_redundant_actions(row: dict) -> dict:
    """Remove redundant action-type entries from a single insight row."""
    for key in ("actions", "action_values", "cost_per_action_type"):
        items = row.get(key)
        if not isinstance(items, list):
            continue
        row[key] = [
            item for item in items
            if not any(
                item.get("action_type", "").startswith(prefix)
                for prefix in _REDUNDANT_ACTION_PREFIXES
            )
        ]
    return row


@mcp_server.tool()
@meta_api_tool
async def get_insights(object_id: str, access_token: Optional[str] = None,
                      time_range: Union[str, Dict[str, str]] = "maximum", breakdown: str = "",
                      level: str = "ad", limit: int = 25, after: str = "",
                      action_attribution_windows: Optional[List[str]] = None,
                      compact: bool = False) -> str:
    """
    Get performance insights for a campaign, ad set, ad or account.
    
    Args:
        object_id: ID of the campaign, ad set, ad or account
        access_token: Meta API access token (optional - will use cached token if not provided)
        time_range: Either a preset time range string or a dictionary with "since" and "until" dates in YYYY-MM-DD format
                   Preset options: today, yesterday, this_month, last_month, this_quarter, maximum, data_maximum, 
                   last_3d, last_7d, last_14d, last_28d, last_30d, last_90d, last_week_mon_sun, 
                   last_week_sun_sat, last_quarter, last_year, this_week_mon_today, this_week_sun_today, this_year
                   Dictionary example: {"since":"2023-01-01","until":"2023-01-31"}
        breakdown: Optional breakdown dimension. Valid values include:
                   Demographic: age, gender, country, region, dma
                   Platform/Device: device_platform, platform_position, publisher_platform, impression_device
                   Creative Assets: ad_format_asset, body_asset, call_to_action_asset, description_asset, 
                                  image_asset, link_url_asset, title_asset, video_asset, media_asset_url,
                                  media_creator, media_destination_url, media_format, media_origin_url,
                                  media_text_content, media_type, creative_relaxation_asset_type,
                                  flexible_format_asset_type, gen_ai_asset_type
                   Campaign/Ad Attributes: breakdown_ad_objective, breakdown_reporting_ad_id, app_id, product_id
                   Conversion Tracking: coarse_conversion_value, conversion_destination, standard_event_content_type,
                                       signal_source_bucket, is_conversion_id_modeled, fidelity_type, redownload
                   Time-based: hourly_stats_aggregated_by_advertiser_time_zone, 
                              hourly_stats_aggregated_by_audience_time_zone, frequency_value
                   Extensions/Landing: ad_extension_domain, ad_extension_url, landing_destination, 
                                      mdsa_landing_destination
                   Attribution: sot_attribution_model_type, sot_attribution_window, sot_channel, 
                               sot_event_type, sot_source
                   Mobile/SKAN: skan_campaign_id, skan_conversion_id, skan_version, postback_sequence_index
                   CRM/Business: crm_advertiser_l12_territory_ids, crm_advertiser_subvertical_id,
                                crm_advertiser_vertical_id, crm_ult_advertiser_id, user_persona_id, user_persona_name
                   Advanced: hsid, is_auto_advance, is_rendered_as_delayed_skip_ad, mmm, place_page_id,
                            marketing_messages_btn_name, impression_view_time_advertiser_hour_v2, comscore_market,
                            comscore_market_code
        level: Level of aggregation (ad, adset, campaign, account)
        limit: Maximum number of results to return per page (default: 25, Meta API allows much higher values)
        after: Pagination cursor to get the next set of results. Use the 'after' cursor from previous response's paging.next field.
        action_attribution_windows: Optional list of attribution windows (e.g., ["1d_click", "7d_click", "1d_view"]).
                   When specified, actions include additional fields for each window. The 'value' field always shows 7d_click.
        compact: When True, strips redundant action-type duplicates from the response
                 (omni_*, onsite_web_*, offsite_conversion.fb_pixel_*, etc.) to reduce
                 payload size by ~60%. The canonical action types (purchase, add_to_cart,
                 view_content, etc.) are always preserved. Default: False.

    Note on response size: This tool always returns a fixed set of fields (impressions, clicks,
    spend, cpc, cpm, ctr, reach, actions, action_values, etc.) and cannot filter to a subset.
    For large result sets (50+ rows), the actions/action_values arrays can make responses very
    large (1–2MB+). If you only need specific metrics like spend or impressions, consider using
    bulk_get_insights with compact=true and the fields parameter:
        bulk_get_insights(level="ad", account_ids=[...], compact=true, fields=["spend", "impressions"])
    bulk_get_insights supports level="ad", "adset", "campaign", and "account".
    """
    if not object_id:
        return json.dumps({"error": "No object ID provided"}, indent=2)
        
    endpoint = f"{object_id}/insights"
    params = {
        "fields": "account_id,account_name,campaign_id,campaign_name,adset_id,adset_name,ad_id,ad_name,impressions,clicks,spend,cpc,cpm,ctr,reach,frequency,actions,action_values,conversions,unique_clicks,cost_per_action_type",
        "level": level,
        "limit": limit
    }
    
    # Handle time range based on type
    if isinstance(time_range, dict):
        # Use custom date range with since/until parameters
        if "since" in time_range and "until" in time_range:
            params["time_range"] = json.dumps(time_range)
        else:
            return json.dumps({"error": "Custom time_range must contain both 'since' and 'until' keys in YYYY-MM-DD format"}, indent=2)
    else:
        # Use preset date range
        params["date_preset"] = time_range
    
    if breakdown:
        params["breakdowns"] = breakdown
    
    if after:
        params["after"] = after

    if action_attribution_windows:
        # Meta API expects single-quote format: ['1d_click','7d_click']
        params["action_attribution_windows"] = "[" + ",".join(f"'{w}'" for w in action_attribution_windows) + "]"

    data = await make_api_request(endpoint, access_token, params)

    # In compact mode, strip redundant action-type duplicates to reduce response size.
    if compact and isinstance(data, dict):
        for row in data.get("data", []):
            if isinstance(row, dict):
                _strip_redundant_actions(row)

    return json.dumps(data, indent=2)