jonas/meta_api_grabber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: jonas:main
Branches Tags
jonas:main jonas:view_manager_only
..
compare: jonas:10d5bdc518697e328335109a06c0609475631822
Branches Tags
jonas:view_manager_only jonas:main

6 Commits
main ... 10d5bdc518

Author SHA1 Message Date
Jonas Linter
10d5bdc518 Removed the dumb extension stuf 2025-11-12 21:40:35 +00:00
Jonas Linter
5ecf58ba6a Once again missed something 2025-11-12 21:37:45 +00:00
Jonas Linter
5002528eae More views and no longer dropping schema 2025-11-12 21:36:00 +00:00
Jonas Linter
4826c4a744 More new views + timescaledb stuff 2025-11-12 21:19:55 +00:00
Jonas Linter
56e1edb0db Updated views in a few key ways 2025-11-12 14:58:09 +00:00
Jonas Linter
45bc817428 Updated meta_api_grabber to view manager 2025-11-12 13:42:17 +00:00
25 changed files with 816 additions and 4447 deletions
Expand all files Collapse all files

22
Dockerfile
View File

@@ -1,5 +1,5 @@
# Dockerfile for Meta API Grabber # Dockerfile for View Manager
# Production-ready container for scheduled data collection # Production-ready container for database view management
FROM python:3.13-slim FROM python:3.13-slim
@@ -15,26 +15,20 @@ RUN apt-get update && apt-get install -y \
RUN pip install --no-cache-dir uv RUN pip install --no-cache-dir uv
# Copy project files # Copy project files
COPY pyproject.toml uv.lock README.md ./ COPY pyproject.toml README.md ./
COPY src/ ./src/ COPY src/ ./src/
COPY metadata.yaml ./metadata.yaml
# Install Python dependencies using uv # Install Python dependencies using uv
RUN uv pip install --system -e . RUN uv pip install --system -e .
# Copy environment file template (will be overridden by volume mount)
# This is just for documentation - actual .env should be mounted
COPY .env.example .env.example
# Create directory for token metadata
RUN mkdir -p /app/data
# Set environment variables # Set environment variables
ENV PYTHONUNBUFFERED=1 ENV PYTHONUNBUFFERED=1
ENV PYTHONPATH=/app ENV PYTHONPATH=/app
# Health check - verify the script can at least import # Health check - verify database connection
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
CMD python -c "from src.meta_api_grabber.scheduled_grabber import ScheduledInsightsGrabber; print('OK')" || exit 1 CMD python -c "import asyncio; from meta_api_grabber.database import TimescaleDBClient; asyncio.run(TimescaleDBClient().connect())" || exit 1
# Run the scheduled grabber # Run the view manager setup
CMD ["python", "-m", "src.meta_api_grabber.scheduled_grabber"] CMD ["uv", "run", "view-manager-setup"]

16
docker-compose.yml
View File

@@ -19,22 +19,22 @@ services:
retries: 5 retries: 5
restart: unless-stopped restart: unless-stopped
# Meta API Grabber - Scheduled data collection # View Manager - Setup database schema and views
meta-grabber: view-manager:
build: build:
context: . context: .
dockerfile: Dockerfile dockerfile: Dockerfile
container_name: meta_api_grabber container_name: view_manager
environment: environment:
# Database connection (connects to timescaledb service) # Database connection (connects to timescaledb service)
DATABASE_URL: postgresql://meta_user:meta_password@timescaledb:5432/meta_insights DATABASE_URL: postgresql://meta_user:meta_password@timescaledb:5432/meta_insights
env_file: env_file:
- .env # Must contain META_ACCESS_TOKEN, META_APP_ID, META_APP_SECRET - .env # Optional: for any additional configuration
volumes: volumes:
# Mount .env for token updates (auto-refresh will update the file) # Mount metadata.yaml for account configuration
- ./.env:/app/.env - ./metadata.yaml:/app/metadata.yaml:ro
# Mount token metadata file (preserves token refresh state across restarts) # Optional: Mount custom db_schema.sql if you want to edit it externally
- ./.meta_token.json:/app/.meta_token.json # - ./src/meta_api_grabber/db_schema.sql:/app/src/meta_api_grabber/db_schema.sql:ro
depends_on: depends_on:
timescaledb: timescaledb:
condition: service_healthy condition: service_healthy

24
metadata.yaml Normal file
View File

@@ -0,0 +1,24 @@
# Metadata configuration for account mapping
# This file stores important metadata that links different accounts and identifiers
accounts:
- label: "Hotel Bemelmans Post"
meta_account_id: "238334370765317"
google_account_id: "7581209925"
alpinebits_hotel_code: "39054_001"
- label: "Jagdhof-Kaltern"
#meta_account_id: "act_987654321"
google_account_id: "1951919786"
alpinebits_hotel_code: "39052_001"
- label: "Residence Erika"
google_account_id: "6604634947"
alpinebits_hotel_code: "39040_001"
# Add more accounts as needed
# - label: "Your Account Name"
# meta_account_id: "act_xxxxx"
# google_account_id: "xxx-xxx-xxxx"
# alpinebits_hotel_code: "HOTELxxx"

17
pyproject.toml
View File

@@ -1,18 +1,13 @@
[project] [project]
name = "meta-api-grabber" name = "meta-api-grabber"
version = "0.1.0" version = "0.1.0"
description = "Meta Marketing API data grabber with TimescaleDB storage" description = "View manager for TimescaleDB with metadata handling"
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.13"
dependencies = [ dependencies = [
"aiohttp>=3.13.1",
"alembic>=1.17.0",
"asyncpg>=0.30.0", "asyncpg>=0.30.0",
"facebook-business>=23.0.3",
"google-ads>=28.3.0",
"python-dotenv>=1.1.1", "python-dotenv>=1.1.1",
"requests-oauthlib>=2.0.0", "pyyaml>=6.0.0",
"sqlalchemy[asyncio]>=2.0.44",
] ]
[project.optional-dependencies] [project.optional-dependencies]
@@ -22,13 +17,7 @@ test = [
] ]
[project.scripts] [project.scripts]
meta-auth = "meta_api_grabber.auth:main" view-manager-setup = "meta_api_grabber.setup_and_wait:main"
meta-scheduled = "meta_api_grabber.scheduled_grabber:main"
meta-insights = "meta_api_grabber.insights_grabber:main"
meta-test-accounts = "meta_api_grabber.test_ad_accounts:main"
meta-test-leads = "meta_api_grabber.test_page_leads:main"
meta-token = "meta_api_grabber.token_manager:main"
google-ads-test = "meta_api_grabber.test_google_ads_accounts:main"
[build-system] [build-system]
requires = ["hatchling"] requires = ["hatchling"]

291
src/meta_api_grabber/auth.py
View File

@@ -1,291 +0,0 @@
"""
OAuth2 authentication module for Meta/Facebook API.
Handles:
- Initial OAuth2 flow for short-lived tokens
- Exchange short-lived for long-lived tokens (60 days)
- Automatic token refresh
- Token persistence
"""
import json
import os
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Optional
import requests
from dotenv import load_dotenv
from requests_oauthlib import OAuth2Session
from requests_oauthlib.compliance_fixes import facebook_compliance_fix
class MetaOAuth2:
"""Handle OAuth2 authentication flow for Meta/Facebook API."""
def __init__(
self,
app_id: Optional[str] = None,
app_secret: Optional[str] = None,
redirect_uri: str = "https://localhost/",
api_version: str = "v18.0",
):
"""
Initialize OAuth2 handler.
Args:
app_id: Facebook App ID (or set META_APP_ID env var)
app_secret: Facebook App Secret (or set META_APP_SECRET env var)
redirect_uri: OAuth redirect URI
api_version: Facebook API version
"""
load_dotenv()
self.app_id = app_id or os.getenv("META_APP_ID")
self.app_secret = app_secret or os.getenv("META_APP_SECRET")
self.redirect_uri = redirect_uri
self.api_version = api_version
if not self.app_id or not self.app_secret:
raise ValueError(
"App ID and App Secret are required. "
"Provide them as arguments or set META_APP_ID and META_APP_SECRET env vars."
)
self.authorization_base_url = f"https://www.facebook.com/{api_version}/dialog/oauth"
self.token_url = f"https://graph.facebook.com/{api_version}/oauth/access_token"
# Scopes needed for ads insights
self.scopes = ["ads_management", "ads_read"]
def get_authorization_url(self) -> tuple[str, str]:
"""
Get the authorization URL for the user to visit.
Returns:
Tuple of (authorization_url, state)
"""
facebook = OAuth2Session(
self.app_id,
redirect_uri=self.redirect_uri,
scope=self.scopes,
)
facebook = facebook_compliance_fix(facebook)
authorization_url, state = facebook.authorization_url(self.authorization_base_url)
return authorization_url, state
def fetch_token(self, authorization_response: str) -> Dict[str, str]:
"""
Exchange the authorization code for an access token.
Args:
authorization_response: The full redirect URL after authorization
Returns:
Token dictionary containing access_token and other info
"""
facebook = OAuth2Session(
self.app_id,
redirect_uri=self.redirect_uri,
scope=self.scopes,
)
facebook = facebook_compliance_fix(facebook)
token = facebook.fetch_token(
self.token_url,
client_secret=self.app_secret,
authorization_response=authorization_response,
)
return token
def exchange_for_long_lived_token(self, short_lived_token: str) -> Dict[str, any]:
"""
Exchange short-lived token for long-lived token (60 days).
Args:
short_lived_token: The short-lived access token from OAuth flow
Returns:
Dictionary with access_token and expires_in
"""
url = "https://graph.facebook.com/oauth/access_token"
params = {
"grant_type": "fb_exchange_token",
"client_id": self.app_id,
"client_secret": self.app_secret,
"fb_exchange_token": short_lived_token,
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
return {
"access_token": data["access_token"],
"expires_in": data.get("expires_in", 5184000), # Default 60 days
"token_type": data.get("token_type", "bearer"),
}
def get_token_info(self, access_token: str) -> Dict[str, any]:
"""
Get information about an access token including expiry.
Args:
access_token: Access token to inspect
Returns:
Dictionary with token info including expires_at, is_valid, etc.
"""
url = "https://graph.facebook.com/debug_token"
params = {
"input_token": access_token,
"access_token": f"{self.app_id}|{self.app_secret}",
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
return data.get("data", {})
def interactive_auth(self, exchange_for_long_lived: bool = True) -> str:
"""
Run interactive OAuth2 flow to get access token.
Args:
exchange_for_long_lived: If True, exchanges short-lived for long-lived token (60 days)
Returns:
Access token string (long-lived if exchange_for_long_lived=True)
"""
print("\n" + "="*60)
print("META/FACEBOOK OAUTH2 AUTHENTICATION")
print("="*60)
# Get authorization URL
auth_url, state = self.get_authorization_url()
print("\n1. Please visit this URL in your browser:")
print(f"\n {auth_url}\n")
print("2. Authorize the application")
print("3. You will be redirected to localhost (this is expected)")
print("4. Copy the FULL URL from your browser's address bar")
print(" (it will look like: https://localhost/?code=...)")
# Wait for user to paste the redirect URL
redirect_response = input("\nPaste the full redirect URL here: ").strip()
# Fetch the short-lived token
print("\nFetching short-lived access token...")
token = self.fetch_token(redirect_response)
short_lived_token = token["access_token"]
# Exchange for long-lived token
if exchange_for_long_lived:
print("Exchanging for long-lived token (60 days)...")
long_lived_data = self.exchange_for_long_lived_token(short_lived_token)
access_token = long_lived_data["access_token"]
expires_in = long_lived_data["expires_in"]
print(f"\n✅ Long-lived token obtained!")
print(f" Valid for: {expires_in / 86400:.0f} days (~{expires_in / 3600:.0f} hours)")
else:
access_token = short_lived_token
print("\n✅ Short-lived token obtained (valid for ~1-2 hours)")
print("="*60)
# Save token with metadata
self._offer_to_save_token_with_metadata(access_token)
return access_token
def _offer_to_save_token_with_metadata(self, access_token: str):
"""Offer to save the access token with metadata to both .env and JSON file."""
save = input("\nWould you like to save this token? (y/n): ").strip().lower()
if save == "y":
# Get token info
try:
token_info = self.get_token_info(access_token)
expires_at = token_info.get("expires_at", 0)
is_valid = token_info.get("is_valid", False)
issued_at = token_info.get("issued_at", int(time.time()))
except Exception as e:
print(f"Warning: Could not get token info: {e}")
expires_at = int(time.time()) + 5184000 # Assume 60 days
is_valid = True
issued_at = int(time.time())
# Save to .env
env_path = Path(".env")
if env_path.exists():
env_content = env_path.read_text()
else:
env_content = ""
lines = env_content.split("\n")
token_line = f"META_ACCESS_TOKEN={access_token}"
updated = False
for i, line in enumerate(lines):
if line.startswith("META_ACCESS_TOKEN="):
lines[i] = token_line
updated = True
break
if not updated:
lines.append(token_line)
env_path.write_text("\n".join(lines))
print(f"\n✅ Token saved to {env_path}")
# Save metadata to JSON for token refresh logic
token_metadata = {
"access_token": access_token,
"expires_at": expires_at,
"issued_at": issued_at,
"is_valid": is_valid,
"updated_at": int(time.time()),
}
token_file = Path(".meta_token.json")
token_file.write_text(json.dumps(token_metadata, indent=2))
print(f"✅ Token metadata saved to {token_file}")
# Print expiry info
if expires_at:
expires_dt = datetime.fromtimestamp(expires_at)
days_until_expiry = (expires_dt - datetime.now()).days
print(f"\n📅 Token expires: {expires_dt.strftime('%Y-%m-%d %H:%M:%S')}")
print(f" ({days_until_expiry} days from now)")
else:
print(f"\nAccess token: {access_token}")
print("You can manually add it to .env as META_ACCESS_TOKEN")
def main():
"""Run interactive OAuth2 flow."""
try:
oauth = MetaOAuth2()
access_token = oauth.interactive_auth()
print(f"\nYou can now use this access token with the insights grabber.")
except ValueError as e:
print(f"\nConfiguration error: {e}")
print("\nPlease set META_APP_ID and META_APP_SECRET in .env")
return 1
except Exception as e:
print(f"\nError: {e}")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
exit(main())

35
src/meta_api_grabber/database.py
View File

@@ -85,16 +85,36 @@ class TimescaleDBClient:
# Execute schema statement by statement for better error handling # Execute schema statement by statement for better error handling
async with self.pool.acquire() as conn: async with self.pool.acquire() as conn:
statements = [s.strip() for s in schema_sql.split(';') if s.strip()] # Set a reasonable timeout for schema operations (5 minutes per statement)
await conn.execute("SET statement_timeout = '300s'")
# Split statements and track their line numbers in the original file
statements = []
current_line = 1
for stmt in schema_sql.split(';'):
stmt_stripped = stmt.strip()
if stmt_stripped:
# Count newlines before this statement to get starting line
statements.append((stmt_stripped, current_line))
# Update line counter (count newlines in the original chunk including ';')
current_line += stmt.count('\n') + 1
errors = [] errors = []
compression_warnings = [] compression_warnings = []
for i, stmt in enumerate(statements, 1): for i, (stmt, line_num) in enumerate(statements, 1):
if not stmt: if not stmt:
continue continue
# Show progress for potentially slow operations
stmt_lower = stmt.lower()
if any(keyword in stmt_lower for keyword in ['refresh materialized view', 'drop schema', 'create index']):
print(f" Executing statement {i} (line {line_num})...")
try: try:
await conn.execute(stmt) await conn.execute(stmt)
except asyncio.TimeoutError:
errors.append((i, line_num, "Statement execution timed out (>5 minutes)"))
except Exception as stmt_error: except Exception as stmt_error:
error_msg = str(stmt_error).lower() error_msg = str(stmt_error).lower()
@@ -104,19 +124,22 @@ class TimescaleDBClient:
continue continue
elif "columnstore not enabled" in error_msg: elif "columnstore not enabled" in error_msg:
# Track compression warnings separately # Track compression warnings separately
compression_warnings.append(i) compression_warnings.append((i, line_num))
elif "'nonetype' object has no attribute 'decode'" in error_msg: elif "'nonetype' object has no attribute 'decode'" in error_msg:
# Silently ignore decode errors (usually comments/extensions) # Silently ignore decode errors (usually comments/extensions)
continue continue
elif "canceling statement due to statement timeout" in error_msg:
# Handle PostgreSQL timeout errors
errors.append((i, line_num, "Statement execution timed out (>5 minutes)"))
else: else:
# Real errors # Real errors
errors.append((i, stmt_error)) errors.append((i, line_num, stmt_error))
# Report results # Report results
if errors: if errors:
print(f"⚠️ {len(errors)} error(s) during schema initialization:") print(f"⚠️ {len(errors)} error(s) during schema initialization:")
for stmt_num, error in errors: for stmt_num, line_num, error in errors:
print(f" Statement {stmt_num}: {error}") print(f" Statement {stmt_num} (line {line_num}): {error}")
if compression_warnings: if compression_warnings:
print("ℹ️ Note: Data compression not available (TimescaleDB columnstore not enabled)") print("ℹ️ Note: Data compression not available (TimescaleDB columnstore not enabled)")

750
src/meta_api_grabber/db_schema.sql
View File

@@ -1,363 +1,451 @@
-- TimescaleDB Schema for Meta Ad Insights
-- This schema is optimized for time-series data collection and dashboard queries
-- Enable TimescaleDB extension (run as superuser)
-- CREATE EXTENSION IF NOT EXISTS timescaledb; -- Cleanup schema public first.
---DROP SCHEMA IF EXISTS public CASCADE;
-- Recreate schema public.
CREATE SCHEMA public if not exists;
-- Set ownership to meta_user
ALTER SCHEMA public OWNER TO meta_user;
-- ============================================================================ -- ============================================================================
-- MIGRATIONS (Add new columns to existing tables) -- EXTENSIONS
-- ============================================================================ -- ============================================================================
-- Create TimescaleDB extension if it doesn't exist
-- This provides time_bucket() and other time-series functions
-- Add date_start and date_stop columns (idempotent - safe to run multiple times) --CREATE EXTENSION IF NOT EXISTS timescaledb;
ALTER TABLE IF EXISTS account_insights ADD COLUMN IF NOT EXISTS date_start DATE;
ALTER TABLE IF EXISTS account_insights ADD COLUMN IF NOT EXISTS date_stop DATE;
ALTER TABLE IF EXISTS campaign_insights ADD COLUMN IF NOT EXISTS date_start DATE;
ALTER TABLE IF EXISTS campaign_insights ADD COLUMN IF NOT EXISTS date_stop DATE;
ALTER TABLE IF EXISTS campaign_insights ADD COLUMN IF NOT EXISTS frequency NUMERIC(10, 4);
ALTER TABLE IF EXISTS campaign_insights ADD COLUMN IF NOT EXISTS cpp NUMERIC(10, 4);
ALTER TABLE IF EXISTS campaign_insights ADD COLUMN IF NOT EXISTS cost_per_action_type JSONB;
ALTER TABLE IF EXISTS adset_insights ADD COLUMN IF NOT EXISTS date_start DATE;
ALTER TABLE IF EXISTS adset_insights ADD COLUMN IF NOT EXISTS date_stop DATE;
ALTER TABLE IF EXISTS campaign_insights_by_country ADD COLUMN IF NOT EXISTS frequency NUMERIC(10, 4);
ALTER TABLE IF EXISTS campaign_insights_by_country ADD COLUMN IF NOT EXISTS cpp NUMERIC(10, 4);
ALTER TABLE IF EXISTS campaign_insights_by_country ADD COLUMN IF NOT EXISTS cost_per_action_type JSONB;
-- ============================================================================ -- ============================================================================
-- METADATA TABLES (Regular PostgreSQL tables for caching) -- METADATA TABLE
-- ============================================================================ -- ============================================================================
-- This table stores account mappings and identifiers
-- Data is loaded from metadata.yaml file
-- Ad Accounts (rarely changes, cached)
CREATE TABLE IF NOT EXISTS ad_accounts ( CREATE TABLE IF NOT EXISTS public.account_metadata (
account_id VARCHAR(50) PRIMARY KEY, id SERIAL PRIMARY KEY,
account_name VARCHAR(255), label VARCHAR(255) NOT NULL,
currency VARCHAR(10), meta_account_id VARCHAR(100),
timezone_name VARCHAR(100), google_account_id VARCHAR(100),
alpinebits_hotel_code VARCHAR(100),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-- Ensure at least one account ID is provided
CONSTRAINT at_least_one_account_id CHECK (
meta_account_id IS NOT NULL OR
google_account_id IS NOT NULL OR
alpinebits_hotel_code IS NOT NULL
)
); );
-- Campaigns (metadata cache) -- Permission grants for Grafana user
CREATE TABLE IF NOT EXISTS campaigns (
campaign_id VARCHAR(50) PRIMARY KEY,
account_id VARCHAR(50) REFERENCES ad_accounts(account_id),
campaign_name VARCHAR(255) NOT NULL,
status VARCHAR(50),
objective VARCHAR(100),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_campaigns_account ON campaigns(account_id); GRANT USAGE ON SCHEMA public TO grafana;
-- Ad Sets (metadata cache) -- Grant SELECT on all existing tables and views in the schema
CREATE TABLE IF NOT EXISTS adsets ( GRANT SELECT ON ALL TABLES IN SCHEMA public TO grafana;
adset_id VARCHAR(50) PRIMARY KEY,
campaign_id VARCHAR(50) REFERENCES campaigns(campaign_id),
adset_name VARCHAR(255) NOT NULL,
status VARCHAR(50),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_adsets_campaign ON adsets(campaign_id); -- Grant SELECT on all future tables and views in the schema
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO grafana;
-- ============================================================================
-- TIME-SERIES TABLES (Hypertables)
-- ============================================================================
-- Account-level insights (time-series data)
CREATE TABLE IF NOT EXISTS account_insights (
time TIMESTAMPTZ NOT NULL,
account_id VARCHAR(50) NOT NULL REFERENCES ad_accounts(account_id),
-- Core metrics
impressions BIGINT,
clicks BIGINT,
spend NUMERIC(12, 2),
reach BIGINT,
frequency NUMERIC(10, 4),
-- Calculated metrics
ctr NUMERIC(10, 6), -- Click-through rate
cpc NUMERIC(10, 4), -- Cost per click
cpm NUMERIC(10, 4), -- Cost per mille (thousand impressions)
cpp NUMERIC(10, 4), -- Cost per reach
-- Actions (stored as JSONB for flexibility)
actions JSONB,
cost_per_action_type JSONB,
-- Metadata
date_preset VARCHAR(50),
date_start DATE, -- Actual start date of the data range from Meta API
date_stop DATE, -- Actual end date of the data range from Meta API
fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-- Composite primary key
PRIMARY KEY (time, account_id)
);
-- Convert to hypertable (partition by time)
SELECT create_hypertable('account_insights', 'time',
if_not_exists => TRUE,
chunk_time_interval => INTERVAL '1 day'
);
-- Create index for efficient queries
CREATE INDEX IF NOT EXISTS idx_account_insights_account_time
ON account_insights (account_id, time DESC);
-- Campaign-level insights (time-series data)
CREATE TABLE IF NOT EXISTS campaign_insights (
time TIMESTAMPTZ NOT NULL,
campaign_id VARCHAR(50) NOT NULL REFERENCES campaigns(campaign_id),
account_id VARCHAR(50) NOT NULL REFERENCES ad_accounts(account_id),
-- Core metrics
impressions BIGINT,
clicks BIGINT,
spend NUMERIC(12, 2),
reach BIGINT,
frequency NUMERIC(10, 4),
-- Calculated metrics
ctr NUMERIC(10, 6),
cpc NUMERIC(10, 4),
cpm NUMERIC(10, 4),
cpp NUMERIC(10, 4), -- Cost per reach
-- Actions
actions JSONB,
cost_per_action_type JSONB,
-- Metadata
date_preset VARCHAR(50),
date_start DATE, -- Actual start date of the data range from Meta API
date_stop DATE, -- Actual end date of the data range from Meta API
fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (time, campaign_id)
);
-- Convert to hypertable
SELECT create_hypertable('campaign_insights', 'time',
if_not_exists => TRUE,
chunk_time_interval => INTERVAL '1 day'
);
CREATE INDEX IF NOT EXISTS idx_campaign_insights_campaign_time
ON campaign_insights (campaign_id, time DESC);
CREATE INDEX IF NOT EXISTS idx_campaign_insights_account_time
ON campaign_insights (account_id, time DESC);
-- Ad Set level insights (time-series data)
CREATE TABLE IF NOT EXISTS adset_insights (
time TIMESTAMPTZ NOT NULL,
adset_id VARCHAR(50) NOT NULL REFERENCES adsets(adset_id),
campaign_id VARCHAR(50) NOT NULL REFERENCES campaigns(campaign_id),
account_id VARCHAR(50) NOT NULL REFERENCES ad_accounts(account_id),
-- Core metrics
impressions BIGINT,
clicks BIGINT,
spend NUMERIC(12, 2),
reach BIGINT,
-- Calculated metrics
ctr NUMERIC(10, 6),
cpc NUMERIC(10, 4),
cpm NUMERIC(10, 4),
-- Actions
actions JSONB,
-- Metadata
date_preset VARCHAR(50),
date_start DATE, -- Actual start date of the data range from Meta API
date_stop DATE, -- Actual end date of the data range from Meta API
fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (time, adset_id)
);
-- Convert to hypertable
SELECT create_hypertable('adset_insights', 'time',
if_not_exists => TRUE,
chunk_time_interval => INTERVAL '1 day'
);
CREATE INDEX IF NOT EXISTS idx_adset_insights_adset_time
ON adset_insights (adset_id, time DESC);
CREATE INDEX IF NOT EXISTS idx_adset_insights_campaign_time
ON adset_insights (campaign_id, time DESC);
CREATE INDEX IF NOT EXISTS idx_adset_insights_account_time
ON adset_insights (account_id, time DESC);
-- Campaign-level insights by country (time-series data)
CREATE TABLE IF NOT EXISTS campaign_insights_by_country (
time TIMESTAMPTZ NOT NULL,
campaign_id VARCHAR(50) NOT NULL REFERENCES campaigns(campaign_id),
account_id VARCHAR(50) NOT NULL REFERENCES ad_accounts(account_id),
country VARCHAR(2) NOT NULL, -- ISO 2-letter country code
-- Core metrics
impressions BIGINT,
clicks BIGINT,
spend NUMERIC(12, 2),
reach BIGINT,
frequency NUMERIC(10, 4),
-- Calculated metrics
ctr NUMERIC(10, 6),
cpc NUMERIC(10, 4),
cpm NUMERIC(10, 4),
cpp NUMERIC(10, 4), -- Cost per reach
-- Actions
actions JSONB,
cost_per_action_type JSONB,
-- Metadata
date_preset VARCHAR(50),
date_start DATE, -- Actual start date of the data range from Meta API
date_stop DATE, -- Actual end date of the data range from Meta API
fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (time, campaign_id, country)
);
-- Convert to hypertable
SELECT create_hypertable('campaign_insights_by_country', 'time',
if_not_exists => TRUE,
chunk_time_interval => INTERVAL '1 day'
);
CREATE INDEX IF NOT EXISTS idx_campaign_insights_by_country_campaign_time
ON campaign_insights_by_country (campaign_id, time DESC);
CREATE INDEX IF NOT EXISTS idx_campaign_insights_by_country_account_time
ON campaign_insights_by_country (account_id, time DESC);
CREATE INDEX IF NOT EXISTS idx_campaign_insights_by_country_country
ON campaign_insights_by_country (country, time DESC);
-- Compression policy for campaign_insights_by_country
SELECT add_compression_policy('campaign_insights_by_country', INTERVAL '7 days', if_not_exists => TRUE);
-- ============================================================================ -- ============================================================================
-- CONTINUOUS AGGREGATES (Pre-computed rollups for dashboards) -- VIEWS
-- ============================================================================ -- ============================================================================
-- All views below reference data in separate schemas (meta, google, etc.)
-- Hourly aggregates for account insights -- Add your view definitions here...
CREATE MATERIALIZED VIEW IF NOT EXISTS account_insights_hourly
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 hour', time) AS bucket,
account_id,
AVG(impressions) as avg_impressions,
AVG(clicks) as avg_clicks,
AVG(spend) as avg_spend,
AVG(ctr) as avg_ctr,
AVG(cpc) as avg_cpc,
AVG(cpm) as avg_cpm,
MAX(reach) as max_reach,
COUNT(*) as sample_count
FROM account_insights
GROUP BY bucket, account_id;
-- Refresh policy: refresh last 2 days every hour
SELECT add_continuous_aggregate_policy('account_insights_hourly',
start_offset => INTERVAL '2 days',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '1 hour',
if_not_exists => TRUE
);
-- Daily aggregates for account insights
CREATE MATERIALIZED VIEW IF NOT EXISTS account_insights_daily
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 day', time) AS bucket,
account_id,
AVG(impressions) as avg_impressions,
SUM(impressions) as total_impressions,
AVG(clicks) as avg_clicks,
SUM(clicks) as total_clicks,
AVG(spend) as avg_spend,
SUM(spend) as total_spend,
AVG(ctr) as avg_ctr,
AVG(cpc) as avg_cpc,
AVG(cpm) as avg_cpm,
MAX(reach) as max_reach,
COUNT(*) as sample_count
FROM account_insights
GROUP BY bucket, account_id;
-- Refresh policy: refresh last 7 days every 4 hours DROP VIEW IF EXISTS campaign_insights CASCADE;
SELECT add_continuous_aggregate_policy('account_insights_daily',
start_offset => INTERVAL '7 days',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '4 hours',
if_not_exists => TRUE
);
-- general campaign insights view
-- ============================================================================ CREATE VIEW campaign_insights AS
-- DATA RETENTION POLICIES (Optional - uncomment to enable) SELECT date_start AS "time",
-- ============================================================================ account_id AS account_id,
campaign_id,
-- Keep raw data for 90 days, then drop campaign_name,
-- SELECT add_retention_policy('account_insights', INTERVAL '90 days', if_not_exists => TRUE); objective,
-- SELECT add_retention_policy('campaign_insights', INTERVAL '90 days', if_not_exists => TRUE);
-- SELECT add_retention_policy('adset_insights', INTERVAL '90 days', if_not_exists => TRUE);
-- Compress data older than 7 days for better storage efficiency
SELECT add_compression_policy('account_insights', INTERVAL '7 days', if_not_exists => TRUE);
SELECT add_compression_policy('campaign_insights', INTERVAL '7 days', if_not_exists => TRUE);
SELECT add_compression_policy('adset_insights', INTERVAL '7 days', if_not_exists => TRUE);
-- ============================================================================
-- HELPER VIEWS FOR DASHBOARDS
-- ============================================================================
-- Latest metrics per account
CREATE OR REPLACE VIEW latest_account_metrics AS
SELECT DISTINCT ON (account_id)
account_id,
time,
impressions, impressions,
clicks, clicks,
spend, spend,
reach,
ctr, ctr,
cpc, cpc,
cpm, cpm,
reach, cpp,
frequency frequency,
FROM account_insights ( SELECT (jsonb_array_elements.value ->> 'value'::text)::numeric AS "numeric"
ORDER BY account_id, time DESC; FROM jsonb_array_elements(customcampaign_insights.actions) jsonb_array_elements(value)
WHERE (jsonb_array_elements.value ->> 'action_type'::text) = 'link_click'::text) AS link_click,
( SELECT (jsonb_array_elements.value ->> 'value'::text)::numeric AS "numeric"
FROM jsonb_array_elements(customcampaign_insights.actions) jsonb_array_elements(value)
WHERE (jsonb_array_elements.value ->> 'action_type'::text) = 'landing_page_view'::text) AS landing_page_view,
( SELECT (jsonb_array_elements.value ->> 'value'::text)::numeric AS "numeric"
FROM jsonb_array_elements(customcampaign_insights.actions) jsonb_array_elements(value)
WHERE (jsonb_array_elements.value ->> 'action_type'::text) = 'lead'::text) AS lead
FROM meta.customcampaign_insights;
-- age and gender
DROP MATERIALIZED VIEW IF EXISTS campaign_insights_by_gender_and_age CASCADE;
CREATE MATERIALIZED VIEW campaign_insights_by_gender_and_age AS
SELECT date_start AS "time",
account_id AS account_id,
campaign_id,
campaign_name,
gender,
age,
impressions,
clicks,
spend,
reach,
frequency,
ctr,
cpc,
cpm,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'link_click') AS link_click,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'landing_page_view') AS landing_page_view,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'lead') AS lead
FROM meta.custom_campaign_gender;
-- Create indexes for efficient querying
CREATE INDEX idx_campaign_insights_by_gender_and_age_date
ON campaign_insights_by_gender_and_age(time);
CREATE UNIQUE INDEX idx_campaign_insights_by_gender_and_age_unique
ON campaign_insights_by_gender_and_age(time, campaign_id, gender, age);
REFRESH MATERIALIZED VIEW CONCURRENTLY campaign_insights_by_gender_and_age;
DROP VIEW IF EXISTS campaign_insights_by_gender CASCADE;
create view campaign_insights_by_gender as
Select time,
sum(clicks) as clicks,
sum(link_click) as link_click,
sum(lead) as lead,
sum(landing_page_view) as landing_page_view,
sum(spend) as spend,
sum(reach) as reach,
sum(impressions) as impressions,
gender,
campaign_id,
campaign_name,
account_id
from campaign_insights_by_gender_and_age
group by time, gender, account_id, campaign_id, campaign_name;
DROP VIEW IF EXISTS campaign_insights_by_age CASCADE;
create view campaign_insights_by_age as
Select time,
sum(clicks) as clicks,
sum(link_click) as link_click,
sum(lead) as lead,
sum(landing_page_view) as landing_page_view,
sum(spend) as spend,
sum(reach) as reach,
sum(impressions) as impressions,
age,
campaign_id,
campaign_name,
account_id
from campaign_insights_by_gender_and_age
group by time, age, account_id, campaign_id, campaign_name;
-- device
DROP VIEW IF EXISTS campaign_insights_by_device_flattened CASCADE;
CREATE VIEW campaign_insights_by_device_flattened AS
SELECT date_start AS "time",
account_id AS account_id,
campaign_id,
device_platform,
impressions,
clicks,
spend,
reach,
frequency,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'link_click') AS link_click,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'landing_page_view') AS landing_page_view,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'lead') AS lead
FROM meta.custom_campaign_device;
-- country
--- campaign insights by country
DROP VIEW IF EXISTS campaign_insights_by_country_flattened CASCADE;
CREATE VIEW campaign_insights_by_country_flattened AS
SELECT date_start AS "time",
account_id AS account_id,
campaign_id,
country,
impressions,
clicks,
spend,
reach,
frequency,
ctr,
cpc,
cpm,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'link_click') AS link_click,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'landing_page_view') AS landing_page_view,
(SELECT (value->>'value')::numeric
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'lead') AS lead
FROM meta.custom_campaign_country;
-- ============================================================================
-- account views
DROP VIEW IF EXISTS account_insights_by_gender CASCADE;
CREATE VIEW account_insights_by_gender AS
SELECT
time,
account_id,
gender,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
SUM(link_click) AS link_click,
SUM(landing_page_view) AS landing_page_view,
SUM(lead) AS lead
FROM campaign_insights_by_gender
GROUP BY time, account_id, gender;
DROP VIEW IF EXISTS account_insights_by_device CASCADE;
CREATE VIEW account_insights_by_device AS
SELECT
time,
account_id,
device_platform,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
SUM(link_click) AS link_click,
SUM(landing_page_view) AS landing_page_view,
SUM(lead) AS lead
FROM campaign_insights_by_device_flattened
GROUP BY time, account_id, device_platform;
DROP VIEW IF EXISTS account_insights_by_age CASCADE;
CREATE VIEW account_insights_by_age AS
SELECT
time,
account_id,
age,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
SUM(link_click) AS link_click,
SUM(landing_page_view) AS landing_page_view,
SUM(lead) AS lead
FROM campaign_insights_by_age
GROUP BY time, account_id, age;
DROP VIEW IF EXISTS account_insights_by_gender_and_age CASCADE;
CREATE VIEW account_insights_by_gender_and_age AS
SELECT
time,
account_id,
gender,
age,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
SUM(link_click) AS link_click,
SUM(landing_page_view) AS landing_page_view,
SUM(lead) AS lead
FROM campaign_insights_by_gender_and_age
GROUP BY time, account_id, age, gender;
DROP VIEW IF EXISTS account_insights;
CREATE VIEW account_insights AS
SELECT
time,
account_id,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
SUM(link_click) AS link_click,
SUM(landing_page_view) AS landing_page_view,
SUM(lead) AS lead,
SUM(reach) as reach,
AVG(frequency) as frequency,
avg(cpc) as cpc,
avg(cpm) as cpm,
avg(cpp) as cpp,
avg(ctr) as ctr
FROM campaign_insights
group by time, account_id;
DROP MATERIALIZED VIEW IF EXISTS ads_insights CASCADE;
-- ads view
CREATE MATERIALIZED VIEW ads_insights AS
SELECT
date_start AS time,
name as ad_name,
ins.account_id,
ins.ad_id,
ins.adset_id,
ins.campaign_id,
impressions,
reach,
clicks,
(SELECT SUM((elem.value ->> 'value')::numeric)
FROM jsonb_array_elements(ins.actions) AS elem
WHERE (elem.value ->> 'action_type') = 'link_click') AS link_click,
(SELECT SUM((elem.value ->> 'value')::numeric)
FROM jsonb_array_elements(ins.actions) AS elem
WHERE (elem.value ->> 'action_type') = 'lead') AS lead,
(SELECT sum((value->>'value')::numeric)
FROM jsonb_array_elements(actions)
WHERE value->>'action_type' = 'landing_page_view') AS landing_page_view,
spend,
frequency,
cpc,
cpm,
ctr,
cpp
FROM meta.ads_insights ins
join meta.ads as a on a.id = ins.ad_id;
-- Create indexes for efficient querying
CREATE INDEX idx_ads_insights
ON ads_insights(time, ad_id);
CREATE UNIQUE INDEX ads_insights_unique
ON ads_insights(time, account_id, ad_id);
REFRESH MATERIALIZED VIEW CONCURRENTLY ads_insights;
DROP VIEW IF EXISTS adset_insights CASCADE;
CREATE VIEW adset_insights AS
SELECT
time,
account_id,
adset_id,
campaign_id,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
sum(link_click) as link_click,
sum(lead) as lead,
sum(landing_page_view) as landing_page_view,
SUM(spend) AS spend,
sum(reach),
AVG(frequency) as frequency,
avg(cpc) as cpc,
avg(cpm) as cpm,
avg(cpp) as cpp,
avg(ctr) as ctr
FROM ads_insights
group by time, account_id, adset_id, campaign_id;
DROP VIEW IF EXISTS g_account_insights CASCADE;
CREATE VIEW g_account_insights AS
SELECT
time,
account_id,
clicks,
impressions,
interactions,
cost_micros,
cost_micros / 1000000.0 as cost,
leads,
engagements,
customer_currency_code,
account_name,
-- CTR (Click-Through Rate)
(clicks::numeric / impressions_nz) * 100 as ctr,
-- CPM (Cost Per Mille) in micros and standard units
(cost_micros::numeric / impressions_nz) * 1000 as cpm_micros,
(cost_micros::numeric / impressions_nz) * 1000 / 1000000.0 as cpm,
-- CPC (Cost Per Click) in micros and standard units
cost_micros::numeric / clicks_nz as cpc_micros,
cost_micros::numeric / clicks_nz / 1000000.0 as cpc,
-- CPL (Cost Per Lead) in micros and standard units
cost_micros::numeric / leads_nz as cpl_micros,
cost_micros::numeric / leads_nz / 1000000.0 as cpl,
-- Conversion Rate
(leads::numeric / clicks_nz) * 100 as conversion_rate,
-- Engagement Rate
(engagements::numeric / impressions_nz) * 100 as engagement_rate
FROM (
SELECT
segments_date as time,
customer_id as account_id,
sum(metrics_clicks) as clicks,
sum(metrics_impressions) as impressions,
sum(metrics_interactions) as interactions,
sum(metrics_cost_micros) as cost_micros,
sum(metrics_conversions) as leads,
sum(metrics_engagements) as engagements,
customer_currency_code,
customer_descriptive_name as account_name,
-- Null-safe denominators
NULLIF(sum(metrics_clicks), 0) as clicks_nz,
NULLIF(sum(metrics_impressions), 0) as impressions_nz,
NULLIF(sum(metrics_conversions), 0) as leads_nz
FROM google.account_performance_report
GROUP BY account_id, time, customer_currency_code, account_name
) base;
-- Campaign performance summary (last 24 hours)
CREATE OR REPLACE VIEW campaign_performance_24h AS
SELECT
c.campaign_id,
c.campaign_name,
c.account_id,
SUM(ci.impressions) as total_impressions,
SUM(ci.clicks) as total_clicks,
SUM(ci.spend) as total_spend,
AVG(ci.ctr) as avg_ctr,
AVG(ci.cpc) as avg_cpc
FROM campaigns c
LEFT JOIN campaign_insights ci ON c.campaign_id = ci.campaign_id
WHERE ci.time >= NOW() - INTERVAL '24 hours'
GROUP BY c.campaign_id, c.campaign_name, c.account_id
ORDER BY total_spend DESC;

140
src/meta_api_grabber/discover_accounts.py
View File

@@ -1,140 +0,0 @@
"""
Discover all ad accounts accessible via different methods:
1. User ad accounts (personal access)
2. Business Manager ad accounts (app-level access)
"""
import asyncio
import os
from dotenv import load_dotenv
from facebook_business.adobjects.user import User
from facebook_business.adobjects.business import Business
from facebook_business.api import FacebookAdsApi
def discover_accounts():
"""Discover ad accounts through multiple methods."""
load_dotenv()
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
print("❌ Missing required environment variables")
return 1
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
print("="*70)
print("AD ACCOUNT DISCOVERY")
print("="*70)
print()
# Method 1: User ad accounts (what we currently use)
print("METHOD 1: User Ad Accounts (Personal Access)")
print("-"*70)
try:
me = User(fbid='me')
account_fields = ['name', 'account_id']
user_accounts = me.get_ad_accounts(fields=account_fields)
user_account_list = list(user_accounts)
print(f"Found {len(user_account_list)} user ad account(s):")
for acc in user_account_list:
print(f" - {acc.get('name')} ({acc.get('id')})")
print()
except Exception as e:
print(f"❌ Error getting user accounts: {e}\n")
user_account_list = []
# Method 2: Business Manager accounts
print("METHOD 2: Business Manager Ad Accounts (App-Level Access)")
print("-"*70)
try:
me = User(fbid='me')
# Get businesses this user/app has access to
businesses = me.get_businesses(fields=['id', 'name'])
business_list = list(businesses)
if not business_list:
print("No Business Managers found.")
print()
print("ℹ️ To access ad accounts at the app level, you need to:")
print(" 1. Have a Meta Business Manager")
print(" 2. Add your app to the Business Manager")
print(" 3. Grant the app access to ad accounts")
print()
else:
print(f"Found {len(business_list)} Business Manager(s):")
all_business_accounts = []
for biz in business_list:
biz_id = biz.get('id')
biz_name = biz.get('name')
print(f"\n Business: {biz_name} ({biz_id})")
try:
business = Business(fbid=biz_id)
# Get all ad accounts owned/managed by this business
biz_accounts = business.get_owned_ad_accounts(
fields=['id', 'name', 'account_status']
)
biz_account_list = list(biz_accounts)
print(f" Ad Accounts: {len(biz_account_list)}")
for acc in biz_account_list:
print(f" - {acc.get('name')} ({acc.get('id')})")
all_business_accounts.append(acc)
except Exception as e:
print(f" ❌ Error accessing business accounts: {e}")
print()
print(f"Total Business Manager ad accounts: {len(all_business_accounts)}")
except Exception as e:
print(f"❌ Error getting businesses: {e}")
print()
# Method 3: App-level access token (for reference)
print()
print("METHOD 3: App Access Token (NOT recommended for ad accounts)")
print("-"*70)
print("App access tokens can be generated with:")
print(f" curl 'https://graph.facebook.com/oauth/access_token")
print(f" ?client_id={app_id}")
print(f" &client_secret=YOUR_SECRET")
print(f" &grant_type=client_credentials'")
print()
print("⚠️ However, app access tokens CANNOT access ad accounts directly.")
print(" The Marketing API requires user-level permissions for privacy/security.")
print()
# Summary
print("="*70)
print("SUMMARY")
print("="*70)
print(f"User Ad Accounts: {len(user_account_list)}")
print()
print("💡 RECOMMENDATION:")
print(" The current implementation using User.get_ad_accounts() is correct.")
print(" To access MORE ad accounts, you have two options:")
print()
print(" Option 1: Use a System User token from Business Manager")
print(" (grants access to all Business Manager ad accounts)")
print()
print(" Option 2: Have other users authorize your app")
print(" (each user's token will see their ad accounts)")
print()
return 0
if __name__ == "__main__":
exit(discover_accounts())

349
src/meta_api_grabber/insights_grabber.py
View File

@@ -1,349 +0,0 @@
"""
Async script to grab ad insights data from Meta's Marketing API.
Conservative rate limiting to avoid API limits.
"""
import asyncio
import json
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List
from dotenv import load_dotenv
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.adsinsights import AdsInsights
from facebook_business.api import FacebookAdsApi
from .rate_limiter import MetaRateLimiter
class MetaInsightsGrabber:
"""
Async grabber for Meta ad insights with intelligent rate limiting.
Features:
- Monitors x-fb-ads-insights-throttle header
- Auto-throttles when approaching rate limits (>75%)
- Exponential backoff on rate limit errors
- Automatic retries with progressive delays
"""
def __init__(self, access_token: str = None):
"""
Initialize the grabber with credentials from environment.
Args:
access_token: Optional access token. If not provided, will load from env.
"""
load_dotenv()
self.access_token = access_token or os.getenv("META_ACCESS_TOKEN")
self.app_secret = os.getenv("META_APP_SECRET")
self.app_id = os.getenv("META_APP_ID")
self.ad_account_id = os.getenv("META_AD_ACCOUNT_ID")
if not self.access_token:
raise ValueError(
"Access token is required. Either:\n"
"1. Set META_ACCESS_TOKEN in .env, or\n"
"2. Run 'uv run python src/meta_api_grabber/auth.py' to get a token via OAuth2, or\n"
"3. Pass access_token to the constructor"
)
if not all([self.app_secret, self.app_id, self.ad_account_id]):
raise ValueError(
"Missing required environment variables (META_APP_SECRET, META_APP_ID, META_AD_ACCOUNT_ID). "
"Please check your .env file against .env.example"
)
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=self.app_id,
app_secret=self.app_secret,
access_token=self.access_token,
)
self.ad_account = AdAccount(self.ad_account_id)
# Rate limiter with backoff (Meta best practices)
self.rate_limiter = MetaRateLimiter(
base_delay=2.0,
throttle_threshold=75.0,
max_retry_delay=300.0,
max_retries=5,
)
async def _rate_limited_request(self, func, *args, **kwargs):
"""
Execute a request with intelligent rate limiting and backoff.
Monitors x-fb-ads-insights-throttle header and auto-throttles
when usage exceeds 75%. Implements exponential backoff on errors.
"""
return await self.rate_limiter.execute_with_retry(func, *args, **kwargs)
async def get_account_insights(
self,
date_preset: str = "last_7d",
fields: List[str] = None,
) -> Dict[str, Any]:
"""
Get account-level insights for a given time period.
Args:
date_preset: Time period (last_7d, last_14d, last_30d, etc.)
fields: List of fields to retrieve. If None, uses default interesting fields.
Returns:
Dictionary containing insights data
"""
if fields is None:
# Default interesting fields - conservative selection
fields = [
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.cpc,
AdsInsights.Field.cpm,
AdsInsights.Field.ctr,
AdsInsights.Field.reach,
AdsInsights.Field.frequency,
AdsInsights.Field.actions,
AdsInsights.Field.cost_per_action_type,
]
params = {
"date_preset": date_preset,
"level": "account",
}
print(f"Fetching account insights for {date_preset}...")
insights = await self._rate_limited_request(
self.ad_account.get_insights,
fields=fields,
params=params,
)
# Convert to dict
result = {
"account_id": self.ad_account_id,
"date_preset": date_preset,
"insights": [dict(insight) for insight in insights],
"fetched_at": datetime.now().isoformat(),
}
return result
async def get_campaign_insights(
self,
date_preset: str = "last_7d",
limit: int = 10,
) -> Dict[str, Any]:
"""
Get campaign-level insights.
Args:
date_preset: Time period
limit: Maximum number of campaigns to fetch (conservative default)
Returns:
Dictionary containing campaign insights
"""
fields = [
AdsInsights.Field.campaign_name,
AdsInsights.Field.campaign_id,
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.ctr,
AdsInsights.Field.cpc,
]
params = {
"date_preset": date_preset,
"level": "campaign",
"limit": limit,
}
print(f"Fetching campaign insights for {date_preset} (limit: {limit})...")
insights = await self._rate_limited_request(
self.ad_account.get_insights,
fields=fields,
params=params,
)
result = {
"account_id": self.ad_account_id,
"date_preset": date_preset,
"level": "campaign",
"campaigns": [dict(insight) for insight in insights],
"fetched_at": datetime.now().isoformat(),
}
return result
async def get_ad_set_insights(
self,
date_preset: str = "last_7d",
limit: int = 10,
) -> Dict[str, Any]:
"""
Get ad set level insights.
Args:
date_preset: Time period
limit: Maximum number of ad sets to fetch
Returns:
Dictionary containing ad set insights
"""
fields = [
AdsInsights.Field.adset_name,
AdsInsights.Field.adset_id,
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.ctr,
AdsInsights.Field.cpm,
]
params = {
"date_preset": date_preset,
"level": "adset",
"limit": limit,
}
print(f"Fetching ad set insights for {date_preset} (limit: {limit})...")
insights = await self._rate_limited_request(
self.ad_account.get_insights,
fields=fields,
params=params,
)
result = {
"account_id": self.ad_account_id,
"date_preset": date_preset,
"level": "adset",
"ad_sets": [dict(insight) for insight in insights],
"fetched_at": datetime.now().isoformat(),
}
return result
async def grab_all_insights(self, date_preset: str = "last_7d") -> Dict[str, Any]:
"""
Grab all interesting insights (account, campaign, ad set level).
Args:
date_preset: Time period to fetch
Returns:
Dictionary containing all insights
"""
print(f"\nStarting conservative data grab for {date_preset}...")
print(f"Rate limit: {self.request_delay}s between requests\n")
# Fetch sequentially to be conservative
account_insights = await self.get_account_insights(date_preset)
campaign_insights = await self.get_campaign_insights(date_preset, limit=10)
ad_set_insights = await self.get_ad_set_insights(date_preset, limit=10)
result = {
"account": account_insights,
"campaigns": campaign_insights,
"ad_sets": ad_set_insights,
"summary": {
"date_preset": date_preset,
"total_requests": 3,
"fetched_at": datetime.now().isoformat(),
}
}
return result
async def save_insights_to_json(
self,
insights_data: Dict[str, Any],
output_dir: str = "data",
) -> Path:
"""
Save insights data to a JSON file.
Args:
insights_data: The insights data to save
output_dir: Directory to save the file
Returns:
Path to the saved file
"""
# Create output directory if it doesn't exist
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
# Generate filename with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"meta_insights_{timestamp}.json"
filepath = output_path / filename
# Save to JSON with pretty formatting
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
lambda: filepath.write_text(
json.dumps(insights_data, indent=2, default=str)
)
)
print(f"\nData saved to: {filepath}")
print(f"File size: {filepath.stat().st_size / 1024:.2f} KB")
return filepath
async def main():
"""Main entry point for the insights grabber."""
try:
grabber = MetaInsightsGrabber()
# Grab insights for the last 7 days
insights = await grabber.grab_all_insights(date_preset=AdsInsights.DatePreset.last_7d)
# Save to JSON
filepath = await grabber.save_insights_to_json(insights)
# Print summary
print("\n" + "="*50)
print("SUMMARY")
print("="*50)
print(f"Account ID: {grabber.ad_account_id}")
print(f"Date Range: last_7d")
print(f"Output File: {filepath}")
if insights["account"]["insights"]:
account_data = insights["account"]["insights"][0]
print(f"\nAccount Metrics:")
print(f" Impressions: {account_data.get('impressions', 'N/A')}")
print(f" Clicks: {account_data.get('clicks', 'N/A')}")
print(f" Spend: ${account_data.get('spend', 'N/A')}")
print(f" CTR: {account_data.get('ctr', 'N/A')}%")
print(f"\nCampaigns fetched: {len(insights['campaigns']['campaigns'])}")
print(f"Ad Sets fetched: {len(insights['ad_sets']['ad_sets'])}")
print("\n" + "="*50)
except ValueError as e:
print(f"Configuration error: {e}")
return 1
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)

6
src/meta_api_grabber/main.py
View File

@@ -1,6 +0,0 @@
def main():
print("Hello from meta-api-grabber!")
if __name__ == "__main__":
main()

657
src/meta_api_grabber/rate_limiter.py
View File

@@ -1,657 +0,0 @@
"""
Rate limiting and backoff mechanism for Meta Marketing API.
Based on Meta's best practices:
https://developers.facebook.com/docs/marketing-api/insights/best-practices/
https://developers.facebook.com/docs/graph-api/overview/rate-limiting
"""
import asyncio
import json
import logging
import time
from typing import Any, Callable, Dict, List, Optional
from facebook_business.api import FacebookAdsApi
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class MetaRateLimiter:
"""
Rate limiter with exponential backoff for Meta Marketing API.
Features:
- Monitors X-App-Usage header (platform rate limits)
- Monitors X-Ad-Account-Usage header (ad account specific)
- Monitors X-Business-Use-Case-Usage header (business use case limits)
- Monitors x-fb-ads-insights-throttle header (legacy)
- Automatic throttling when usage > 75%
- Exponential backoff on rate limit errors
- Uses reset_time_duration and estimated_time_to_regain_access
- Configurable thresholds
"""
def __init__(
self,
base_delay: float = 2.0,
throttle_threshold: float = 75.0,
max_retry_delay: float = 300.0, # 5 minutes
max_retries: int = 5,
):
"""
Initialize rate limiter.
Args:
base_delay: Base delay between requests in seconds
throttle_threshold: Throttle when usage exceeds this % (0-100)
max_retry_delay: Maximum delay for exponential backoff
max_retries: Maximum number of retries on rate limit errors
"""
self.base_delay = base_delay
self.throttle_threshold = throttle_threshold
self.max_retry_delay = max_retry_delay
self.max_retries = max_retries
# Track current usage percentages from different headers
# X-App-Usage (platform rate limits)
self.app_call_count: float = 0.0
self.app_total_cputime: float = 0.0
self.app_total_time: float = 0.0
# X-Ad-Account-Usage (ad account specific) - tracked per account
# Key: account_id (e.g., "act_123456789"), Value: dict with metrics
self.ad_account_usage: Dict[str, Dict[str, Any]] = {}
# X-Business-Use-Case-Usage (business use case limits)
self.buc_usage: List[Dict[str, Any]] = []
self.estimated_time_to_regain_access: int = 0 # minutes
# Legacy x-fb-ads-insights-throttle
self.legacy_app_usage_pct: float = 0.0
self.legacy_account_usage_pct: float = 0.0
self.last_check_time: float = time.time()
# Stats
self.total_requests: int = 0
self.throttled_requests: int = 0
self.rate_limit_errors: int = 0
def _get_headers(self, response: Any) -> Optional[Dict[str, str]]:
"""
Extract headers from various response object types.
Args:
response: API response object
Returns:
Dictionary of headers or None
"""
# Facebook SDK response object
if hasattr(response, '_headers'):
return response._headers
elif hasattr(response, 'headers'):
return response.headers
elif hasattr(response, '_api_response'):
return getattr(response._api_response, 'headers', None)
return None
def parse_x_app_usage(self, response: Any) -> Dict[str, float]:
"""
Parse X-App-Usage header (Platform rate limits).
Header format: {"call_count": 28, "total_time": 25, "total_cputime": 25}
Args:
response: API response object
Returns:
Dictionary with call_count, total_time, total_cputime
"""
try:
headers = self._get_headers(response)
if headers:
header_value = headers.get('x-app-usage') or headers.get('X-App-Usage', '')
if header_value:
logger.debug(f"X-App-Usage header: {header_value}")
data = json.loads(header_value)
result = {
'call_count': float(data.get('call_count', 0)),
'total_time': float(data.get('total_time', 0)),
'total_cputime': float(data.get('total_cputime', 0)),
}
logger.debug(f"Parsed X-App-Usage: {result}")
return result
except Exception as e:
logger.debug(f"Failed to parse X-App-Usage header: {e}")
return {'call_count': 0.0, 'total_time': 0.0, 'total_cputime': 0.0}
def parse_x_ad_account_usage(self, response: Any) -> Optional[Dict[str, Any]]:
"""
Parse X-Ad-Account-Usage header (Ad account specific limits).
Header format: {
"acc_id_util_pct": 9.67,
"reset_time_duration": 100,
"ads_api_access_tier": "standard_access"
}
Args:
response: API response object
Returns:
Dictionary with metrics, or None if header not present.
To determine account_id, check response object or URL.
"""
try:
headers = self._get_headers(response)
if headers:
header_value = headers.get('x-ad-account-usage') or headers.get('X-Ad-Account-Usage', '')
if header_value:
logger.debug(f"X-Ad-Account-Usage header: {header_value}")
data = json.loads(header_value)
result = {
'acc_id_util_pct': float(data.get('acc_id_util_pct', 0)),
'reset_time_duration': int(data.get('reset_time_duration', 0)),
'ads_api_access_tier': data.get('ads_api_access_tier'),
}
logger.debug(f"Parsed X-Ad-Account-Usage: {result}")
return result
except Exception as e:
logger.debug(f"Failed to parse X-Ad-Account-Usage header: {e}")
return None
def _extract_account_id(self, response: Any) -> Optional[str]:
"""
Extract account ID from response object.
Tries multiple methods to find the account ID from the response.
Args:
response: API response object
Returns:
Account ID string (e.g., "act_123456789") or None
"""
# Try to get account_id from response attributes
if hasattr(response, 'account_id'):
return response.account_id
if hasattr(response, '_data') and isinstance(response._data, dict):
return response._data.get('account_id')
# Try to get from parent object
if hasattr(response, '_parent_object'):
parent = response._parent_object
if hasattr(parent, 'get_id'):
return parent.get_id()
if hasattr(parent, '_data') and isinstance(parent._data, dict):
return parent._data.get('account_id') or parent._data.get('id')
# Try to get from API context
if hasattr(response, '_api_context'):
context = response._api_context
if hasattr(context, 'account_id'):
return context.account_id
return None
def parse_x_business_use_case_usage(self, response: Any) -> List[Dict[str, Any]]:
"""
Parse X-Business-Use-Case-Usage header (Business use case limits).
Header format: {
"business-id": [{
"type": "ads_management",
"call_count": 95,
"total_cputime": 20,
"total_time": 20,
"estimated_time_to_regain_access": 0,
"ads_api_access_tier": "development_access"
}],
...
}
Args:
response: API response object
Returns:
List of usage dictionaries for each business use case
"""
try:
headers = self._get_headers(response)
if headers:
header_value = headers.get('x-business-use-case-usage') or headers.get('X-Business-Use-Case-Usage', '')
if header_value:
logger.debug(f"X-Business-Use-Case-Usage header: {header_value}")
data = json.loads(header_value)
# Flatten the nested structure
all_usage = []
for business_id, use_cases in data.items():
if isinstance(use_cases, list):
for use_case in use_cases:
use_case['business_id'] = business_id
all_usage.append(use_case)
logger.debug(f"Parsed X-Business-Use-Case-Usage: {len(all_usage)} use cases")
return all_usage
except Exception as e:
logger.debug(f"Failed to parse X-Business-Use-Case-Usage header: {e}")
return []
def parse_throttle_header(self, response: Any) -> Dict[str, float]:
"""
Parse x-fb-ads-insights-throttle header from response (legacy).
Header format: {"app_id_util_pct": 25.5, "acc_id_util_pct": 10.0}
Args:
response: API response object
Returns:
Dictionary with app_id_util_pct and acc_id_util_pct
"""
try:
headers = self._get_headers(response)
if headers:
throttle_header = headers.get('x-fb-ads-insights-throttle', '')
if throttle_header:
logger.debug(f"x-fb-ads-insights-throttle header: {throttle_header}")
throttle_data = json.loads(throttle_header)
result = {
'app_id_util_pct': float(throttle_data.get('app_id_util_pct', 0)),
'acc_id_util_pct': float(throttle_data.get('acc_id_util_pct', 0)),
}
logger.debug(f"Parsed x-fb-ads-insights-throttle: {result}")
return result
except Exception as e:
logger.debug(f"Failed to parse x-fb-ads-insights-throttle header: {e}")
return {'app_id_util_pct': 0.0, 'acc_id_util_pct': 0.0}
def update_usage(self, response: Any, account_id: Optional[str] = None):
"""
Update usage statistics from all API response headers.
Parses and updates metrics from:
- X-App-Usage
- X-Ad-Account-Usage (per account)
- X-Business-Use-Case-Usage
- x-fb-ads-insights-throttle (legacy)
Args:
response: API response object
account_id: Optional account ID (e.g., "act_123456789").
If not provided, will attempt to extract from response.
"""
# Parse all headers
app_usage = self.parse_x_app_usage(response)
ad_account_usage = self.parse_x_ad_account_usage(response)
buc_usage = self.parse_x_business_use_case_usage(response)
legacy_throttle = self.parse_throttle_header(response)
# Update X-App-Usage metrics
self.app_call_count = app_usage['call_count']
self.app_total_cputime = app_usage['total_cputime']
self.app_total_time = app_usage['total_time']
# Update X-Ad-Account-Usage metrics (per account)
if ad_account_usage:
# Try to get account_id
if not account_id:
account_id = self._extract_account_id(response)
# Use 'unknown' as fallback if we can't determine account
if not account_id:
account_id = 'unknown'
logger.debug("Could not determine account_id, using 'unknown'")
# Store usage for this account
self.ad_account_usage[account_id] = ad_account_usage
logger.debug(f"Updated ad account usage for {account_id}")
# Update X-Business-Use-Case-Usage metrics
self.buc_usage = buc_usage
# Find the maximum estimated_time_to_regain_access across all use cases
if buc_usage:
self.estimated_time_to_regain_access = max(
(uc.get('estimated_time_to_regain_access', 0) for uc in buc_usage),
default=0
)
# Update legacy metrics
self.legacy_app_usage_pct = legacy_throttle['app_id_util_pct']
self.legacy_account_usage_pct = legacy_throttle['acc_id_util_pct']
self.last_check_time = time.time()
# Log warnings if approaching limits
self._log_rate_limit_warnings()
def _log_rate_limit_warnings(self):
"""Log warnings if any rate limit metric is approaching threshold."""
warnings = []
# Check X-App-Usage metrics
if self.app_call_count > self.throttle_threshold:
warnings.append(f"App call count: {self.app_call_count:.1f}%")
if self.app_total_cputime > self.throttle_threshold:
warnings.append(f"App CPU time: {self.app_total_cputime:.1f}%")
if self.app_total_time > self.throttle_threshold:
warnings.append(f"App total time: {self.app_total_time:.1f}%")
# Check X-Ad-Account-Usage (per account)
for account_id, usage in self.ad_account_usage.items():
acc_pct = usage.get('acc_id_util_pct', 0)
if acc_pct > self.throttle_threshold:
warnings.append(f"Account {account_id}: {acc_pct:.1f}%")
reset_time = usage.get('reset_time_duration', 0)
if reset_time > 0:
warnings.append(f"Resets in {reset_time}s")
# Check X-Business-Use-Case-Usage
for buc in self.buc_usage:
buc_type = buc.get('type', 'unknown')
call_count = buc.get('call_count', 0)
if call_count > self.throttle_threshold:
warnings.append(f"BUC {buc_type}: {call_count:.1f}%")
eta = buc.get('estimated_time_to_regain_access', 0)
if eta > 0:
warnings.append(f"Regain access in {eta} min")
# Check legacy metrics
if self.legacy_app_usage_pct > self.throttle_threshold:
warnings.append(f"Legacy app: {self.legacy_app_usage_pct:.1f}%")
if self.legacy_account_usage_pct > self.throttle_threshold:
warnings.append(f"Legacy account: {self.legacy_account_usage_pct:.1f}%")
if warnings:
logger.warning(f"⚠️ Rate limit warning: {', '.join(warnings)}")
def get_max_usage_pct(self) -> float:
"""
Get the maximum usage percentage across all rate limit metrics.
Returns:
Maximum usage percentage (0-100)
"""
usage_values = [
self.app_call_count,
self.app_total_cputime,
self.app_total_time,
self.legacy_app_usage_pct,
self.legacy_account_usage_pct,
]
# Add ad account usage percentages (per account)
for usage in self.ad_account_usage.values():
usage_values.append(usage.get('acc_id_util_pct', 0))
# Add BUC usage percentages
for buc in self.buc_usage:
usage_values.extend([
buc.get('call_count', 0),
buc.get('total_cputime', 0),
buc.get('total_time', 0),
])
return max(usage_values) if usage_values else 0.0
def should_throttle(self) -> bool:
"""
Check if we should throttle based on current usage.
Returns:
True if usage exceeds threshold
"""
return self.get_max_usage_pct() > self.throttle_threshold
def get_throttle_delay(self) -> float:
"""
Calculate delay based on current usage and reset times.
Uses estimated_time_to_regain_access and reset_time_duration when available.
Returns:
Delay in seconds
"""
max_usage = self.get_max_usage_pct()
if max_usage < self.throttle_threshold:
return self.base_delay
# If we have estimated_time_to_regain_access from BUC header, use it
if self.estimated_time_to_regain_access > 0:
# Convert minutes to seconds and use as delay
delay = self.estimated_time_to_regain_access * 60
logger.info(f"Using BUC estimated_time_to_regain_access: {self.estimated_time_to_regain_access} min ({delay}s)")
return min(delay, self.max_retry_delay)
# Check if any ad account has reset_time_duration and high usage
for account_id, usage in self.ad_account_usage.items():
acc_pct = usage.get('acc_id_util_pct', 0)
reset_time = usage.get('reset_time_duration', 0)
if reset_time > 0 and acc_pct >= 90:
# Use a fraction of reset_time_duration as delay
delay = min(reset_time * 0.5, self.max_retry_delay)
logger.info(f"Using Ad Account {account_id} reset_time_duration: {reset_time}s (delay: {delay}s)")
return delay
# Progressive delay based on usage
# 75% = base_delay, 90% = 2x, 95% = 5x, 99% = 10x
if max_usage >= 95:
multiplier = 10.0
elif max_usage >= 90:
multiplier = 5.0
elif max_usage >= 85:
multiplier = 3.0
else: # 75-85%
multiplier = 2.0
delay = self.base_delay * multiplier
return min(delay, self.max_retry_delay)
async def wait_with_backoff(self, retry_count: int = 0):
"""
Wait with exponential backoff.
Args:
retry_count: Current retry attempt (0-indexed)
"""
if retry_count == 0:
# Normal delay based on throttle
delay = self.get_throttle_delay()
else:
# Exponential backoff: 2^retry * base_delay
delay = min(
(2 ** retry_count) * self.base_delay,
self.max_retry_delay
)
if delay > self.base_delay:
self.throttled_requests += 1
max_usage = self.get_max_usage_pct()
logger.info(f"⏸️ Throttling for {delay:.1f}s (max usage: {max_usage:.1f}%)")
await asyncio.sleep(delay)
async def execute_with_retry(
self,
func: Callable,
*args,
**kwargs
) -> Any:
"""
Execute API call with automatic retry and backoff.
Args:
func: Function to execute (blocking, will be run in executor)
*args: Positional arguments for func
**kwargs: Keyword arguments for func
Returns:
Result from func
Raises:
Exception: If all retries exhausted
"""
self.total_requests += 1
for retry in range(self.max_retries):
try:
# Wait before request (with potential throttling)
await self.wait_with_backoff(retry_count=retry if retry > 0 else 0)
# Execute request in thread pool
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, lambda: func(*args, **kwargs))
# Update usage from response
self.update_usage(result)
return result
except Exception as e:
error_message = str(e).lower()
# Check if it's a rate limit error (expanded list based on Meta docs)
is_rate_limit = (
'rate limit' in error_message or
'too many requests' in error_message or
'throttle' in error_message or
'error code 4' in error_message or # App rate limit
'error code 17' in error_message or # User rate limit
'error code 32' in error_message or # Pages rate limit
'error code 613' in error_message or # Custom rate limit
'error code 80000' in error_message or # Ads Insights BUC
'error code 80001' in error_message or # Pages BUC
'error code 80002' in error_message or # Instagram BUC
'error code 80003' in error_message or # Custom Audience BUC
'error code 80004' in error_message or # Ads Management BUC
'error code 80005' in error_message or # LeadGen BUC
'error code 80006' in error_message or # Messenger BUC
'error code 80008' in error_message or # WhatsApp BUC
'error code 80009' in error_message or # Catalog Management BUC
'error code 80014' in error_message # Catalog Batch BUC
)
if is_rate_limit:
self.rate_limit_errors += 1
if retry < self.max_retries - 1:
backoff_delay = min(
(2 ** (retry + 1)) * self.base_delay,
self.max_retry_delay
)
logger.warning(f"🔄 Rate limit hit! Retrying in {backoff_delay:.1f}s (attempt {retry + 1}/{self.max_retries})")
logger.warning(f" Error: {e}")
await asyncio.sleep(backoff_delay)
continue
else:
logger.error(f"❌ Rate limit error - max retries exhausted: {e}")
raise
# Not a rate limit error, re-raise immediately
raise
# Should never reach here
raise Exception("Max retries exhausted")
def get_stats(self) -> Dict[str, Any]:
"""
Get current rate limiter statistics.
Returns:
Dictionary with comprehensive stats from all headers
"""
return {
# Request stats
'total_requests': self.total_requests,
'throttled_requests': self.throttled_requests,
'rate_limit_errors': self.rate_limit_errors,
# X-App-Usage metrics
'app_call_count': self.app_call_count,
'app_total_cputime': self.app_total_cputime,
'app_total_time': self.app_total_time,
# X-Ad-Account-Usage metrics (per account)
'ad_account_usage': self.ad_account_usage,
# X-Business-Use-Case-Usage metrics
'buc_usage': self.buc_usage,
'estimated_time_to_regain_access': self.estimated_time_to_regain_access,
# Legacy metrics
'legacy_app_usage_pct': self.legacy_app_usage_pct,
'legacy_account_usage_pct': self.legacy_account_usage_pct,
# Computed metrics
'max_usage_pct': self.get_max_usage_pct(),
'is_throttling': self.should_throttle(),
}
def print_stats(self):
"""Print current statistics with all rate limit metrics."""
stats = self.get_stats()
output = []
output.append("\n" + "="*70)
output.append("RATE LIMITER STATISTICS")
output.append("="*70)
# Request stats
output.append(f"Total Requests: {stats['total_requests']}")
output.append(f"Throttled Requests: {stats['throttled_requests']}")
output.append(f"Rate Limit Errors: {stats['rate_limit_errors']}")
output.append("")
# X-App-Usage
output.append("X-App-Usage (Platform Rate Limits):")
output.append(f" Call Count: {stats['app_call_count']:.1f}%")
output.append(f" Total CPU Time: {stats['app_total_cputime']:.1f}%")
output.append(f" Total Time: {stats['app_total_time']:.1f}%")
output.append("")
# X-Ad-Account-Usage (per account)
if stats['ad_account_usage']:
# Only show accounts with data (skip "unknown" accounts with 0 usage)
accounts_to_show = {
account_id: usage
for account_id, usage in stats['ad_account_usage'].items()
if account_id != 'unknown' or usage.get('acc_id_util_pct', 0) > 0
}
if accounts_to_show:
output.append("X-Ad-Account-Usage (Per Account):")
for account_id, usage in accounts_to_show.items():
output.append(f" Account: {account_id}")
output.append(f" Usage: {usage.get('acc_id_util_pct', 0):.1f}%")
output.append(f" Reset Time: {usage.get('reset_time_duration', 0)}s")
output.append(f" API Access Tier: {usage.get('ads_api_access_tier') or 'N/A'}")
output.append("")
# X-Business-Use-Case-Usage
if stats['buc_usage']:
output.append("X-Business-Use-Case-Usage:")
for buc in stats['buc_usage']:
output.append(f" Type: {buc.get('type', 'unknown')}")
output.append(f" Call Count: {buc.get('call_count', 0):.1f}%")
output.append(f" Total CPU Time: {buc.get('total_cputime', 0):.1f}%")
output.append(f" Total Time: {buc.get('total_time', 0):.1f}%")
output.append(f" Est. Time to Regain: {buc.get('estimated_time_to_regain_access', 0)} min")
output.append("")
# Legacy metrics
output.append("Legacy (x-fb-ads-insights-throttle):")
output.append(f" App Usage: {stats['legacy_app_usage_pct']:.1f}%")
output.append(f" Account Usage: {stats['legacy_account_usage_pct']:.1f}%")
output.append("")
# Summary
output.append(f"Max Usage Across All Metrics: {stats['max_usage_pct']:.1f}%")
output.append(f"Currently Throttled: {stats['is_throttling']}")
output.append("="*70 + "\n")
logger.info("\n".join(output))

1227
src/meta_api_grabber/scheduled_grabber.py
View File

File diff suppressed because it is too large Load Diff

305
src/meta_api_grabber/setup_and_wait.py Normal file
View File

@@ -0,0 +1,305 @@
"""
View Manager Setup and Daemon
This script:
1. Initializes database schema from db_schema.sql (drops and recreates public schema with all views)
2. Loads metadata from metadata.yaml into the database
3. Auto-detects materialized views from db_schema.sql
4. Periodically refreshes materialized views (configurable interval)
Usage:
uv run view-manager-setup
Environment variables:
REFRESH_INTERVAL_MINUTES: How often to refresh materialized views (default: 60)
"""
import asyncio
import logging
import os
import re
import signal
import sys
from pathlib import Path
from typing import List
import yaml
from meta_api_grabber.database import TimescaleDBClient
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ViewManagerSetup:
"""Manages database schema setup, metadata loading, and view refresh."""
def __init__(self, refresh_interval_minutes: int = 60):
self.db = TimescaleDBClient()
self.shutdown_event = asyncio.Event()
self.refresh_interval_minutes = refresh_interval_minutes
self.materialized_views: List[str] = []
async def load_metadata_from_yaml(self, yaml_path: Path):
"""
Load account metadata from YAML file and sync to database.
Args:
yaml_path: Path to metadata.yaml file
"""
if not yaml_path.exists():
logger.warning(f"Metadata file not found: {yaml_path}")
logger.info("Skipping metadata loading. Create metadata.yaml to configure accounts.")
return
logger.info(f"Loading metadata from {yaml_path}")
with open(yaml_path, 'r') as f:
data = yaml.safe_load(f)
accounts = data.get('accounts', [])
if not accounts:
logger.warning("No accounts found in metadata.yaml")
return
# Sync accounts to database using upsert logic
async with self.db.pool.acquire() as conn:
# Clear existing metadata (or implement smarter sync logic)
await conn.execute("DELETE FROM public.account_metadata")
for account in accounts:
label = account.get('label')
meta_account_id = account.get('meta_account_id')
google_account_id = account.get('google_account_id')
alpinebits_hotel_code = account.get('alpinebits_hotel_code')
if not label:
logger.warning(f"Skipping account without label: {account}")
continue
await conn.execute(
"""
INSERT INTO public.account_metadata
(label, meta_account_id, google_account_id, alpinebits_hotel_code, updated_at)
VALUES ($1, $2, $3, $4, NOW())
""",
label, meta_account_id, google_account_id, alpinebits_hotel_code
)
logger.info(f"Loaded metadata for: {label}")
logger.info(f"Successfully loaded {len(accounts)} accounts from metadata.yaml")
def detect_materialized_views(self, schema_path: Path) -> List[str]:
"""
Auto-detect materialized views from db_schema.sql.
Parses the SQL file to find:
1. CREATE MATERIALIZED VIEW statements
2. REFRESH MATERIALIZED VIEW statements (explicit refresh requests)
Args:
schema_path: Path to db_schema.sql
Returns:
List of materialized view names (deduplicated, in order of appearance)
"""
if not schema_path.exists():
logger.warning(f"Schema file not found: {schema_path}")
return []
with open(schema_path, 'r') as f:
content = f.read()
views = []
# Find CREATE MATERIALIZED VIEW statements
# Matches: CREATE MATERIALIZED VIEW [IF NOT EXISTS] view_name AS
create_pattern = r'CREATE\s+MATERIALIZED\s+VIEW\s+(?:IF\s+NOT\s+EXISTS\s+)?(\w+)\s+AS'
create_matches = re.findall(create_pattern, content, re.IGNORECASE)
views.extend(create_matches)
# Find REFRESH MATERIALIZED VIEW statements
# Matches: REFRESH MATERIALIZED VIEW [CONCURRENTLY] view_name
refresh_pattern = r'REFRESH\s+MATERIALIZED\s+VIEW\s+(?:CONCURRENTLY\s+)?(?:public\.)?(\w+)'
refresh_matches = re.findall(refresh_pattern, content, re.IGNORECASE)
views.extend(refresh_matches)
# Deduplicate while preserving order
seen = set()
deduplicated = []
for view in views:
if view not in seen:
seen.add(view)
deduplicated.append(view)
if deduplicated:
logger.info(f"Detected {len(deduplicated)} materialized view(s): {', '.join(deduplicated)}")
else:
logger.info("No materialized views detected in db_schema.sql")
return deduplicated
async def refresh_materialized_views(self):
"""Refresh all detected materialized views."""
if not self.materialized_views:
logger.debug("No materialized views to refresh")
return
logger.info(f"Refreshing {len(self.materialized_views)} materialized view(s)...")
for view_name in self.materialized_views:
try:
# Try CONCURRENTLY first (requires unique index)
await self.db.pool.execute(
f"REFRESH MATERIALIZED VIEW CONCURRENTLY public.{view_name}"
)
logger.info(f"✓ Refreshed {view_name} (CONCURRENTLY)")
except Exception as e:
error_msg = str(e).lower()
# If CONCURRENTLY fails, try regular refresh
if 'concurrently' in error_msg or 'unique index' in error_msg:
try:
await self.db.pool.execute(
f"REFRESH MATERIALIZED VIEW public.{view_name}"
)
logger.info(f"✓ Refreshed {view_name}")
except Exception as e2:
logger.error(f"✗ Failed to refresh {view_name}: {e2}")
else:
logger.error(f"✗ Failed to refresh {view_name}: {e}")
logger.info("Materialized view refresh completed")
async def setup(self):
"""Initialize database schema and load metadata."""
try:
# Connect to database
await self.db.connect()
# Initialize schema (this will drop and recreate public schema)
logger.info("Initializing database schema...")
schema_path = Path(__file__).parent / "db_schema.sql"
await self.db.initialize_schema()
logger.info("Database schema initialized successfully")
# Auto-detect materialized views
self.materialized_views = self.detect_materialized_views(schema_path)
# Load metadata from YAML
metadata_path = Path(__file__).parent.parent.parent / "metadata.yaml"
await self.load_metadata_from_yaml(metadata_path)
# Initial refresh of materialized views
if self.materialized_views:
logger.info("Performing initial refresh of materialized views...")
await self.refresh_materialized_views()
logger.info("Setup completed successfully")
except Exception as e:
logger.error(f"Setup failed: {e}", exc_info=True)
raise
async def periodic_refresh_loop(self):
"""
Periodically refresh materialized views.
Runs in a loop until shutdown is signaled.
"""
if not self.materialized_views:
logger.info("No materialized views to refresh, skipping periodic refresh")
return
logger.info(f"Starting periodic refresh loop (interval: {self.refresh_interval_minutes} minutes)")
while not self.shutdown_event.is_set():
try:
# Wait for the refresh interval or shutdown signal
await asyncio.wait_for(
self.shutdown_event.wait(),
timeout=self.refresh_interval_minutes * 60
)
# If we get here, shutdown was signaled
break
except asyncio.TimeoutError:
# Timeout means it's time to refresh
logger.info("Starting scheduled materialized view refresh...")
try:
await self.refresh_materialized_views()
except Exception as e:
logger.error(f"Error during scheduled refresh: {e}", exc_info=True)
# Continue the loop even if refresh fails
async def wait_for_tasks(self):
"""
Run periodic tasks and wait for shutdown signal.
Currently runs:
- Periodic materialized view refresh (configurable interval)
In the future, this could also:
- Poll Airbyte API for completed syncs
- Serve health check endpoints
- Listen to webhooks
"""
logger.info("View manager is ready")
logger.info("Press Ctrl+C to shutdown")
# Run periodic refresh loop
await self.periodic_refresh_loop()
logger.info("Periodic refresh loop stopped")
async def run(self):
"""Main entry point."""
try:
# Setup database and metadata
await self.setup()
# Wait for tasks or shutdown
await self.wait_for_tasks()
finally:
logger.info("Shutting down...")
await self.db.close()
logger.info("Shutdown complete")
def handle_shutdown(self, signum, frame):
"""Handle shutdown signals gracefully."""
logger.info(f"Received signal {signum}, initiating shutdown...")
self.shutdown_event.set()
def main():
"""CLI entry point."""
# Get refresh interval from environment variable
refresh_interval = int(os.getenv('REFRESH_INTERVAL_MINUTES', '60'))
logger.info(f"Configured refresh interval: {refresh_interval} minutes")
manager = ViewManagerSetup(refresh_interval_minutes=refresh_interval)
# Register signal handlers for graceful shutdown
signal.signal(signal.SIGINT, manager.handle_shutdown)
signal.signal(signal.SIGTERM, manager.handle_shutdown)
# Run the async main function
try:
asyncio.run(manager.run())
except KeyboardInterrupt:
logger.info("Interrupted by user")
sys.exit(0)
except Exception as e:
logger.error(f"Fatal error: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
main()

145
src/meta_api_grabber/test_ad_accounts.py
View File

@@ -1,145 +0,0 @@
"""
Simple test script to initialize database and grab ad_accounts metadata.
This is useful for testing the database setup and verifying ad account access.
Grabs ALL ad accounts accessible to the token.
"""
import asyncio
import os
from dotenv import load_dotenv
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.user import User
from facebook_business.api import FacebookAdsApi
from meta_api_grabber.database import TimescaleDBClient
async def test_ad_accounts():
"""Test database initialization and ad account metadata collection."""
load_dotenv()
# Get credentials from environment
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
print("❌ Missing required environment variables")
print(" Please ensure META_ACCESS_TOKEN, META_APP_SECRET, and META_APP_ID")
print(" are set in .env")
return 1
print("="*60)
print("AD ACCOUNTS TEST - GRABBING ALL ACCESSIBLE ACCOUNTS")
print("="*60)
print()
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
print("\nFetching all ad accounts accessible to this token...")
me = User(fbid='me')
account_fields = ['name', 'currency', 'timezone_name', 'account_status']
ad_accounts = me.get_ad_accounts(fields=account_fields)
print(f"Found {len(ad_accounts)} ad account(s)\n")
stored_count = 0
for account in ad_accounts:
account_id = account['id']
print(f"Ad Account {stored_count + 1}:")
print(f" ID: {account_id}")
print(f" Name: {account.get('name', 'N/A')}")
print(f" Currency: {account.get('currency', 'N/A')}")
print(f" Timezone: {account.get('timezone_name', 'N/A')}")
print(f" Status: {account.get('account_status', 'N/A')}")
try:
# Connect to database
print("Connecting to database...")
db = TimescaleDBClient()
await db.connect()
# Initialize schema
print("\nInitializing database schema...")
await db.initialize_schema()
# Get all ad accounts accessible to this token
print("\nFetching all ad accounts accessible to this token...")
me = User(fbid='me')
account_fields = ['name', 'currency', 'timezone_name', 'account_status']
ad_accounts = me.get_ad_accounts(fields=account_fields)
print(f"Found {len(ad_accounts)} ad account(s)\n")
stored_count = 0
for account in ad_accounts:
account_id = account['id']
print(f"Ad Account {stored_count + 1}:")
print(f" ID: {account_id}")
print(f" Name: {account.get('name', 'N/A')}")
print(f" Currency: {account.get('currency', 'N/A')}")
print(f" Timezone: {account.get('timezone_name', 'N/A')}")
print(f" Status: {account.get('account_status', 'N/A')}")
# Store in database
await db.upsert_ad_account(
account_id=account_id,
account_name=account.get('name'),
currency=account.get('currency'),
timezone_name=account.get('timezone_name'),
)
print(f" ✓ Stored in database\n")
stored_count += 1
# Verify by querying the database
print("="*60)
print(f"Verifying database storage ({stored_count} account(s))...")
print("="*60)
async with db.pool.acquire() as conn:
rows = await conn.fetch("SELECT * FROM ad_accounts ORDER BY account_name")
if rows:
print(f"\n✓ {len(rows)} ad account(s) found in database:\n")
for i, row in enumerate(rows, 1):
print(f"{i}. {row['account_name']} ({row['account_id']})")
print(f" Currency: {row['currency']} | Timezone: {row['timezone_name']}")
print(f" Updated: {row['updated_at']}\n")
else:
print("❌ No ad accounts found in database")
print("="*60)
print("TEST COMPLETED SUCCESSFULLY")
print("="*60)
print(f"\n✓ Successfully grabbed and stored {stored_count} ad account(s)")
print("\nNext steps:")
print("1. Check your database: docker exec meta_timescaledb psql -U meta_user -d meta_insights -c 'SELECT * FROM ad_accounts;'")
print("2. Run scheduled grabber for all accounts: meta-scheduled")
print("3. The scheduled grabber will now process all these accounts!")
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
return 1
finally:
await db.close()
return 0
def main():
"""Entry point for the test script."""
exit_code = asyncio.run(test_ad_accounts())
exit(exit_code)
if __name__ == "__main__":
main()

91
src/meta_api_grabber/test_campaign_insights.py
View File

@@ -1,91 +0,0 @@
"""
Test to see what campaign insights API actually returns for campaign_name.
"""
import os
import json
from dotenv import load_dotenv
from facebook_business.api import FacebookAdsApi
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.adsinsights import AdsInsights
# Load environment variables
load_dotenv()
# Initialize Facebook Ads API
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
raise ValueError("Missing required environment variables")
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
account_id = "act_238334370765317"
print(f"Testing campaign INSIGHTS fetch for account: {account_id}")
print("=" * 60)
# Get campaign insights (this is what scheduled_grabber.py does)
print("\nFetching campaign-level insights with 'today' preset...")
print("-" * 60)
fields = [
AdsInsights.Field.campaign_id,
AdsInsights.Field.campaign_name,
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.ctr,
AdsInsights.Field.cpc,
AdsInsights.Field.cpm,
AdsInsights.Field.reach,
AdsInsights.Field.actions,
AdsInsights.Field.date_start,
AdsInsights.Field.date_stop,
]
params = {
"date_preset": "today",
"level": "campaign",
"limit": 10,
}
ad_account = AdAccount(account_id)
insights = ad_account.get_insights(
fields=fields,
params=params,
)
insights_list = list(insights)
print(f"Found {len(insights_list)} campaign insights\n")
campaigns_without_names = []
for i, insight in enumerate(insights_list, 1):
insight_dict = dict(insight)
campaign_id = insight_dict.get('campaign_id')
campaign_name = insight_dict.get('campaign_name')
print(f"Campaign Insight {i}:")
print(f" Campaign ID: {campaign_id}")
print(f" Campaign Name: {campaign_name if campaign_name else '❌ MISSING'}")
print(f" Impressions: {insight_dict.get('impressions')}")
print(f" Spend: {insight_dict.get('spend')}")
print(f" Keys available: {list(insight_dict.keys())}")
print(f" Raw JSON: {json.dumps(insight_dict, indent=4, default=str)}")
print()
if not campaign_name:
campaigns_without_names.append(campaign_id)
print("=" * 60)
if campaigns_without_names:
print(f"⚠️ {len(campaigns_without_names)} campaigns WITHOUT names in insights:")
print(f" {campaigns_without_names}")
else:
print("✓ All campaigns have names in insights!")

118
src/meta_api_grabber/test_campaigns.py
View File

@@ -1,118 +0,0 @@
"""
Test script to diagnose campaign name issues for a specific ad account.
"""
import os
import json
from dotenv import load_dotenv
from facebook_business.api import FacebookAdsApi
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.campaign import Campaign
# Load environment variables
load_dotenv()
# Initialize Facebook Ads API
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
raise ValueError("Missing required environment variables")
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
# Test account ID
account_id = "act_238334370765317"
print(f"Testing campaign fetch for account: {account_id}")
print("=" * 60)
# Get campaigns using get_campaigns
print("\n1. Testing AdAccount.get_campaigns()...")
print("-" * 60)
ad_account = AdAccount(account_id)
campaigns = ad_account.get_campaigns(
fields=[
Campaign.Field.name,
Campaign.Field.status,
Campaign.Field.objective,
Campaign.Field.id,
],
params={'limit': 10}
)
campaigns_list = list(campaigns)
print(f"Found {len(campaigns_list)} campaigns\n")
campaigns_without_names = []
for i, campaign in enumerate(campaigns_list, 1):
campaign_dict = dict(campaign)
campaign_id = campaign_dict.get('id')
campaign_name = campaign_dict.get('name')
print(f"Campaign {i}:")
print(f" ID: {campaign_id}")
print(f" Name: {campaign_name if campaign_name else '❌ MISSING'}")
print(f" Status: {campaign_dict.get('status')}")
print(f" Objective: {campaign_dict.get('objective')}")
print(f" Raw data: {json.dumps(campaign_dict, indent=4)}")
print()
if not campaign_name:
campaigns_without_names.append(campaign_id)
# If we found campaigns without names, try fetching them individually
if campaigns_without_names:
print("\n2. Retrying campaigns without names (individual fetch)...")
print("-" * 60)
for campaign_id in campaigns_without_names:
print(f"\nFetching campaign {campaign_id} directly...")
try:
campaign_obj = Campaign(campaign_id)
campaign_data = campaign_obj.api_get(
fields=[
Campaign.Field.name,
Campaign.Field.status,
Campaign.Field.objective,
Campaign.Field.account_id,
]
)
campaign_dict = dict(campaign_data)
print(f" Direct fetch result:")
print(f" Name: {campaign_dict.get('name', '❌ STILL MISSING')}")
print(f" Status: {campaign_dict.get('status')}")
print(f" Raw data: {json.dumps(campaign_dict, indent=4)}")
except Exception as e:
print(f" Error fetching campaign: {e}")
print("\n" + "=" * 60)
print(f"Summary: {len(campaigns_without_names)} out of {len(campaigns_list)} campaigns have missing names")
print(f"Missing campaign IDs: {campaigns_without_names}")
else:
print("\n" + "=" * 60)
print("✓ All campaigns have names!")
print("\n3. Checking account permissions...")
print("-" * 60)
try:
account_data = ad_account.api_get(
fields=['name', 'account_status', 'capabilities', 'business']
)
account_dict = dict(account_data)
print(f"Account Name: {account_dict.get('name')}")
print(f"Account Status: {account_dict.get('account_status')}")
print(f"Capabilities: {json.dumps(account_dict.get('capabilities', []), indent=2)}")
print(f"Business ID: {account_dict.get('business')}")
except Exception as e:
print(f"Error fetching account details: {e}")
print("\n" + "=" * 60)
print("Test complete!")

125
src/meta_api_grabber/test_date_fields.py
View File

@@ -1,125 +0,0 @@
"""
Test that date_start and date_stop are properly stored in the database.
"""
import asyncio
import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.adsinsights import AdsInsights
from facebook_business.api import FacebookAdsApi
from meta_api_grabber.database import TimescaleDBClient
async def test_date_fields():
"""Test that date fields are stored correctly."""
load_dotenv()
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
account_id = "act_238334370765317"
if not all([access_token, app_secret, app_id]):
print("❌ Missing required environment variables")
return 1
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
print("="*70)
print("TESTING DATE_START AND DATE_STOP STORAGE")
print("="*70)
print()
# Connect to database
print("Connecting to database...")
db = TimescaleDBClient()
await db.connect()
try:
print("Fetching insights for 'today'...")
ad_account = AdAccount(account_id)
fields = [
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.date_start,
AdsInsights.Field.date_stop,
]
params = {
"date_preset": "today",
"level": "account",
}
insights = ad_account.get_insights(fields=fields, params=params)
# Store in database
timestamp = datetime.now(timezone.utc)
for insight in insights:
insight_dict = dict(insight)
print(f"\nAPI Response:")
print(f" date_start: {insight_dict.get('date_start')}")
print(f" date_stop: {insight_dict.get('date_stop')}")
print(f" impressions: {insight_dict.get('impressions')}")
await db.insert_account_insights(
time=timestamp,
account_id=account_id,
data=insight_dict,
date_preset="today",
)
print("\n✓ Stored in database")
# Verify from database
print("\nQuerying database...")
async with db.pool.acquire() as conn:
row = await conn.fetchrow("""
SELECT date_start, date_stop, date_preset, impressions, spend
FROM account_insights
WHERE account_id = $1
ORDER BY time DESC
LIMIT 1
""", account_id)
if row:
print("\n✓ Retrieved from database:")
print(f" date_start: {row['date_start']}")
print(f" date_stop: {row['date_stop']}")
print(f" date_preset: {row['date_preset']}")
print(f" impressions: {row['impressions']}")
print(f" spend: {row['spend']}")
print("\n" + "="*70)
print("✓ TEST PASSED - Date fields are stored correctly!")
print("="*70)
print("\nYou can now query historical data by date_stop:")
print(" - For clean daily trends, use: GROUP BY date_stop")
print(" - For latest value per day, use: ORDER BY time DESC with date_stop")
print()
else:
print("\n❌ No data found in database")
return 1
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
return 1
finally:
await db.close()
return 0
if __name__ == "__main__":
exit(asyncio.run(test_date_fields()))

132
src/meta_api_grabber/test_google_ads_accounts.py
View File

@@ -1,132 +0,0 @@
#!/usr/bin/env python3
"""
Test script to grab ad accounts from Google Ads API.
This script reads configuration from google-ads.yaml and authenticates using
a service account JSON key file to retrieve accessible customer accounts.
"""
from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException
import sys
def list_accessible_customers(client):
"""Lists all customer IDs accessible to the authenticated user.
Args:
client: An initialized GoogleAdsClient instance.
Returns:
List of customer resource names.
"""
customer_service = client.get_service("CustomerService")
try:
accessible_customers = customer_service.list_accessible_customers()
print(f"\nFound {len(accessible_customers.resource_names)} accessible customers:")
for resource_name in accessible_customers.resource_names:
customer_id = resource_name.split('/')[-1]
print(f" - Customer ID: {customer_id}")
print(f" Resource Name: {resource_name}")
return accessible_customers.resource_names
except GoogleAdsException as ex:
print(f"Request failed with status {ex.error.code().name}")
for error in ex.failure.errors:
print(f"\tError: {error.message}")
if error.location:
for field in error.location.field_path_elements:
print(f"\t\tField: {field.field_name}")
sys.exit(1)
def get_customer_details(client, customer_id):
"""Retrieves detailed information about a customer account.
Args:
client: An initialized GoogleAdsClient instance.
customer_id: The customer ID (without dashes).
"""
ga_service = client.get_service("GoogleAdsService")
query = """
SELECT
customer.id,
customer.descriptive_name,
customer.currency_code,
customer.time_zone,
customer.manager,
customer.test_account
FROM customer
WHERE customer.id = {customer_id}
""".format(customer_id=customer_id)
try:
response = ga_service.search(customer_id=customer_id, query=query)
for row in response:
customer = row.customer
print(f"\n--- Customer Details for {customer_id} ---")
print(f" ID: {customer.id}")
print(f" Name: {customer.descriptive_name}")
print(f" Currency: {customer.currency_code}")
print(f" Time Zone: {customer.time_zone}")
print(f" Is Manager: {customer.manager}")
print(f" Is Test Account: {customer.test_account}")
except GoogleAdsException as ex:
print(f"\nFailed to get details for customer {customer_id}")
print(f"Status: {ex.error.code().name}")
for error in ex.failure.errors:
print(f" Error: {error.message}")
def main():
"""Main function to test Google Ads API connection and list accounts."""
print("=" * 60)
print("Google Ads API - Account Listing Test")
print("=" * 60)
# Load client from YAML configuration
# By default, this looks for google-ads.yaml in the current directory
# or in the home directory
try:
print("\nLoading Google Ads client from configuration...")
client = GoogleAdsClient.load_from_storage(path="google-ads.yaml")
print("✓ Client loaded successfully")
except Exception as e:
print(f"✗ Failed to load client: {e}")
print("\nPlease ensure:")
print(" 1. google-ads.yaml exists and is properly configured")
print(" 2. google_ads_key.json exists and contains valid credentials")
print(" 3. All required fields are filled in google-ads.yaml")
sys.exit(1)
# List accessible customers
print("\n" + "=" * 60)
print("Listing Accessible Customers")
print("=" * 60)
resource_names = list_accessible_customers(client)
# Get detailed information for each customer
if resource_names:
print("\n" + "=" * 60)
print("Customer Details")
print("=" * 60)
for resource_name in resource_names:
customer_id = resource_name.split('/')[-1]
get_customer_details(client, customer_id)
print("\n" + "=" * 60)
print("Test completed successfully!")
print("=" * 60)
if __name__ == "__main__":
main()

171
src/meta_api_grabber/test_page_leads.py
View File

@@ -1,171 +0,0 @@
"""
Test script to retrieve leads from a specific Facebook page.
This uses the existing Meta API credentials to test leads retrieval.
"""
import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
from facebook_business.api import FacebookAdsApi
from facebook_business.adobjects.page import Page
async def test_page_leads():
"""Test retrieving leads from a specific Facebook page."""
load_dotenv()
# Get credentials from environment
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
print("❌ Missing required environment variables")
print(" Please ensure META_ACCESS_TOKEN, META_APP_SECRET, and META_APP_ID")
print(" are set in .env")
return 1
print("="*60)
print("PAGE LEADS TEST - RETRIEVING LEADS FROM A SPECIFIC PAGE")
print("="*60)
print()
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
# Prompt for page ID
page_id = input("Enter the Facebook Page ID: ").strip()
if not page_id:
print("❌ No page ID provided")
return 1
try:
# Initialize the Page object
print(f"\nConnecting to Page {page_id}...")
page = Page(fbid=page_id)
# First, get basic page information to verify access
print("\nFetching page information...")
page_fields = ['name', 'id', 'access_token']
page_info = page.api_get(fields=page_fields)
print(f"\n✓ Page Information:")
print(f" Name: {page_info.get('name', 'N/A')}")
print(f" ID: {page_info.get('id', 'N/A')}")
print(f" Has Access Token: {'Yes' if page_info.get('access_token') else 'No'}")
# Get leadgen forms associated with this page
print("\n" + "="*60)
print("Fetching Lead Generation Forms...")
print("="*60)
leadgen_forms = page.get_lead_gen_forms(
fields=['id', 'name', 'status', 'leads_count', 'created_time']
)
if not leadgen_forms or len(leadgen_forms) == 0:
print("\n⚠️ No lead generation forms found for this page")
print(" This could mean:")
print(" 1. The page has no lead forms")
print(" 2. The access token doesn't have permission to view lead forms")
print(" 3. The page ID is incorrect")
return 0
print(f"\nFound {len(leadgen_forms)} lead generation form(s):\n")
total_leads = 0
for idx, form in enumerate(leadgen_forms, 1):
form_id = form.get('id')
form_name = form.get('name', 'N/A')
form_status = form.get('status', 'N/A')
leads_count = form.get('leads_count', 0)
created_time = form.get('created_time', 'N/A')
print(f"Form {idx}:")
print(f" ID: {form_id}")
print(f" Name: {form_name}")
print(f" Status: {form_status}")
print(f" Leads Count: {leads_count}")
print(f" Created: {created_time}")
# Try to fetch actual leads from this form
try:
print(f"\n Fetching leads from form '{form_name}'...")
# Get the form object to retrieve leads
from facebook_business.adobjects.leadgenform import LeadgenForm
form_obj = LeadgenForm(fbid=form_id)
leads = form_obj.get_leads(
fields=['id', 'created_time', 'field_data']
)
leads_list = list(leads)
print(f" ✓ Retrieved {len(leads_list)} lead(s)")
if leads_list:
print(f"\n Sample leads from '{form_name}':")
for lead_idx, lead in enumerate(leads_list[:5], 1): # Show first 5 leads
lead_id = lead.get('id')
lead_created = lead.get('created_time', 'N/A')
field_data = lead.get('field_data', [])
print(f"\n Lead {lead_idx}:")
print(f" ID: {lead_id}")
print(f" Created: {lead_created}")
print(f" Fields:")
for field in field_data:
field_name = field.get('name', 'unknown')
field_values = field.get('values', [])
print(f" {field_name}: {', '.join(field_values)}")
if len(leads_list) > 5:
print(f"\n ... and {len(leads_list) - 5} more lead(s)")
total_leads += len(leads_list)
except Exception as lead_error:
print(f" ❌ Error fetching leads: {lead_error}")
print()
print("="*60)
print("TEST COMPLETED")
print("="*60)
print(f"\n✓ Total forms found: {len(leadgen_forms)}")
print(f"✓ Total leads retrieved: {total_leads}")
if total_leads == 0:
print("\n⚠️ No leads were retrieved. This could mean:")
print(" 1. The forms have no leads yet")
print(" 2. Your access token needs 'leads_retrieval' permission")
print(" 3. You need to request advanced access for leads_retrieval")
print("\nRequired permissions:")
print(" - pages_manage_ads")
print(" - pages_read_engagement")
print(" - leads_retrieval")
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
return 1
return 0
def main():
"""Entry point for the test script."""
exit_code = asyncio.run(test_page_leads())
exit(exit_code)
if __name__ == "__main__":
main()

108
src/meta_api_grabber/test_rate_limiter_issue.py
View File

@@ -1,108 +0,0 @@
"""
Test to diagnose if the rate limiter is causing campaign name issues.
"""
import os
import asyncio
from dotenv import load_dotenv
from facebook_business.api import FacebookAdsApi
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.campaign import Campaign
from rate_limiter import MetaRateLimiter
# Load environment variables
load_dotenv()
# Initialize Facebook Ads API
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
raise ValueError("Missing required environment variables")
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
account_id = "act_238334370765317"
# Initialize rate limiter
rate_limiter = MetaRateLimiter(
base_delay=0.1, # Fast for testing
throttle_threshold=75.0,
max_retry_delay=300.0,
max_retries=5,
)
async def _rate_limited_request(func, *args, **kwargs):
"""Execute a request with rate limiting (same as in scheduled_grabber.py)."""
return await rate_limiter.execute_with_retry(func, *args, **kwargs)
async def test_direct_fetch():
"""Test direct fetch without rate limiter."""
print("1. DIRECT FETCH (no rate limiter)")
print("=" * 60)
ad_account = AdAccount(account_id)
campaigns = ad_account.get_campaigns(
fields=[
Campaign.Field.name,
Campaign.Field.status,
Campaign.Field.objective,
],
params={'limit': 5}
)
print(f"Type of campaigns: {type(campaigns)}")
print(f"Campaigns object: {campaigns}\n")
for i, campaign in enumerate(campaigns, 1):
campaign_dict = dict(campaign)
print(f"Campaign {i}:")
print(f" ID: {campaign_dict.get('id')}")
print(f" Name: {campaign_dict.get('name', '❌ MISSING')}")
print(f" Keys in dict: {list(campaign_dict.keys())}")
print()
async def test_rate_limited_fetch():
"""Test fetch WITH rate limiter."""
print("\n2. RATE LIMITED FETCH (with rate limiter)")
print("=" * 60)
ad_account = AdAccount(account_id)
campaigns = await _rate_limited_request(
ad_account.get_campaigns,
fields=[
Campaign.Field.name,
Campaign.Field.status,
Campaign.Field.objective,
],
params={'limit': 5}
)
print(f"Type of campaigns: {type(campaigns)}")
print(f"Campaigns object: {campaigns}\n")
for i, campaign in enumerate(campaigns, 1):
campaign_dict = dict(campaign)
print(f"Campaign {i}:")
print(f" ID: {campaign_dict.get('id')}")
print(f" Name: {campaign_dict.get('name', '❌ MISSING')}")
print(f" Keys in dict: {list(campaign_dict.keys())}")
print()
async def main():
await test_direct_fetch()
await test_rate_limited_fetch()
if __name__ == "__main__":
asyncio.run(main())

87
src/meta_api_grabber/test_today_preset.py
View File

@@ -1,87 +0,0 @@
"""
Test script to check what date_start and date_stop look like for "today" preset.
"""
import os
import json
from dotenv import load_dotenv
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.adsinsights import AdsInsights
from facebook_business.api import FacebookAdsApi
def test_today_preset():
"""Test the 'today' date preset to see date_start and date_stop values."""
load_dotenv()
access_token = os.getenv("META_ACCESS_TOKEN")
app_secret = os.getenv("META_APP_SECRET")
app_id = os.getenv("META_APP_ID")
if not all([access_token, app_secret, app_id]):
print("❌ Missing required environment variables")
return 1
# Initialize Facebook Ads API
FacebookAdsApi.init(
app_id=app_id,
app_secret=app_secret,
access_token=access_token,
)
# Use the first account we know exists
account_id = "act_238334370765317"
ad_account = AdAccount(account_id)
print("="*70)
print("TESTING DATE_PRESET='TODAY'")
print("="*70)
print(f"Account: {account_id}")
print()
# Request with date_preset="today"
fields = [
AdsInsights.Field.impressions,
AdsInsights.Field.clicks,
AdsInsights.Field.spend,
AdsInsights.Field.date_start,
AdsInsights.Field.date_stop,
]
params = {
"date_preset": "today",
"level": "account",
}
print("Making API request with date_preset='today'...")
insights = ad_account.get_insights(fields=fields, params=params)
print("\nResponse:")
print("-"*70)
for insight in insights:
insight_dict = dict(insight)
print(json.dumps(insight_dict, indent=2))
print()
print("Key observations:")
print(f" date_start: {insight_dict.get('date_start')}")
print(f" date_stop: {insight_dict.get('date_stop')}")
print(f" Are they the same? {insight_dict.get('date_start') == insight_dict.get('date_stop')}")
print(f" impressions: {insight_dict.get('impressions')}")
print(f" spend: {insight_dict.get('spend')}")
print()
print("="*70)
print("CONCLUSION")
print("="*70)
print("For 'today' preset:")
print(" - date_start and date_stop should both be today's date")
print(" - Metrics are cumulative from midnight to now")
print(" - Multiple collections during the day will have same dates")
print(" but increasing metric values")
print()
return 0
if __name__ == "__main__":
exit(test_today_preset())

307
src/meta_api_grabber/token_manager.py
View File

@@ -1,307 +0,0 @@
"""
Token manager for automatic refresh of Meta access tokens.
Handles:
- Loading token metadata
- Checking token validity and expiry
- Automatic refresh before expiry
- Persistence of new tokens
"""
import json
import os
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
from dotenv import load_dotenv
from .auth import MetaOAuth2
class MetaTokenManager:
"""
Manages Meta access tokens with automatic refresh.
Features:
- Loads token from .env and metadata from .meta_token.json
- Checks if token is expired or about to expire
- Automatically refreshes tokens before expiry
- Persists refreshed tokens
"""
def __init__(
self,
token_file: str = ".meta_token.json",
refresh_before_days: int = 7,
):
"""
Initialize token manager.
Args:
token_file: Path to token metadata JSON file
refresh_before_days: Refresh token this many days before expiry
"""
load_dotenv()
self.token_file = Path(token_file)
self.refresh_before_days = refresh_before_days
self.oauth = MetaOAuth2()
self._current_token: Optional[str] = None
self._token_metadata: Optional[dict] = None
def load_token(self) -> Optional[str]:
"""
Load access token from environment.
Returns:
Access token or None
"""
return os.getenv("META_ACCESS_TOKEN")
def load_metadata(self) -> Optional[dict]:
"""
Load token metadata from JSON file.
Returns:
Token metadata dict or None
"""
if not self.token_file.exists():
return None
try:
return json.loads(self.token_file.read_text())
except Exception as e:
print(f"Warning: Could not load token metadata: {e}")
return None
def is_token_expired(self, metadata: dict) -> bool:
"""
Check if token is expired.
Args:
metadata: Token metadata dictionary
Returns:
True if expired
"""
expires_at = metadata.get("expires_at", 0)
if not expires_at:
return False
return time.time() >= expires_at
def should_refresh(self, metadata: dict) -> bool:
"""
Check if token should be refreshed.
Args:
metadata: Token metadata dictionary
Returns:
True if token should be refreshed
"""
expires_at = metadata.get("expires_at", 0)
if not expires_at:
return False
# Refresh if expiring within refresh_before_days
threshold = time.time() + (self.refresh_before_days * 86400)
return expires_at <= threshold
def refresh_token(self, current_token: str) -> str:
"""
Refresh token by exchanging for a new long-lived token.
Args:
current_token: Current access token
Returns:
New access token
"""
print("\n🔄 Refreshing access token...")
# Exchange current token for new long-lived token
token_data = self.oauth.exchange_for_long_lived_token(current_token)
new_token = token_data["access_token"]
expires_in = token_data["expires_in"]
print(f"✅ Token refreshed! Valid for {expires_in / 86400:.0f} days")
# Get token info
try:
token_info = self.oauth.get_token_info(new_token)
expires_at = token_info.get("expires_at", int(time.time()) + expires_in)
is_valid = token_info.get("is_valid", True)
except Exception:
expires_at = int(time.time()) + expires_in
is_valid = True
# Save new token
self._save_token(new_token, expires_at, is_valid)
return new_token
def _save_token(self, access_token: str, expires_at: int, is_valid: bool):
"""
Save token to .env and metadata to JSON.
Args:
access_token: New access token
expires_at: Expiry timestamp
is_valid: Whether token is valid
"""
# Update .env
env_path = Path(".env")
if env_path.exists():
env_content = env_path.read_text()
lines = env_content.split("\n")
updated = False
for i, line in enumerate(lines):
if line.startswith("META_ACCESS_TOKEN="):
lines[i] = f"META_ACCESS_TOKEN={access_token}"
updated = True
break
if not updated:
lines.append(f"META_ACCESS_TOKEN={access_token}")
env_path.write_text("\n".join(lines))
# Update metadata JSON
metadata = {
"access_token": access_token,
"expires_at": expires_at,
"issued_at": int(time.time()),
"is_valid": is_valid,
"updated_at": int(time.time()),
}
self.token_file.write_text(json.dumps(metadata, indent=2))
# Reload environment
load_dotenv(override=True)
def get_valid_token(self) -> str:
"""
Get a valid access token, refreshing if necessary.
Returns:
Valid access token
Raises:
ValueError: If no token available or refresh fails
"""
# Load token and metadata
token = self.load_token()
metadata = self.load_metadata()
if not token:
raise ValueError(
"No access token found. Run 'uv run python src/meta_api_grabber/auth.py' to authenticate."
)
# If no metadata, assume token is valid (but warn)
if not metadata:
print("⚠️ Warning: No token metadata found. Cannot check expiry.")
print(" Run 'uv run python src/meta_api_grabber/auth.py' to re-authenticate and save metadata.")
return token
# Check if expired
if self.is_token_expired(metadata):
print("❌ Token expired! Attempting to refresh...")
try:
return self.refresh_token(token)
except Exception as e:
raise ValueError(
f"Token expired and refresh failed: {e}\n"
"Please re-authenticate: uv run python src/meta_api_grabber/auth.py"
)
# Check if should refresh (within threshold)
if self.should_refresh(metadata):
expires_at = metadata.get("expires_at", 0)
expires_dt = datetime.fromtimestamp(expires_at)
days_left = (expires_dt - datetime.now()).days
print(f"\n⚠️ Token expiring in {days_left} days ({expires_dt.strftime('%Y-%m-%d')})")
print(f" Refreshing token to extend validity...")
try:
return self.refresh_token(token)
except Exception as e:
print(f"⚠️ Token refresh failed: {e}")
print(f" Continuing with current token ({days_left} days remaining)")
return token
# Token is valid
expires_at = metadata.get("expires_at", 0)
if expires_at:
expires_dt = datetime.fromtimestamp(expires_at)
days_left = (expires_dt - datetime.now()).days
print(f"✅ Token valid ({days_left} days remaining)")
return token
def print_token_status(self):
"""Print current token status."""
token = self.load_token()
metadata = self.load_metadata()
print("\n" + "="*60)
print("TOKEN STATUS")
print("="*60)
if not token:
print("❌ No access token found in .env")
print("\nRun: uv run python src/meta_api_grabber/auth.py")
return
print("✅ Access token found")
if not metadata:
print("⚠️ No metadata file (.meta_token.json)")
print(" Cannot determine expiry. Re-authenticate to save metadata.")
return
expires_at = metadata.get("expires_at", 0)
is_valid = metadata.get("is_valid", False)
if expires_at:
expires_dt = datetime.fromtimestamp(expires_at)
days_left = (expires_dt - datetime.now()).days
print(f"\nExpires: {expires_dt.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Days remaining: {days_left}")
if days_left < 0:
print("Status: ❌ EXPIRED")
elif days_left <= self.refresh_before_days:
print(f"Status: ⚠️ EXPIRING SOON (will auto-refresh)")
else:
print("Status: ✅ VALID")
print(f"Is valid: {'✅' if is_valid else '❌'}")
print("="*60 + "\n")
def main():
"""Check token status and refresh if needed."""
manager = MetaTokenManager()
manager.print_token_status()
try:
token = manager.get_valid_token()
print(f"\n✅ Valid token ready for use")
except ValueError as e:
print(f"\n❌ Error: {e}")
return 1
return 0
if __name__ == "__main__":
exit(main())

118
src/meta_api_grabber/view_manager.py
View File

@@ -1,118 +0,0 @@
"""
View manager for TimescaleDB materialized views.
Handles creation, updates, and refresh of materialized views for flattened insights data.
Views are loaded from individual SQL files in the views directory.
"""
import logging
import pathlib
from typing import List, Optional
import asyncpg
logger = logging.getLogger(__name__)
class ViewManager:
"""Manages materialized views for insights data flattening."""
def __init__(self, pool: asyncpg.Pool):
"""
Initialize view manager with a database connection pool.
Args:
pool: asyncpg connection pool
"""
self.pool = pool
self.views_dir = pathlib.Path(__file__).parent / "views"
async def initialize_views(self) -> None:
"""
Initialize all materialized views at startup.
Loads and executes SQL files from the views directory in alphabetical order.
Creates views if they don't exist, idempotent operation.
"""
logger.info("Initializing materialized views...")
if not self.views_dir.exists():
logger.warning(f"Views directory not found at {self.views_dir}")
return
# Get all .sql files in alphabetical order for consistent execution
view_files = sorted(self.views_dir.glob("*.sql"))
if not view_files:
logger.warning(f"No SQL files found in {self.views_dir}")
return
async with self.pool.acquire() as conn:
for view_file in view_files:
logger.debug(f"Loading view file: {view_file.name}")
await self._execute_view_file(conn, view_file)
logger.info("✓ Materialized views initialized successfully")
async def _execute_view_file(self, conn: asyncpg.Connection, view_file: pathlib.Path) -> None:
"""
Execute SQL statements from a view file.
Args:
conn: asyncpg connection
view_file: Path to SQL file
"""
with open(view_file, 'r') as f:
view_sql = f.read()
statements = [s.strip() for s in view_sql.split(';') if s.strip()]
for i, stmt in enumerate(statements, 1):
if not stmt:
continue
try:
await conn.execute(stmt)
logger.debug(f"{view_file.name}: Executed statement {i}")
except Exception as e:
error_msg = str(e).lower()
if "does not exist" in error_msg:
# Could be a missing dependent view or table, log it
logger.debug(f"{view_file.name}: View or table does not exist (statement {i})")
else:
# Log other errors but don't fail - could be incompatible schema changes
logger.warning(f"{view_file.name}: Error in statement {i}: {e}")
async def refresh_views(self, view_names: Optional[List[str]] = None) -> None:
"""
Refresh specified materialized views.
Args:
view_names: List of view names to refresh. If None, refreshes all views.
"""
if view_names is None:
view_names = [
"adset_insights_flattened",
"account_insights_flattened",
"campaign_insights_flattened",
"campaign_insights_by_country_flattened",
#"campaign_insights_by_device_flattened",
#"campaign_insights_by_gender_flattened",
]
async with self.pool.acquire() as conn:
for view_name in view_names:
try:
# Use CONCURRENTLY to avoid locking
await conn.execute(
f"REFRESH MATERIALIZED VIEW CONCURRENTLY {view_name};"
)
logger.debug(f"Refreshed materialized view: {view_name}")
except Exception as e:
error_msg = str(e).lower()
# View might not exist if not initialized, that's okay
if "does not exist" in error_msg:
logger.debug(f"View does not exist, skipping refresh: {view_name}")
else:
logger.warning(f"Error refreshing view {view_name}: {e}")
async def refresh_all_views(self) -> None:
"""Refresh all materialized views."""
await self.refresh_views()

20
src/meta_api_grabber/views/adset_insights.sql
View File

@@ -35,4 +35,22 @@ CREATE INDEX idx_adset_insights_flat_date ON adset_insights_flattened(date_start
CREATE UNIQUE INDEX idx_adset_insights_flat_unique ON adset_insights_flattened(time, adset_id); CREATE UNIQUE INDEX idx_adset_insights_flat_unique ON adset_insights_flattened(time, adset_id);
REFRESH MATERIALIZED VIEW CONCURRENTLY adset_insights_flattened; REFRESH MATERIALIZED VIEW CONCURRENTLY adset_insights_flattened;
SELECT
date_start as time,
account_id,
adset_id,
campaign_id,
SUM(impressions) AS impressions,
SUM(clicks) AS clicks,
SUM(spend) AS spend,
AVG(frequency) as frequency,
avg(cpc) as cpc,
avg(cpm) as cpm,
avg(cpp) as cpp,
avg(ctr) as ctr
FROM meta.ads_insights
group by time, account_id, adset_id, campaign_id

2
src/meta_api_grabber/views/campaign_insights.sql
View File

@@ -30,4 +30,4 @@ CREATE INDEX idx_campaign_insights_flat_date ON campaign_insights_flattened(date
CREATE UNIQUE INDEX idx_campaign_insights_flat_unique ON campaign_insights_flattened(time, campaign_id); CREATE UNIQUE INDEX idx_campaign_insights_flat_unique ON campaign_insights_flattened(time, campaign_id);
REFRESH MATERIALIZED VIEW CONCURRENTLY campaign_insights_flattened; REFRESH MATERIALIZED VIEW CONCURRENTLY campaign_insights_flattened;