Analytics

The analytics module provides utilities for working with various analytics platforms including Facebook Business API, Google Analytics, and Google Workspace write APIs (Sheets, Docs, Slides).

For Google Workspace (Sheets, Docs, Slides, Drive), see Google Workspace Write APIs.

Module Overview

Analytics Module — lazy-loaded.

Provides analytics integration: Google Analytics, Facebook Business, Snowflake, Data.world connectors, and Google Workspace write APIs (Sheets, Docs, Slides).

class siege_utilities.analytics.SnowflakeConnector

Bases: object

Snowflake data warehouse connector with advanced features.

__init__(account, user, password=None, warehouse=None, database=None, schema=None, role=None, config_file=None)

Initialize Snowflake connector.

Parameters:

• account (str) – Snowflake account identifier

• user (str) – Snowflake username

• password (str | None) – Snowflake password (optional if using config file)

• warehouse (str | None) – Default warehouse to use

• database (str | None) – Default database to use

• schema (str | None) – Default schema to use

• role (str | None) – Default role to use

• config_file (str | Path | None) – Path to configuration file

connect()

Establish connection to Snowflake.

Raises:

ConnectionError – If the connection attempt fails.

Return type:

None

disconnect()

Close Snowflake connection.

Return type:

None

execute_query(query, params=None)

Execute a SQL query.

Parameters:

• query (str) – SQL query string

• params (Dict[str, Any] | None) – Query parameters (optional)

Returns:

Query results as list of tuples

Raises:

• ConnectionError – If not connected and connection fails.

• RuntimeError – If query execution fails.

Return type:

List[tuple]

execute_ddl(ddl_statement)

Execute DDL statements (CREATE, ALTER, DROP, etc.).

Parameters:

ddl_statement (str) – DDL SQL statement

Raises:

• ConnectionError – If not connected and connection fails.

• RuntimeError – If DDL execution fails.

Return type:

None

upload_dataframe(df, table_name, database=None, schema=None, auto_create_table=True, overwrite=False)

Upload pandas DataFrame to Snowflake table.

Parameters:

• df (pd.DataFrame) – Pandas DataFrame to upload

• table_name (str) – Target table name

• database (str | None) – Target database (uses default if not specified)

• schema (str | None) – Target schema (uses default if not specified)

• auto_create_table (bool) – Whether to automatically create table if it doesn’t exist

• overwrite (bool) – Whether to overwrite existing table

Raises:

• ImportError – If pandas is not available.

• ConnectionError – If not connected and connection fails.

• RuntimeError – If the upload fails.

Return type:

None

download_dataframe(query, params=None)

Download data from Snowflake as pandas DataFrame.

Parameters:

• query (str) – SQL query to execute

• params (Dict[str, Any] | None) – Query parameters (optional)

Returns:

Pandas DataFrame with query results

Raises:

• ImportError – If pandas is not available.

• ConnectionError – If not connected and connection fails.

• RuntimeError – If the query fails.

Return type:

pd.DataFrame

get_table_info(table_name, database=None, schema=None)

Get information about a Snowflake table.

Parameters:

• table_name (str) – Name of the table

• database (str | None) – Database name (uses default if not specified)

• schema (str | None) – Schema name (uses default if not specified)

Returns:

Dictionary with table information

Raises:

• ConnectionError – If not connected and connection fails.

• RuntimeError – If the query fails.

Return type:

Dict[str, Any]

list_tables(database=None, schema=None)

List all tables in a database/schema.

Parameters:

• database (str | None) – Database name (uses default if not specified)

• schema (str | None) – Schema name (uses default if not specified)

Returns:

List of table names

Raises:

• ConnectionError – If not connected and connection fails.

• RuntimeError – If the query fails.

Return type:

List[str]

siege_utilities.analytics.get_snowflake_connector(config_file=None)

Get Snowflake connector instance from configuration.

Parameters:

config_file (str | Path | None)

Return type:

SnowflakeConnector

siege_utilities.analytics.upload_to_snowflake(df, table_name, config_file=None, **kwargs)

Convenience function to upload DataFrame to Snowflake.

Parameters:

• df (pd.DataFrame) – Pandas DataFrame to upload.

• table_name (str) – Target table name.

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to SnowflakeConnector.upload_dataframe(). See that method for accepted parameters.

Return type:

None

siege_utilities.analytics.download_from_snowflake(query, config_file=None, **kwargs)

Convenience function to download DataFrame from Snowflake.

Parameters:

• query (str) – SQL query to execute.

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to SnowflakeConnector.download_dataframe(). See that method for accepted parameters.

Return type:

pd.DataFrame

siege_utilities.analytics.execute_snowflake_query(query, config_file=None, **kwargs)

Convenience function to execute Snowflake query.

Parameters:

• query (str) – SQL query string.

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to SnowflakeConnector.execute_query(). See that method for accepted parameters.

Return type:

List[tuple]

class siege_utilities.analytics.DataDotWorldConnector

Bases: object

Data.world connector with advanced data discovery and access features.

__init__(api_token=None, config_file=None)

Initialize Data.world connector.

Parameters:

• api_token (str | None) – Data.world API token (optional if using config file)

• config_file (str | Path | None) – Path to configuration file

search_datasets(query, limit=50, owner=None, tags=None)

Search for datasets on data.world.

Parameters:

• query (str) – Search query string

• limit (int) – Maximum number of results to return

• owner (str | None) – Filter by dataset owner

• tags (List[str] | None) – Filter by tags

Returns:

List of dataset information dictionaries

Raises:

ConnectionError – If the API request fails.

Return type:

List[Dict[str, Any]]

get_dataset_info(dataset_id)

Get detailed information about a specific dataset.

Parameters:

dataset_id (str) – Dataset identifier (format: owner/dataset)

Returns:

Dictionary with detailed dataset information

Raises:

ConnectionError – If the API request fails.

Return type:

Dict[str, Any]

list_dataset_files(dataset_id)

List all files in a dataset.

Parameters:

dataset_id (str) – Dataset identifier (format: owner/dataset)

Returns:

List of file information dictionaries

Raises:

ConnectionError – If the API request fails.

Return type:

List[Dict[str, Any]]

download_file(dataset_id, file_name, output_path=None)

Download a specific file from a dataset.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset)

• file_name (str) – Name of the file to download

• output_path (str | Path | None) – Local path to save the file (optional)

Returns:

Path to the downloaded file

Raises:

• ConnectionError – If the download fails.

• OSError – If writing to disk fails.

Return type:

Path

load_dataset_as_dataframe(dataset_id, file_name=None)

Load a dataset file directly as a pandas DataFrame.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset)

• file_name (str | None) – Name of the file to load (optional, loads first file if not specified)

Returns:

Pandas DataFrame with the data

Raises:

• ImportError – If pandas is not available.

• ConnectionError – If the API request fails.

Return type:

pd.DataFrame

query_dataset(dataset_id, query, query_type='sql')

Execute a query against a dataset.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset)

• query (str) – Query string (SQL or SPARQL)

• query_type (str) – Type of query (‘sql’ or ‘sparql’)

Returns:

Pandas DataFrame with query results

Raises:

• ImportError – If pandas is not available.

• ValueError – If query_type is invalid.

• ConnectionError – If the query fails.

Return type:

pd.DataFrame

get_dataset_schema(dataset_id)

Get the schema information for a dataset.

Parameters:

dataset_id (str) – Dataset identifier (format: owner/dataset)

Returns:

Dictionary with schema information

Raises:

ConnectionError – If the API request fails.

Return type:

Dict[str, Any]

create_dataset(owner, dataset_id, title, description, tags=None, license='Public Domain', visibility='OPEN')

Create a new dataset on data.world.

Parameters:

• owner (str) – Dataset owner username

• dataset_id (str) – Unique dataset identifier

• title (str) – Dataset title

• description (str) – Dataset description

• tags (List[str] | None) – List of tags (optional)

• license (str) – Dataset license (optional)

• visibility (str) – Dataset visibility (‘OPEN’ or ‘PRIVATE’)

Returns:

owner/dataset_id)

Return type:

Created dataset ID (format

Raises:

• ValueError – If API token is not configured.

• ConnectionError – If the API request fails.

upload_file(dataset_id, file_path, file_name=None)

Upload a file to an existing dataset.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset)

• file_path (str | Path) – Local path to the file to upload

• file_name (str | None) – Name to use for the file on data.world (optional)

Raises:

• ValueError – If API token is not configured.

• FileNotFoundError – If the local file does not exist.

• ConnectionError – If the upload fails.

Return type:

None

update_dataset(dataset_id, title=None, description=None, tags=None, license=None)

Update an existing dataset.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset)

• title (str | None) – New title (optional)

• description (str | None) – New description (optional)

• tags (List[str] | None) – New tags (optional)

• license (str | None) – New license (optional)

Raises:

• ValueError – If API token is not configured or no updates provided.

• ConnectionError – If the API request fails.

Return type:

None

delete_dataset(dataset_id)

Delete a dataset.

Parameters:

dataset_id (str) – Dataset identifier (format: owner/dataset)

Raises:

• ValueError – If API token is not configured.

• ConnectionError – If the API request fails.

Return type:

None

siege_utilities.analytics.get_datadotworld_connector(config_file=None)

Get Data.world connector instance from configuration.

Parameters:

config_file (str | Path | None)

Return type:

DataDotWorldConnector

siege_utilities.analytics.search_datasets(query, config_file=None, **kwargs)

Standalone function to search for datasets.

Parameters:

• query (str) – Search query string

• config_file (str | Path | None) – Path to configuration file (optional)

• **kwargs – Additional search parameters

Returns:

List of matching datasets

Return type:

List[Dict[str, Any]]

siege_utilities.analytics.list_datasets(limit=50, config_file=None, **kwargs)

Standalone function to list available datasets.

Parameters:

• limit (int) – Maximum number of results to return

• config_file (str | Path | None) – Optional configuration file path

• **kwargs – Additional arguments to pass to search_datasets

Returns:

List of dataset information dictionaries

Return type:

List[Dict[str, Any]]

siege_utilities.analytics.search_datadotworld_datasets(query, config_file=None, **kwargs)

Convenience function to search for datasets.

Parameters:

• query (str) – Search query string.

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to DataDotWorldConnector.search_datasets(). See that method for accepted parameters.

Return type:

List[Dict[str, Any]]

siege_utilities.analytics.load_datadotworld_dataset(dataset_id, config_file=None, **kwargs)

Convenience function to load a dataset as DataFrame.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset).

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to DataDotWorldConnector.load_dataset_as_dataframe(). See that method for accepted parameters.

Return type:

pd.DataFrame

siege_utilities.analytics.query_datadotworld_dataset(dataset_id, query, config_file=None, **kwargs)

Convenience function to query a dataset.

Parameters:

• dataset_id (str) – Dataset identifier (format: owner/dataset).

• query (str) – Query string (SQL or SPARQL).

• config_file (str | Path | None) – Path to configuration file (optional).

• **kwargs – Forwarded to DataDotWorldConnector.query_dataset(). See that method for accepted parameters.

Return type:

pd.DataFrame

class siege_utilities.analytics.GoogleWorkspaceClient

Bases: object

Authenticated client that builds Google API service objects.

Do not instantiate directly — use the from_oauth() or from_service_account() class methods.

__init__(credentials)

classmethod from_oauth(client_id, client_secret, token_file=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', scopes=None)

Authenticate via OAuth2 interactive flow.

If token_file exists and contains a valid/refreshable token, no browser interaction is needed.

Parameters:

• client_id (str)

• client_secret (str)

• token_file (str | Path | None)

• redirect_uri (str)

• scopes (List[str] | None)

Return type:

GoogleWorkspaceClient

classmethod from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account='TLTQ3ANAABGCNEK7KIAOTDNK2Q', token_file=None, scopes=None)

Authenticate using credentials stored in a 1Password Document item.

Auto-detects whether the JSON document is an OAuth client secret (has "installed" or "web" key) or a service account key (has "type": "service_account"), and routes to the appropriate auth flow.

For OAuth: runs the installed-app flow (browser required on first use; cached token reused afterward).

For service account: authenticates server-to-server with no browser interaction.

Parameters:

• item_title (str) – Title of the 1Password Document item.

• vault (str | None) – 1Password vault name.

• account (str | None) – 1Password account shorthand or UUID.

• token_file (str | Path | None) – Path to cache the OAuth token (ignored for service accounts). Default: ~/.siege/tokens/workspace_token.json.

• scopes (List[str] | None) – OAuth scopes (defaults to WORKSPACE_SCOPES).

Returns:

Authenticated GoogleWorkspaceClient.

Return type:

GoogleWorkspaceClient

classmethod from_service_account(service_account_data=None, service_account_file=None, scopes=None)

Authenticate via service account credentials.

Provide service_account_data (dict) or service_account_file (path). If neither is given, attempts to fetch from 1Password via CredentialManager.get_google_service_account_from_1password().

Parameters:

• service_account_data (Dict[str, Any] | None)

• service_account_file (str | Path | None)

• scopes (List[str] | None)

Return type:

GoogleWorkspaceClient

classmethod from_credentials(credentials)

Wrap an already-authenticated google.auth.credentials.Credentials.

Return type:

GoogleWorkspaceClient

classmethod from_account(account, person=None, scopes=None)

Build a client from a GoogleAccount.

For OAuth accounts the method looks for a cached token_file first, then resolves oauth_integration_name from person to get client_id/client_secret for the OAuth flow.

For service accounts the method resolves service_account_ref via CredentialManager (1Password) or treats it as a file path.

Parameters:

• account (GoogleAccount)

• person (Optional[Person])

• scopes (Optional[List[str]])

Return type:

GoogleWorkspaceClient

classmethod from_registry(registry, google_account_id=None, person=None, scopes=None)

Build a client from a registry’s default or specified account.

If google_account_id is given, looks it up directly; otherwise uses the registry’s default account.

Parameters:

• registry (GoogleAccountRegistry)

• google_account_id (Optional[str])

• person (Optional[Person])

• scopes (Optional[List[str]])

Return type:

GoogleWorkspaceClient

sheets_service()

Return the Google Sheets API v4 service object.

docs_service()

Return the Google Docs API v1 service object.

slides_service()

Return the Google Slides API v1 service object.

drive_service()

Return the Google Drive API v3 service object.

property credentials

batch_update_spreadsheet(spreadsheet_id, requests)

Execute a batch of Sheets API requests.

Parameters:

• spreadsheet_id (str) – Target spreadsheet ID.

• requests (List[Dict[str, Any]]) – List of request dicts per the Sheets API batchUpdate spec.

Returns:

The API response dict.

Return type:

Dict[str, Any]

batch_update_document(document_id, requests)

Execute a batch of Docs API requests.

Parameters:

• document_id (str) – Target document ID.

• requests (List[Dict[str, Any]]) – List of request dicts per the Docs API batchUpdate spec.

Returns:

The API response dict.

Return type:

Dict[str, Any]

batch_update_presentation(presentation_id, requests)

Execute a batch of Slides API requests.

Parameters:

• presentation_id (str) – Target presentation ID.

• requests (List[Dict[str, Any]]) – List of request dicts per the Slides API batchUpdate spec.

Returns:

The API response dict.

Return type:

Dict[str, Any]

static spreadsheet_url(spreadsheet_id)

Return the live Google Sheets URL for a spreadsheet ID.

Parameters:

spreadsheet_id (str)

Return type:

str

static document_url(document_id)

Return the live Google Docs URL for a document ID.

Parameters:

document_id (str)

Return type:

str

static presentation_url(presentation_id)

Return the live Google Slides URL for a presentation ID.

Parameters:

presentation_id (str)

Return type:

str

static file_url(file_id, mime_type=None)

Return the live Google Drive URL for a file ID.

If mime_type is provided, returns the appropriate editor URL. Otherwise returns the generic Drive file URL.

Parameters:

• file_id (str)

• mime_type (str | None)

Return type:

str

copy_file(file_id, title=None)

Copy a Drive file (spreadsheet, doc, presentation) and return the new ID.

Parameters:

• file_id (str) – The ID of the file to copy.

• title (str | None) – Optional title for the copy. None keeps the default “Copy of …” naming.

Returns:

The new file’s ID.

Return type:

str

share_file(file_id, email, role='writer', send_notification=False)

Share a Drive file with a user.

Parameters:

• file_id (str) – The file to share.

• email (str) – Email address of the recipient.

• role (str) – "reader", "writer", or "commenter".

• send_notification (bool) – Whether to send an email notification.

Returns:

The permission resource dict.

Return type:

Dict[str, Any]

move_to_folder(file_id, folder_id)

Move a file into a Drive folder.

Parameters:

• file_id (str) – The file to move.

• folder_id (str) – Target folder ID.

Returns:

The updated file resource dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.create_spreadsheet(client, title, sheet_names=None, folder_id=None)

Create a new Google Spreadsheet and return its ID.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• title (str) – Title for the new spreadsheet.

• sheet_names (List[str] | None) – Optional list of sheet/tab names to create. If omitted, a single default “Sheet1” is created.

• folder_id (str | None) – Optional Drive folder ID to create the spreadsheet in.

Returns:

The spreadsheet ID string.

Return type:

str

siege_utilities.analytics.write_values(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')

Write a 2-D list of values to a range.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• spreadsheet_id (str) – Target spreadsheet ID.

• range – A1 notation range (e.g. "Sheet1!A1:D10").

• values (List[List[Any]]) – Row-major list of lists.

• value_input_option (str) – "RAW" or "USER_ENTERED" (default).

• range_ (str)

Returns:

The API response dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.append_rows(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')

Append rows after existing data in a range.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• spreadsheet_id (str) – Target spreadsheet ID.

• range – A1 notation range to search for a table (e.g. "Sheet1").

• values (List[List[Any]]) – Row-major list of lists to append.

• value_input_option (str) – "RAW" or "USER_ENTERED" (default).

• range_ (str)

Returns:

The API response dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.read_values(client, spreadsheet_id, range_)

Read values from a range.

Returns:

Row-major list of lists (may be ragged).

Parameters:

• spreadsheet_id (str)

• range_ (str)

Return type:

List[List[Any]]

siege_utilities.analytics.write_dataframe(client, spreadsheet_id, df, sheet_name='Sheet1', include_header=True, start_cell='A1')

Write a pandas DataFrame to a sheet.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• spreadsheet_id (str) – Target spreadsheet ID.

• df – pandas DataFrame to write.

• sheet_name (str) – Tab name (default "Sheet1").

• include_header (bool) – Whether to include column names as first row.

• start_cell (str) – Top-left cell (default "A1").

Returns:

The API response dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.read_dataframe(client, spreadsheet_id, range_='Sheet1', has_header=True)

Read a range from a spreadsheet into a pandas DataFrame.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• spreadsheet_id (str) – Source spreadsheet ID.

• range – A1 notation range (e.g. "Sheet1" or "Sheet1!A1:D100").

• has_header (bool) – If True, first row is used as column names.

• range_ (str)

Returns:

pandas DataFrame.

siege_utilities.analytics.add_sheet(client, spreadsheet_id, title)

Add a new tab/sheet to an existing spreadsheet.

Returns:

The new sheet ID (integer).

Parameters:

• spreadsheet_id (str)

• title (str)

Return type:

int

siege_utilities.analytics.get_spreadsheet_metadata(client, spreadsheet_id)

Fetch spreadsheet metadata (title, sheets, locale, etc.).

Returns:

The spreadsheet resource dict.

Parameters:

spreadsheet_id (str)

Return type:

Dict[str, Any]

siege_utilities.analytics.copy_spreadsheet(client, spreadsheet_id, title=None)

Copy an entire spreadsheet via the Drive API.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• spreadsheet_id (str) – Source spreadsheet ID.

• title (str | None) – Title for the copy (default: “Copy of …”).

Returns:

The new spreadsheet ID.

Return type:

str

siege_utilities.analytics.create_presentation(client, title, folder_id=None)

Create a new Google Slides presentation and return its ID.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• title (str) – Title for the new presentation.

• folder_id (str | None) – Optional Drive folder ID to create the presentation in.

Returns:

The presentation ID string.

Return type:

str

siege_utilities.analytics.get_presentation(client, presentation_id)

Fetch full presentation metadata.

Returns:

The presentation resource dict.

Parameters:

presentation_id (str)

Return type:

Dict[str, Any]

siege_utilities.analytics.copy_presentation(client, presentation_id, title=None)

Copy an entire presentation via the Drive API.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• presentation_id (str) – Source presentation ID.

• title (str | None) – Title for the copy (default: “Copy of …”).

Returns:

The new presentation ID.

Return type:

str

siege_utilities.analytics.add_blank_slide(client, presentation_id, layout='BLANK', insertion_index=None)

Add a blank slide to a presentation.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• presentation_id (str) – Target presentation ID.

• layout (str) – Predefined layout ("BLANK", "TITLE", "TITLE_AND_BODY", etc.).

• insertion_index (int | None) – Position (0-based). None appends at end.

Returns:

The new slide’s object ID.

Return type:

str

siege_utilities.analytics.create_textbox(client, presentation_id, slide_id, text, left=100, top=100, width=400, height=50)

Create a text box on a slide and populate it with text.

Dimensions are in EMU (English Metric Units). 1 inch = 914400 EMU. For convenience, this function accepts points (1 pt = 12700 EMU) but the values are treated as points internally.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• presentation_id (str) – Target presentation ID.

• slide_id (str) – Slide object ID.

• text (str) – Text to put in the box.

• left (float) – Position in points from top-left.

• top (float) – Position in points from top-left.

• width (float) – Dimensions in points.

• height (float) – Dimensions in points.

Returns:

The textbox object ID.

Return type:

str

siege_utilities.analytics.create_document(client, title, folder_id=None)

Create a new Google Doc and return its document ID.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• title (str) – Title for the new document.

• folder_id (str | None) – Optional Drive folder ID to create the document in.

Returns:

The document ID string.

Return type:

str

siege_utilities.analytics.get_document(client, document_id)

Fetch full document metadata and content structure.

Returns:

The document resource dict.

Parameters:

document_id (str)

Return type:

Dict[str, Any]

siege_utilities.analytics.copy_document(client, document_id, title=None)

Copy a document via the Drive API.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• document_id (str) – Source document ID.

• title (str | None) – Title for the copy (default: “Copy of …”).

Returns:

The new document ID.

Return type:

str

siege_utilities.analytics.read_document_text(client, document_id)

Read the full plain-text content of a document.

Extracts text from all structural elements (paragraphs, tables, etc.).

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• document_id (str) – Target document ID.

Returns:

The document text as a single string.

Return type:

str

siege_utilities.analytics.insert_paragraph(client, document_id, text, index=1, heading=None)

Insert a paragraph with optional heading style.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• document_id (str) – Target document ID.

• text (str) – Paragraph text (newline is appended automatically).

• index (int) – Character index (1-based).

• heading (str | None) – Named style ("HEADING_1" through "HEADING_6", "TITLE", "SUBTITLE", or None for normal text).

Returns:

The API response dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.insert_table(client, document_id, rows, cols, index=1)

Insert an empty table at the given index.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• document_id (str) – Target document ID.

• rows (int) – Number of rows.

• cols (int) – Number of columns.

• index (int) – Character index where the table is inserted.

Returns:

The API response dict.

Return type:

Dict[str, Any]

siege_utilities.analytics.replace_text(client, document_id, find, replace_with, match_case=True)

Find and replace text throughout a document.

Parameters:

• client – Authenticated GoogleWorkspaceClient.

• document_id (str) – Target document ID.

• find (str) – Text to search for.

• replace_with (str) – Replacement text.

• match_case (bool) – Whether the search is case-sensitive.

Returns:

The API response dict.

Return type:

Dict[str, Any]

class siege_utilities.analytics.VistaSocialConnector

Bases: object

Bearer-token authenticated client for the Vista Social REST API.

Usage:

from siege_utilities.analytics.vista_social import VistaSocialConnector

client = VistaSocialConnector(api_token=os.environ["VISTA_SOCIAL_TOKEN"])
accounts = client.list_accounts()
analytics = client.get_account_analytics(
    account_id=accounts[0]["id"],
    start_date="2026-01-01",
    end_date="2026-01-31",
)

All methods raise:

• VistaSocialAuthError on 401/403,

• VistaSocialRateLimitError on 429,

• VistaSocialError on other non-2xx responses or network failures.

The connector follows the project’s failure-mode discipline (see docs/FAILURE_MODES.md): no method silently returns an empty dict on failure. Every failure mode is either a typed exception or a None return with a documented meaning.

__init__(api_token, *, base_url='https://api.vistasocial.com', timeout=30, retry_attempts=3, retry_backoff=1.5)

Parameters:

• api_token (str)

• base_url (str)

• timeout (float)

• retry_attempts (int)

• retry_backoff (float)

Return type:

None

close()

Close the underlying HTTP session.

Return type:

None

list_accounts()

Return the list of accounts this token can access.

Wraps GET /v1/accounts. Returns the decoded data list from the response, or an empty list if the API returned nothing. Raises on auth / rate-limit / network failures (no silent empty-on-error).

Return type:

List[Mapping[str, Any]]

list_profiles(account_id)

Return social profiles connected to account_id.

Wraps GET /v1/accounts/{account_id}/profiles.

Parameters:

account_id (str)

Return type:

List[Mapping[str, Any]]

get_account_analytics(account_id, *, start_date, end_date, metrics=None)

Fetch aggregate analytics for account_id over a date range.

Parameters:

• account_id (str) – Vista Social account id (string, opaque).

• start_date (str) – ISO-8601 date (YYYY-MM-DD), inclusive.

• end_date (str) – ISO-8601 date, inclusive.

• metrics (List[str] | None) – Optional list of metric names (e.g. ["impressions", "engagements", "reach", "follower_count"]). If omitted, Vista Social returns its default metric set.

Return type:

Dict[str, Any]

Wraps GET /v1/accounts/{account_id}/analytics. Returns the decoded JSON body verbatim – Vista Social returns a structured document with metric breakdowns; consumers can extract what they need.

list_posts(account_id, *, start_date=None, end_date=None, limit=100)

Return posts for account_id with optional date filtering.

Wraps GET /v1/accounts/{account_id}/posts. Pagination is intentionally simple here – consumers that need full enumeration should call repeatedly with adjusted date ranges or extend this connector with cursor handling once the real API spec is known.

Parameters:

• account_id (str)

• start_date (str | None)

• end_date (str | None)

• limit (int)

Return type:

List[Mapping[str, Any]]

class siege_utilities.analytics.VistaSocialResponse

Bases: object

Structured response from a Vista Social API call.

The connector returns this rather than the raw requests.Response so consumers don’t need to know about the underlying HTTP library.

status_code

HTTP status code (200 / 201 etc.).

Type:

int

data

Parsed JSON body. Empty dict if the response had no body (e.g. 204 No Content) or wasn’t JSON-decodable.

Type:

Mapping[str, Any]

headers

Response headers – useful for inspecting rate-limit counters (X-RateLimit-Remaining, Retry-After).

Type:

Mapping[str, str]

url

The fully-qualified URL that was called, for logging.

Type:

str

status_code: int

data: Mapping[str, Any]

headers: Mapping[str, str]

url: str = ''

__init__(status_code, data, headers=<factory>, url='')

Parameters:

• status_code (int)

• data (Mapping[str, Any])

• headers (Mapping[str, str])

• url (str)

Return type:

None

exception siege_utilities.analytics.VistaSocialError

Bases: RuntimeError

Base class for all Vista Social connector failures.

exception siege_utilities.analytics.VistaSocialAuthError

Bases: VistaSocialError

Authentication failed (401/403). Token is missing, wrong, or expired.

exception siege_utilities.analytics.VistaSocialRateLimitError

Bases: VistaSocialError

API rate limit hit (429). Caller can back off and retry later.

class siege_utilities.analytics.Dimension

Bases: object

A scoring dimension with a name and weight (0-100).

name: str

weight: float

__init__(name, weight)

Parameters:

• name (str)

• weight (float)

Return type:

None

class siege_utilities.analytics.ThresholdConfig

Bases: object

Tier thresholds for score classification.

Scores >= green_min are Green, >= yellow_min are Yellow, else Red.

green_min: float = 70.0

yellow_min: float = 40.0

classify(score)

Parameters:

score (float)

Return type:

str

__init__(green_min=70.0, yellow_min=40.0)

Parameters:

• green_min (float)

• yellow_min (float)

Return type:

None

class siege_utilities.analytics.EntityScore

Bases: object

Result of scoring an entity.

entity_id: str

composite_score: float

tier: str

dimension_scores: dict[str, float]

segment: str | None = None

trend: float | None = None

trend_direction: str | None = None

__init__(entity_id, composite_score, tier, dimension_scores, segment=None, trend=None, trend_direction=None)

Parameters:

• entity_id (str)

• composite_score (float)

• tier (str)

• dimension_scores (dict[str, float])

• segment (str | None)

• trend (float | None)

• trend_direction (str | None)

Return type:

None

class siege_utilities.analytics.HealthScorer

Bases: object

Weighted multi-dimensional entity scorer.

Parameters:

• dimensions (list[Dimension]) – Scoring dimensions. Weights must sum to 100.

• default_thresholds (ThresholdConfig, optional) – Default tier thresholds. Can be overridden per-segment.

• segment_thresholds (dict[str, ThresholdConfig], optional) – Per-segment threshold overrides.

Raises:

ValueError – If dimensions is empty or weights don’t sum to 100.

__init__(dimensions, default_thresholds=None, segment_thresholds=None)

Parameters:

• dimensions (list[Dimension])

• default_thresholds (ThresholdConfig | None)

• segment_thresholds (dict[str, ThresholdConfig] | None)

Return type:

None

property dimension_names: list[str]

score(entity_id, scores, *, segment=None, previous_score=None)

Score an entity across all dimensions.

Parameters:

• entity_id (str) – Identifier for the entity being scored.

• scores (dict[str, float]) – Per-dimension scores (0-100). Must include all dimensions.

• segment (str, optional) – Segment for threshold selection.

• previous_score (float, optional) – Previous composite score for trend detection.

Return type:

EntityScore

Raises:

ValueError – If scores dict is missing dimensions or values are out of range.

score_batch(entities)

Score multiple entities.

Each dict must have ‘entity_id’ and ‘scores’ keys. Optional: ‘segment’, ‘previous_score’.

Parameters:

entities (list[dict]) – List of entity dicts with keys: entity_id, scores, and optionally segment, previous_score.

Return type:

list[EntityScore]

Raises:

ValueError – If any entity dict is missing required keys.

class siege_utilities.analytics.SignalType

Bases: Enum

Types of risk signals.

DECLINE = 'decline'

THRESHOLD = 'threshold'

ABSENCE = 'absence'

ANOMALY = 'anomaly'

class siege_utilities.analytics.RiskTier

Bases: Enum

Risk classification tiers, ordered by severity.

CRITICAL = 'Critical'

HIGH = 'High'

MEDIUM = 'Medium'

LOW = 'Low'

class siege_utilities.analytics.RiskSignal

Bases: object

A named risk signal with type and weight.

Parameters:

• name (str) – Signal identifier.

• signal_type (SignalType) – Category of risk this signal measures.

• weight (float) – Relative weight (0, 100]. All signals’ weights must sum to 100.

name: str

signal_type: SignalType

weight: float

__init__(name, signal_type, weight)

Parameters:

• name (str)

• signal_type (SignalType)

• weight (float)

Return type:

None

class siege_utilities.analytics.RiskTierConfig

Bases: object

Thresholds for risk tier classification.

Scores >= critical_min are Critical, >= high_min are High, etc.

critical_min: float = 80.0

high_min: float = 60.0

medium_min: float = 40.0

classify(score)

Parameters:

score (float)

Return type:

RiskTier

__init__(critical_min=80.0, high_min=60.0, medium_min=40.0)

Parameters:

• critical_min (float)

• high_min (float)

• medium_min (float)

Return type:

None

class siege_utilities.analytics.RiskAssessment

Bases: object

Result of a risk assessment.

entity_id: str

composite_score: float

tier: RiskTier

signal_scores: dict[str, float]

signal_types: dict[str, SignalType]

intervention: str | None = None

__init__(entity_id, composite_score, tier, signal_scores, signal_types, intervention=None)

Parameters:

• entity_id (str)

• composite_score (float)

• tier (RiskTier)

• signal_scores (dict[str, float])

• signal_types (dict[str, SignalType])

• intervention (str | None)

Return type:

None

class siege_utilities.analytics.RiskAnalyzer

Bases: object

Multi-signal risk stacking engine.

Parameters:

• signals (list[RiskSignal]) – Risk signals to evaluate. Weights must sum to 100.

• tier_config (RiskTierConfig, optional) – Tier classification thresholds.

• interventions (dict[RiskTier, str], optional) – Mapping from risk tier to recommended intervention.

Raises:

ValueError – If signals is empty or weights don’t sum to 100.

__init__(signals, tier_config=None, interventions=None)

Parameters:

• signals (list[RiskSignal])

• tier_config (RiskTierConfig | None)

• interventions (dict[RiskTier, str] | None)

Return type:

None

property signal_names: list[str]

assess(entity_id, scores)

Assess risk for an entity.

Parameters:

• entity_id (str) – Identifier for the entity.

• scores (dict[str, float]) – Per-signal risk scores (0-100). Higher = more risk.

Return type:

RiskAssessment

Raises:

ValueError – If scores dict is missing signals or values are out of range.

assess_batch(entities)

Assess risk for multiple entities.

Each dict must have ‘entity_id’ and ‘scores’ keys.

Return type:

list[RiskAssessment]

Parameters:

entities (list[dict[str, Any]])

class siege_utilities.analytics.DimensionScore

Bases: object

A score for one entity on one dimension.

Parameters:

• score (int) – Score on a 1-5 scale.

• evidence (str) – Citation supporting the score (URL, doc reference, test result).

score: int

evidence: str

__init__(score, evidence)

Parameters:

• score (int)

• evidence (str)

Return type:

None

class siege_utilities.analytics.ComparisonResult

Bases: object

Result of comparing entities across dimensions.

dimensions: list[str]

entities: list[str]

scores: dict[str, dict[str, DimensionScore]]

weighted_totals: dict[str, float]

gap_analysis: list[tuple[str, float]]

__init__(dimensions, entities, scores, weighted_totals, gap_analysis)

Parameters:

• dimensions (list[str])

• entities (list[str])

• scores (dict[str, dict[str, DimensionScore]])

• weighted_totals (dict[str, float])

• gap_analysis (list[tuple[str, float]])

Return type:

None

class siege_utilities.analytics.FeatureMatrix

Bases: object

Boolean feature support matrix with coverage scoring.

entities: list[str]

features: list[str]

matrix: dict[str, dict[str, bool]]

coverage: dict[str, float]

__init__(entities, features, matrix, coverage)

Parameters:

• entities (list[str])

• features (list[str])

• matrix (dict[str, dict[str, bool]])

• coverage (dict[str, float])

Return type:

None

class siege_utilities.analytics.ComparativeAnalyzer

Bases: object

N-entity x M-dimension comparison engine.

Parameters:

• dimensions (list[str]) – Dimension names for comparison.

• weights (dict[str, float] | None) – Optional dimension weights. If provided, must cover all dimensions. If omitted, equal weights are used.

Raises:

ValueError – If dimensions is empty or weights don’t match dimensions.

__init__(dimensions, weights=None)

Parameters:

• dimensions (list[str])

• weights (dict[str, float] | None)

Return type:

None

compare(entity_scores)

Compare entities across all dimensions.

Parameters:

entity_scores (dict[str, dict[str, DimensionScore]]) – Outer key = entity name, inner key = dimension name.

Return type:

ComparisonResult

Raises:

ValueError – If any entity is missing a dimension score.

static build_feature_matrix(entity_features)

Build a boolean feature support matrix with coverage scores.

Parameters:

entity_features (dict[str, dict[str, bool]]) – Outer key = entity name, inner key = feature name.

Return type:

FeatureMatrix

Raises:

ValueError – If no entities or no features are provided.

class siege_utilities.analytics.ForecastAccuracy

Bases: object

Result of a forecast accuracy calculation.

mape: float

bias: float

bias_direction: str

grade: str

n_observations: int

n_excluded: int

__init__(mape, bias, bias_direction, grade, n_observations, n_excluded)

Parameters:

• mape (float)

• bias (float)

• bias_direction (str)

• grade (str)

• n_observations (int)

• n_excluded (int)

Return type:

None

class siege_utilities.analytics.CategoryBreakdown

Bases: object

Per-category forecast accuracy.

category: str

accuracy: ForecastAccuracy

__init__(category, accuracy)

Parameters:

• category (str)

• accuracy (ForecastAccuracy)

Return type:

None

class siege_utilities.analytics.TrendPoint

Bases: object

Accuracy at a single time period.

period: str

mape: float

bias: float

__init__(period, mape, bias)

Parameters:

• period (str)

• mape (float)

• bias (float)

Return type:

None

class siege_utilities.analytics.ForecastReport

Bases: object

Complete forecast accuracy report.

overall: ForecastAccuracy

by_category: list[CategoryBreakdown]

trend: list[TrendPoint]

trend_direction: str | None

__init__(overall, by_category, trend, trend_direction)

Parameters:

• overall (ForecastAccuracy)

• by_category (list[CategoryBreakdown])

• trend (list[TrendPoint])

• trend_direction (str | None)

Return type:

None

class siege_utilities.analytics.ForecastAnalyzer

Bases: object

Forecast accuracy engine.

Parameters:

• actuals (list[float]) – Actual observed values.

• forecasts (list[float]) – Forecasted values (same length as actuals).

• categories (list[str] | None) – Optional category labels for per-category breakdown.

• periods (list[str] | None) – Optional time period labels for trend analysis.

Raises:

ValueError – If inputs have mismatched lengths or are empty.

__init__(actuals, forecasts, categories=None, periods=None)

Parameters:

• actuals (list[float])

• forecasts (list[float])

• categories (list[str] | None)

• periods (list[str] | None)

Return type:

None

analyze()

Run full forecast accuracy analysis.

Return type:

ForecastReport

class siege_utilities.analytics.PipelineItem

Bases: object

A single item in the pipeline.

Parameters:

• item_id (str) – Unique identifier.

• stage (str) – Current pipeline stage.

• value (float) – Item value (dollars, record count, etc.).

• entered_stage (datetime | None) – When the item entered this stage. Required for velocity metrics.

item_id: str

stage: str

value: float

entered_stage: datetime | None = None

__init__(item_id, stage, value, entered_stage=None)

Parameters:

• item_id (str)

• stage (str)

• value (float)

• entered_stage (datetime | None)

Return type:

None

class siege_utilities.analytics.StageMetrics

Bases: object

Metrics for a single pipeline stage.

stage: str

count: int

total_value: float

avg_value: float

__init__(stage, count, total_value, avg_value)

Parameters:

• stage (str)

• count (int)

• total_value (float)

• avg_value (float)

Return type:

None

class siege_utilities.analytics.ConversionRate

Bases: object

Conversion rate between consecutive stages.

from_stage: str

to_stage: str

rate: float

from_count: int

to_count: int

__init__(from_stage, to_stage, rate, from_count, to_count)

Parameters:

• from_stage (str)

• to_stage (str)

• rate (float)

• from_count (int)

• to_count (int)

Return type:

None

class siege_utilities.analytics.VelocityMetrics

Bases: object

Time-in-stage metrics.

stage: str

avg_days: float

max_days: float

aging_items: list[str]

__init__(stage, avg_days, max_days, aging_items)

Parameters:

• stage (str)

• avg_days (float)

• max_days (float)

• aging_items (list[str])

Return type:

None

class siege_utilities.analytics.ConcentrationAlert

Bases: object

Alert when a single item dominates the pipeline.

item_id: str

share: float

stage: str

value: float

__init__(item_id, share, stage, value)

Parameters:

• item_id (str)

• share (float)

• stage (str)

• value (float)

Return type:

None

class siege_utilities.analytics.PipelineReport

Bases: object

Complete pipeline health report.

coverage_ratio: float | None

stage_metrics: list[StageMetrics]

conversions: list[ConversionRate]

velocity: list[VelocityMetrics]

concentration_alerts: list[ConcentrationAlert]

total_value: float

target: float | None

__init__(coverage_ratio, stage_metrics, conversions, velocity, concentration_alerts, total_value, target)

Parameters:

• coverage_ratio (float | None)

• stage_metrics (list[StageMetrics])

• conversions (list[ConversionRate])

• velocity (list[VelocityMetrics])

• concentration_alerts (list[ConcentrationAlert])

• total_value (float)

• target (float | None)

Return type:

None

class siege_utilities.analytics.PipelineAnalyzer

Bases: object

Pipeline health metrics engine.

Parameters:

• stages (list[str]) – Ordered stage names (first = entry, last = exit).

• target (float | None) – Target pipeline value for coverage ratio calculation.

• aging_threshold (float) – Multiplier for aging alerts (default 2.0 = alert at 2x average).

• concentration_threshold (float) – Maximum single-item share before alert (default 0.40 = 40%).

Raises:

ValueError – If stages is empty or thresholds are invalid.

__init__(stages, target=None, aging_threshold=2.0, concentration_threshold=0.4)

Parameters:

• stages (list[str])

• target (float | None)

• aging_threshold (float)

• concentration_threshold (float)

Return type:

None

analyze(items, *, as_of=None)

Analyze pipeline health.

Parameters:

• items (list[PipelineItem]) – Items currently in the pipeline.

• as_of (datetime, optional) – Reference time for velocity calculation. Defaults to datetime.now().

Return type:

PipelineReport

Raises:

ValueError – If items is empty or contains unknown stages.

class siege_utilities.analytics.MappingRule

Bases: object

A signal-to-action mapping rule.

Parameters:

• signal (str) – The categorical input signal.

• action (str) – The recommended action.

• priority (int) – Lower number = higher priority. Used when multiple rules match.

signal: str

action: str

priority: int = 0

__init__(signal, action, priority=0)

Parameters:

• signal (str)

• action (str)

• priority (int)

Return type:

None

class siege_utilities.analytics.Outcome

Bases: object

Recorded outcome for an intervention.

signal: str

action: str

success: bool

overridden: bool = False

override_reason: str | None = None

__init__(signal, action, success, overridden=False, override_reason=None)

Parameters:

• signal (str)

• action (str)

• success (bool)

• overridden (bool)

• override_reason (str | None)

Return type:

None

class siege_utilities.analytics.EffectivenessReport

Bases: object

Effectiveness statistics for one action.

action: str

total: int

successes: int

success_rate: float

__init__(action, total, successes, success_rate)

Parameters:

• action (str)

• total (int)

• successes (int)

• success_rate (float)

Return type:

None

class siege_utilities.analytics.InterventionMapper

Bases: object

Rule-based intervention mapping engine.

Parameters:

• rules (list[MappingRule]) – Signal-to-action mapping rules.

• default_action (str | None) – Action for unmapped signals. If None, unmapped signals raise.

• strict (bool) – If True, unmapped signals raise ValueError. If False, warn and return default_action.

Raises:

ValueError – If rules is empty or strict=False without default_action.

__init__(rules, default_action=None, strict=True)

Parameters:

• rules (list[MappingRule])

• default_action (str | None)

• strict (bool)

Return type:

None

property signals: list[str]

map(signal, *, override_action=None, override_reason=None)

Map a signal to an action.

Parameters:

• signal (str) – The categorical input signal.

• override_action (str, optional) – Override the mapped action. Must include override_reason.

• override_reason (str, optional) – Reason for the override (mandatory when overriding).

Returns:

The recommended action.

Return type:

str

Raises:

ValueError – If signal is unmapped and strict=True, or if override is given without a reason.

record_outcome(signal, action, success, *, overridden=False, override_reason=None)

Record the outcome of an intervention.

Parameters:

• signal (str) – The signal that triggered the intervention.

• action (str) – The action taken.

• success (bool) – Whether the intervention succeeded.

• overridden (bool) – Whether the action was an override.

• override_reason (str, optional) – Reason for override (required if overridden=True).

Raises:

ValueError – If overridden=True without override_reason.

Return type:

None

effectiveness()

Compute effectiveness per action.

Returns:

One entry per distinct action, sorted by success rate ascending.

Return type:

list[EffectivenessReport]

Raises:

ValueError – If no outcomes have been recorded.

Facebook Business Analytics

Facebook Business Integration Module

This module provides comprehensive Facebook Business integration capabilities: - Connect to Facebook Business accounts via OAuth2 - Retrieve data from Facebook Ads, Pages, and Business Manager - Save data as Pandas or Spark dataframes - Associate Facebook accounts with client profiles - Batch data retrieval and processing

class siege_utilities.analytics.facebook_business.FacebookBusinessConnector

Bases: object

Facebook Business connection and data retrieval manager.

__init__(access_token, app_id=None, app_secret=None)

Initialize Facebook Business connector.

Parameters:

• access_token (str) – Facebook access token

• app_id (str) – Facebook app ID (optional)

• app_secret (str) – Facebook app secret (optional)

get_ad_accounts()

Get list of accessible ad accounts.

Returns:

List of ad account information

Return type:

List[Dict[str, Any]]

get_ad_insights(ad_account_id, start_date, end_date, fields=None, breakdowns=None)

Retrieve ad insights from Facebook Ads.

Parameters:

• ad_account_id (str) – Facebook ad account ID

• start_date (str) – Start date (YYYY-MM-DD)

• end_date (str) – End date (YYYY-MM-DD)

• fields (List[str]) – List of insight fields to retrieve

• breakdowns (List[str]) – List of breakdowns to apply

Returns:

Pandas DataFrame with ad insights

Return type:

pandas.DataFrame

get_page_insights(page_id, start_date, end_date, metrics=None)

Retrieve page insights from Facebook Pages.

Parameters:

• page_id (str) – Facebook page ID

• start_date (str) – Start date (YYYY-MM-DD)

• end_date (str) – End date (YYYY-MM-DD)

• metrics (List[str]) – List of metrics to retrieve

Returns:

Pandas DataFrame with page insights

Return type:

pandas.DataFrame

get_business_insights(business_id, start_date, end_date, metrics=None)

Retrieve business insights from Facebook Business Manager.

Parameters:

• business_id (str) – Facebook business ID

• start_date (str) – Start date (YYYY-MM-DD)

• end_date (str) – End date (YYYY-MM-DD)

• metrics (List[str]) – List of metrics to retrieve

Returns:

Pandas DataFrame with business insights

Return type:

pandas.DataFrame

save_as_pandas(df, output_path, format='parquet')

Save DataFrame as Pandas format.

Parameters:

• df (pandas.DataFrame) – Pandas DataFrame to save

• output_path (str) – Output file path

• format (str) – Output format (parquet, csv, excel, etc.)

Raises:

• ValueError – If the format is unsupported.

• OSError – If the file cannot be written.

Return type:

None

save_as_spark(df, output_path, spark_session=None)

Save DataFrame as Spark DataFrame and optionally to storage.

Parameters:

• df (pandas.DataFrame) – Pandas DataFrame to convert

• output_path (str) – Output path for Spark DataFrame

• spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)

Raises:

• ImportError – If PySpark is not available.

• OSError – If the output path cannot be written.

Return type:

None

siege_utilities.analytics.facebook_business.batch_retrieve_facebook_data(client_id, start_date, end_date, data_types=None, output_format='pandas', output_directory='output/facebook_data')

Batch retrieve Facebook data for all accounts associated with a client.

Parameters:

• client_id (str) – Client ID to retrieve data for

• start_date (str) – Start date for data retrieval

• end_date (str) – End date for data retrieval

• data_types (List[str]) – Types of data to retrieve (ads, pages, business)

• output_format (str) – Output format (pandas, spark, or both)

• output_directory (str) – Directory to save output files

Returns:

Dictionary with retrieval results

Return type:

Dict[str, Any]

siege_utilities.analytics.facebook_business.create_facebook_account_profile(client_id, fb_account_id, account_type='ad_account', access_token=None)

Create a Facebook account profile linked to a client.

Parameters:

• client_id (str) – Client ID from client management system

• fb_account_id (str) – Facebook account ID

• account_type (str) – Type of FB account (ad_account, page, business)

• access_token (str) – Facebook access token

Returns:

Facebook account profile dictionary

Return type:

Dict[str, Any]

siege_utilities.analytics.facebook_business.list_facebook_accounts_for_client(client_id, config_directory='config')

List all Facebook accounts associated with a client.

Parameters:

• client_id (str) – Client ID to search for

• config_directory (str) – Directory containing config files

Returns:

List of Facebook account profiles

Return type:

List[Dict[str, Any]]

siege_utilities.analytics.facebook_business.load_facebook_account_profile(account_id, config_directory='config')

Load Facebook account profile from JSON file.

Parameters:

• account_id (str) – Facebook account ID to load

• config_directory (str) – Directory containing config files

Returns:

Facebook account profile dictionary or None if not found

Return type:

Dict[str, Any] | None

siege_utilities.analytics.facebook_business.save_facebook_account_profile(profile, config_directory='config')

Save Facebook account profile to JSON file.

Parameters:

• profile (Dict[str, Any]) – Facebook account profile dictionary

• config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Return type:

str

Functions

Usage Examples

Facebook Business API setup:

Campaign data retrieval:

Ad insights and performance:

Google Analytics

Google Analytics Integration Module

This module provides comprehensive Google Analytics integration capabilities: - Connect to GA accounts via OAuth2 - Retrieve data from GA4 and Universal Analytics - Save data as Pandas or Spark dataframes - Associate GA accounts with client profiles - Batch data retrieval and processing

class siege_utilities.analytics.google_analytics.GoogleAnalyticsConnector

Bases: object

Google Analytics connection and data retrieval manager.

__init__(client_id=None, client_secret=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', auth_method='oauth2', service_account_data=None)

Initialize Google Analytics connector.

Parameters:

• client_id (str) – OAuth2 client ID from Google Cloud Console (for OAuth2)

• client_secret (str) – OAuth2 client secret from Google Cloud Console (for OAuth2)

• redirect_uri (str) – OAuth2 redirect URI (for OAuth2)

• auth_method (str) – Authentication method (“oauth2” or “service_account”)

• service_account_data (Dict[str, Any]) – Service account credentials dict (for service account auth)

authenticate(token_file='ga_token.json')

Authenticate with Google Analytics using OAuth2.

Parameters:

token_file (str) – Path to store/retrieve OAuth token

Raises:

• google.auth.exceptions.* – On authentication failure.

• OSError – If token file cannot be read/written.

Return type:

None

authenticate_service_account()

Authenticate with Google Analytics using service account. Based on working implementation from GA project.

Raises:

• ValueError – If no service account data is available.

• google.auth.exceptions.* – On authentication failure.

• OSError – If temp file operations fail.

Return type:

None

get_ga4_data(property_id, start_date, end_date, metrics, dimensions=None, row_limit=100000)

Retrieve data from Google Analytics 4.

Parameters:

• property_id (str) – GA4 property ID

• start_date (str) – Start date as YYYY-MM-DD (also accepts the GA4 relative keywords today, yesterday, or NdaysAgo). Interpreted in the property’s configured timezone, not the caller’s local time — set this in the GA admin UI.

• end_date (str) – End date, same format / TZ rules as start_date.

• metrics (List[str]) – List of metrics to retrieve

• dimensions (List[str]) – List of dimensions to retrieve

• row_limit (int) – Maximum rows to retrieve

Returns:

Pandas DataFrame with GA4 data

Raises:

ValueError – If start_date / end_date is not a valid YYYY-MM-DD date or recognized GA4 relative keyword. Previously, malformed dates were forwarded raw to the GA API and returned an opaque 400 hours later in a batch run.

Return type:

pandas.DataFrame

get_ua_data(view_id, start_date, end_date, metrics, dimensions=None, max_results=100000)

Retrieve data from Universal Analytics.

Parameters:

• view_id (str) – UA view ID

• start_date (str) – Start date (YYYY-MM-DD)

• end_date (str) – End date (YYYY-MM-DD)

• metrics (List[str]) – List of metrics to retrieve

• dimensions (List[str]) – List of dimensions to retrieve

• max_results (int) – Maximum results to retrieve

Returns:

Pandas DataFrame with UA data

Raises:

ValueError – If start_date / end_date is not a valid YYYY-MM-DD date or recognized GA relative keyword.

Return type:

pandas.DataFrame

save_as_pandas(df, output_path, format='parquet')

Save DataFrame as Pandas format.

Parameters:

• df (pandas.DataFrame) – Pandas DataFrame to save

• output_path (str) – Output file path

• format (str) – Output format (parquet, csv, excel, etc.)

Raises:

• ValueError – If the format is unsupported.

• OSError – If the file cannot be written.

Return type:

None

save_as_spark(df, output_path, spark_session=None)

Save DataFrame as Spark DataFrame and optionally to storage.

Parameters:

• df (pandas.DataFrame) – Pandas DataFrame to convert

• output_path (str) – Output path for Spark DataFrame

• spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)

Raises:

• ImportError – If PySpark is not available.

• OSError – If the output path cannot be written.

Return type:

None

siege_utilities.analytics.google_analytics.batch_retrieve_ga_data(client_id, start_date, end_date, metrics, dimensions=None, output_format='pandas', output_directory='output/ga_data')

Batch retrieve GA data for all accounts associated with a client.

Parameters:

• client_id (str) – Client ID to retrieve data for

• start_date (str) – Start date for data retrieval

• end_date (str) – End date for data retrieval

• metrics (List[str]) – List of metrics to retrieve

• dimensions (List[str]) – List of dimensions to retrieve

• output_format (str) – Output format (pandas, spark, or both)

• output_directory (str) – Directory to save output files

Returns:

Dictionary with retrieval results

Return type:

Dict[str, Any]

siege_utilities.analytics.google_analytics.create_ga_account_profile(client_id, ga_property_id, account_type='ga4', credentials_file=None)

Create a Google Analytics account profile linked to a client.

Parameters:

• client_id (str) – Client ID from client management system

• ga_property_id (str) – Google Analytics property ID

• account_type (str) – Type of GA account (ga4, ua, or both)

• credentials_file (str) – Path to OAuth credentials file

Returns:

GA account profile dictionary

Return type:

Dict[str, Any]

siege_utilities.analytics.google_analytics.create_ga_connector_from_1password(item_title='Google Analytics Service Account - Multi-Client Reporter')

Create a GoogleAnalyticsConnector using service account from 1Password.

Deprecated since version The: auth mechanism (1Password) should not leak into an analytics API surface. Use the explicit two-step path instead:

from siege_utilities.config import get_google_service_account_from_1password
from siege_utilities.analytics import create_ga_connector_with_service_account

data = get_google_service_account_from_1password(item_title)
connector = create_ga_connector_with_service_account(data)

This function will be removed in v4.0.0. See docs/INTENT.md (D1) and ELE-2442 for the rationale.

Parameters:

item_title (str) – Title of the 1Password item containing service account

Returns:

GoogleAnalyticsConnector instance or None if failed

Return type:

GoogleAnalyticsConnector | None

siege_utilities.analytics.google_analytics.create_ga_connector_with_oauth2(client_id, client_secret, redirect_uri='urn:ietf:wg:oauth:2.0:oob')

Create a GoogleAnalyticsConnector using OAuth2 authentication.

Parameters:

• client_id (str) – OAuth2 client ID from Google Cloud Console

• client_secret (str) – OAuth2 client secret from Google Cloud Console

• redirect_uri (str) – OAuth2 redirect URI

Returns:

GoogleAnalyticsConnector instance configured for OAuth2 auth

Return type:

GoogleAnalyticsConnector

siege_utilities.analytics.google_analytics.create_ga_connector_with_service_account(service_account_data=None)

Create a GoogleAnalyticsConnector using service account authentication.

Parameters:

service_account_data (Dict[str, Any]) – Service account credentials dict (optional, will try 1Password if None)

Returns:

GoogleAnalyticsConnector instance configured for service account auth

Return type:

GoogleAnalyticsConnector

siege_utilities.analytics.google_analytics.list_ga_accounts_for_client(client_id, config_directory='config')

List all GA accounts associated with a client.

Parameters:

• client_id (str) – Client ID to search for

• config_directory (str) – Directory containing config files

Returns:

List of GA account profiles

Return type:

List[Dict[str, Any]]

siege_utilities.analytics.google_analytics.load_ga_account_profile(account_id, config_directory='config')

Load GA account profile from JSON file.

Parameters:

• account_id (str) – GA account ID to load

• config_directory (str) – Directory containing config files

Returns:

GA account profile dictionary or None if not found

Return type:

Dict[str, Any] | None

siege_utilities.analytics.google_analytics.save_ga_account_profile(profile, config_directory='config')

Save GA account profile to JSON file.

Parameters:

• profile (Dict[str, Any]) – GA account profile dictionary

• config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Return type:

str

Functions

Usage Examples

Google Analytics data retrieval:

User behavior analysis:

Page performance analysis:

Unit Tests

The analytics modules have comprehensive test coverage:

Test Results: All analytics tests pass successfully with comprehensive coverage.