Analytics
Analytics connectors: Google Analytics, Snowflake, Facebook Business.
Analytics Module — lazy-loaded.
Provides analytics integration: Google Analytics, Facebook Business, Snowflake connectors, and Google Workspace write APIs (Sheets, Docs, Slides).
- class siege_utilities.analytics.GoogleAnalyticsConnector[source]
Bases:
objectGoogle 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)[source]
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_service_account()[source]
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)[source]
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 keywordstoday,yesterday, orNdaysAgo). 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.row_limit (int) – Maximum rows to retrieve
- Returns:
Pandas DataFrame with GA4 data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate 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:
- get_ua_data(view_id, start_date, end_date, metrics, dimensions=None, max_results=100000)[source]
Retrieve data from Universal Analytics.
- Parameters:
- Returns:
Pandas DataFrame with UA data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA relative keyword.- Return type:
- save_as_pandas(df, output_path, format='parquet')[source]
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)[source]
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.create_ga_account_profile(client_id, ga_property_id, account_type='ga4', credentials_file=None)[source]
Create a Google Analytics account profile linked to a client.
- Parameters:
- Returns:
GA account profile dictionary
- Return type:
- siege_utilities.analytics.save_ga_account_profile(profile, config_directory=None)[source]
Save GA account profile to JSON file.
- siege_utilities.analytics.load_ga_account_profile(account_id, config_directory=None)[source]
Load GA account profile from JSON file.
- siege_utilities.analytics.list_ga_accounts_for_client(client_id, config_directory=None)[source]
List all GA accounts associated with a client.
- siege_utilities.analytics.batch_retrieve_ga_data(client_id, start_date, end_date, metrics, dimensions=None, output_format='pandas', output_directory='output/ga_data')[source]
Batch retrieve GA data for all accounts associated with a client.
- Parameters:
- Returns:
Dictionary with retrieval results
- Return type:
- siege_utilities.analytics.create_ga_connector_with_service_account(service_account_data=None)[source]
Create a GoogleAnalyticsConnector using service account authentication.
- class siege_utilities.analytics.FacebookBusinessConnector[source]
Bases:
objectFacebook Business connection and data retrieval manager.
- __init__(access_token, app_id=None, app_secret=None)[source]
Initialize Facebook Business connector.
- get_ad_insights(ad_account_id, start_date, end_date, fields=None, breakdowns=None)[source]
Retrieve ad insights from Facebook Ads.
- Parameters:
- Returns:
Pandas DataFrame with ad insights
- Return type:
- get_page_insights(page_id, start_date, end_date, metrics=None)[source]
Retrieve page insights from Facebook Pages.
- get_business_insights(business_id, start_date, end_date, metrics=None)[source]
Retrieve business insights from Facebook Business Manager.
- save_as_pandas(df, output_path, format='parquet')[source]
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)[source]
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.create_facebook_account_profile(client_id, fb_account_id, account_type='ad_account', access_token=None)[source]
Create a Facebook account profile linked to a client.
- siege_utilities.analytics.save_facebook_account_profile(profile, config_directory=None)[source]
Save Facebook account profile to JSON file.
- siege_utilities.analytics.load_facebook_account_profile(account_id, config_directory=None)[source]
Load Facebook account profile from JSON file.
- siege_utilities.analytics.list_facebook_accounts_for_client(client_id, config_directory=None)[source]
List all Facebook accounts associated with a client.
- siege_utilities.analytics.batch_retrieve_facebook_data(client_id, start_date, end_date, data_types=None, output_format='pandas', output_directory='output/facebook_data')[source]
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:
- class siege_utilities.analytics.SnowflakeConnector[source]
Bases:
objectSnowflake data warehouse connector with advanced features.
- __init__(account=None, user=None, password=None, warehouse=None, database=None, schema=None, role=None, config_file=None)[source]
Initialize Snowflake connector.
- Parameters:
account (str | None) – Snowflake account identifier
user (str | None) – 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()[source]
Establish connection to Snowflake.
- Raises:
ConnectionError – If the connection attempt fails.
- Return type:
None
- execute_query(query, params=None)[source]
Execute a SQL query.
- Parameters:
- Returns:
Query results as list of tuples
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If query execution fails.
- Return type:
- execute_ddl(ddl_statement)[source]
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)[source]
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)[source]
Download data from Snowflake as pandas DataFrame.
- Parameters:
- 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)[source]
Get information about a Snowflake table.
- Parameters:
- Returns:
Dictionary with table information
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- list_tables(database=None, schema=None)[source]
List all tables in a database/schema.
- Parameters:
- Returns:
List of table names
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- siege_utilities.analytics.get_snowflake_connector(config_file=None)[source]
Get Snowflake connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.upload_to_snowflake(df, table_name, config_file=None, **kwargs)[source]
Convenience function to upload DataFrame to Snowflake.
- siege_utilities.analytics.download_from_snowflake(query, config_file=None, **kwargs)[source]
Convenience function to download DataFrame from Snowflake.
- siege_utilities.analytics.execute_snowflake_query(query, config_file=None, **kwargs)[source]
Convenience function to execute Snowflake query.
- class siege_utilities.analytics.GoogleWorkspaceClient[source]
Bases:
objectAuthenticated client that builds Google API service objects.
Do not instantiate directly — use the
from_oauth()orfrom_service_account()class methods.- classmethod from_oauth(client_id, client_secret, token_file=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', scopes=None)[source]
Authenticate via OAuth2 interactive flow.
If token_file exists and contains a valid/refreshable token, no browser interaction is needed.
- classmethod from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account='TLTQ3ANAABGCNEK7KIAOTDNK2Q', token_file=None, scopes=None)[source]
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:
- classmethod from_service_account(service_account_data=None, service_account_file=None, scopes=None)[source]
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().
- classmethod from_credentials(credentials)[source]
Wrap an already-authenticated
google.auth.credentials.Credentials.- Return type:
- classmethod from_account(account, person=None, scopes=None)[source]
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:
- classmethod from_registry(registry, google_account_id=None, person=None, scopes=None)[source]
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:
- property credentials
- batch_update_presentation(presentation_id, requests)[source]
Execute a batch of Slides API requests.
- static spreadsheet_url(spreadsheet_id)[source]
Return the live Google Sheets URL for a spreadsheet ID.
- static presentation_url(presentation_id)[source]
Return the live Google Slides URL for a presentation ID.
- static file_url(file_id, mime_type=None)[source]
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.
- copy_file(file_id, title=None)[source]
Copy a Drive file (spreadsheet, doc, presentation) and return the new ID.
Share a Drive file with a user.
- upload_file(local_path, name=None, mime_type=None, folder_id=None)[source]
Upload a local file to Drive and return its file ID.
- Parameters:
- Returns:
The new Drive file’s ID.
- Return type:
- siege_utilities.analytics.create_spreadsheet(client, title, sheet_names=None, folder_id=None)[source]
Create a new Google Spreadsheet and return its ID.
- Parameters:
- Returns:
The spreadsheet ID string.
- Return type:
- siege_utilities.analytics.write_values(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Write a 2-D list of values to a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.append_rows(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Append rows after existing data in a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.read_values(client, spreadsheet_id, range_)[source]
Read values from a range.
- siege_utilities.analytics.write_dataframe(client, spreadsheet_id, df, sheet_name='Sheet1', include_header=True, start_cell='A1')[source]
Write a pandas DataFrame to a sheet.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.read_dataframe(client, spreadsheet_id, range_='Sheet1', has_header=True)[source]
Read a range from a spreadsheet into a pandas DataFrame.
- siege_utilities.analytics.add_sheet(client, spreadsheet_id, title)[source]
Add a new tab/sheet to an existing spreadsheet.
- siege_utilities.analytics.get_spreadsheet_metadata(client, spreadsheet_id)[source]
Fetch spreadsheet metadata (title, sheets, locale, etc.).
- siege_utilities.analytics.copy_spreadsheet(client, spreadsheet_id, title=None)[source]
Copy an entire spreadsheet via the Drive API.
- siege_utilities.analytics.create_presentation(client, title, folder_id=None)[source]
Create a new Google Slides presentation and return its ID.
- siege_utilities.analytics.get_presentation(client, presentation_id)[source]
Fetch full presentation metadata.
- siege_utilities.analytics.copy_presentation(client, presentation_id, title=None)[source]
Copy an entire presentation via the Drive API.
- siege_utilities.analytics.add_blank_slide(client, presentation_id, layout='BLANK', insertion_index=None)[source]
Add a blank slide to a presentation.
- Parameters:
- Returns:
The new slide’s object ID.
- Return type:
- siege_utilities.analytics.create_textbox(client, presentation_id, slide_id, text, left=100, top=100, width=400, height=50)[source]
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:
- siege_utilities.analytics.create_document(client, title, folder_id=None)[source]
Create a new Google Doc and return its document ID.
- siege_utilities.analytics.get_document(client, document_id)[source]
Fetch full document metadata and content structure.
- siege_utilities.analytics.copy_document(client, document_id, title=None)[source]
Copy a document via the Drive API.
- siege_utilities.analytics.read_document_text(client, document_id)[source]
Read the full plain-text content of a document.
Extracts text from all structural elements (paragraphs, tables, etc.).
- siege_utilities.analytics.insert_paragraph(client, document_id, text, index=1, heading=None)[source]
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", orNonefor normal text).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.insert_table(client, document_id, rows, cols, index=1)[source]
Insert an empty table at the given index.
- siege_utilities.analytics.replace_text(client, document_id, find, replace_with, match_case=True)[source]
Find and replace text throughout a document.
- class siege_utilities.analytics.VistaSocialConnector[source]
Bases:
objectBearer-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:
VistaSocialAuthErroron 401/403,VistaSocialRateLimitErroron 429,VistaSocialErroron 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)[source]
- list_accounts()[source]
Return the list of accounts this token can access.
Wraps
GET /v1/accounts. Returns the decodeddatalist from the response, or an empty list if the API returned nothing. Raises on auth / rate-limit / network failures (no silent empty-on-error).
- list_profiles(account_id)[source]
Return social profiles connected to account_id.
Wraps
GET /v1/accounts/{account_id}/profiles.
- get_account_analytics(account_id, *, start_date, end_date, metrics=None)[source]
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:
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)[source]
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.
- class siege_utilities.analytics.VistaSocialResponse[source]
Bases:
objectStructured response from a Vista Social API call.
The connector returns this rather than the raw
requests.Responseso consumers don’t need to know about the underlying HTTP library.- 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).
- exception siege_utilities.analytics.VistaSocialError[source]
Bases:
RuntimeErrorBase class for all Vista Social connector failures.
- exception siege_utilities.analytics.VistaSocialAuthError[source]
Bases:
VistaSocialErrorAuthentication failed (401/403). Token is missing, wrong, or expired.
- exception siege_utilities.analytics.VistaSocialRateLimitError[source]
Bases:
VistaSocialErrorAPI rate limit hit (429). Caller can back off and retry later.
- class siege_utilities.analytics.Dimension[source]
Bases:
objectA scoring dimension with a name and weight (0-100).
- class siege_utilities.analytics.ThresholdConfig[source]
Bases:
objectTier thresholds for score classification.
Scores >= green_min are Green, >= yellow_min are Yellow, else Red.
- class siege_utilities.analytics.HealthScorer[source]
Bases:
objectWeighted 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)[source]
- Parameters:
default_thresholds (ThresholdConfig | None)
segment_thresholds (dict[str, ThresholdConfig] | None)
- Return type:
None
- score(entity_id, scores, *, segment=None, previous_score=None)[source]
Score an entity across all dimensions.
- Parameters:
- Return type:
- Raises:
ValueError – If scores dict is missing dimensions or values are out of range.
- score_batch(entities)[source]
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:
- Raises:
ValueError – If any entity dict is missing required keys.
- class siege_utilities.analytics.SignalType[source]
Bases:
EnumTypes of risk signals.
- DECLINE = 'decline'
- THRESHOLD = 'threshold'
- ABSENCE = 'absence'
- ANOMALY = 'anomaly'
- class siege_utilities.analytics.RiskTier[source]
Bases:
EnumRisk classification tiers, ordered by severity.
- CRITICAL = 'Critical'
- HIGH = 'High'
- MEDIUM = 'Medium'
- LOW = 'Low'
- class siege_utilities.analytics.RiskSignal[source]
Bases:
objectA 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.
- signal_type: SignalType
- __init__(name, signal_type, weight)
- Parameters:
name (str)
signal_type (SignalType)
weight (float)
- Return type:
None
- class siege_utilities.analytics.RiskTierConfig[source]
Bases:
objectThresholds for risk tier classification.
Scores >= critical_min are Critical, >= high_min are High, etc.
- class siege_utilities.analytics.RiskAssessment[source]
Bases:
objectResult of a risk assessment.
- signal_types: dict[str, SignalType]
- class siege_utilities.analytics.RiskAnalyzer[source]
Bases:
objectMulti-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)[source]
- Parameters:
signals (list[RiskSignal])
tier_config (RiskTierConfig | None)
- Return type:
None
- assess(entity_id, scores)[source]
Assess risk for an entity.
- Parameters:
- Return type:
- Raises:
ValueError – If scores dict is missing signals or values are out of range.
- class siege_utilities.analytics.DimensionScore[source]
Bases:
objectA score for one entity on one dimension.
- Parameters:
- class siege_utilities.analytics.ComparisonResult[source]
Bases:
objectResult of comparing entities across dimensions.
- class siege_utilities.analytics.FeatureMatrix[source]
Bases:
objectBoolean feature support matrix with coverage scoring.
- class siege_utilities.analytics.ComparativeAnalyzer[source]
Bases:
objectN-entity x M-dimension comparison engine.
- Parameters:
- Raises:
ValueError – If dimensions is empty or weights don’t match dimensions.
- compare(entity_scores)[source]
Compare entities across all dimensions.
- Parameters:
entity_scores (dict[str, dict[str, DimensionScore]]) – Outer key = entity name, inner key = dimension name.
- Return type:
- Raises:
ValueError – If any entity is missing a dimension score.
- static build_feature_matrix(entity_features)[source]
Build a boolean feature support matrix with coverage scores.
- class siege_utilities.analytics.ForecastAccuracy[source]
Bases:
objectResult of a forecast accuracy calculation.
- class siege_utilities.analytics.CategoryBreakdown[source]
Bases:
objectPer-category forecast accuracy.
- accuracy: ForecastAccuracy
- __init__(category, accuracy)
- Parameters:
category (str)
accuracy (ForecastAccuracy)
- Return type:
None
- class siege_utilities.analytics.ForecastReport[source]
Bases:
objectComplete forecast accuracy report.
- overall: ForecastAccuracy
- by_category: list[CategoryBreakdown]
- trend: list[TrendPoint]
- __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[source]
Bases:
objectForecast accuracy engine.
- Parameters:
- Raises:
ValueError – If inputs have mismatched lengths or are empty.
- class siege_utilities.analytics.PipelineItem[source]
Bases:
objectA single item in the pipeline.
- Parameters:
- class siege_utilities.analytics.StageMetrics[source]
Bases:
objectMetrics for a single pipeline stage.
- class siege_utilities.analytics.ConversionRate[source]
Bases:
objectConversion rate between consecutive stages.
- class siege_utilities.analytics.ConcentrationAlert[source]
Bases:
objectAlert when a single item dominates the pipeline.
- class siege_utilities.analytics.PipelineReport[source]
Bases:
objectComplete pipeline health report.
- stage_metrics: list[StageMetrics]
- conversions: list[ConversionRate]
- velocity: list[VelocityMetrics]
- concentration_alerts: list[ConcentrationAlert]
- __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[source]
Bases:
objectPipeline 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.
- analyze(items, *, as_of=None)[source]
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:
- Raises:
ValueError – If items is empty or contains unknown stages.
- class siege_utilities.analytics.MappingRule[source]
Bases:
objectA signal-to-action mapping rule.
- Parameters:
- class siege_utilities.analytics.Outcome[source]
Bases:
objectRecorded outcome for an intervention.
- class siege_utilities.analytics.EffectivenessReport[source]
Bases:
objectEffectiveness statistics for one action.
- class siege_utilities.analytics.InterventionMapper[source]
Bases:
objectRule-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)[source]
- Parameters:
rules (list[MappingRule])
default_action (str | None)
strict (bool)
- Return type:
None
- map(signal, *, override_action=None, override_reason=None)[source]
Map a signal to an action.
- Parameters:
- Returns:
The recommended action.
- Return type:
- 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)[source]
Record the outcome of an intervention.
- Parameters:
- Raises:
ValueError – If overridden=True without override_reason.
- Return type:
None
- effectiveness()[source]
Compute effectiveness per action.
- Returns:
One entry per distinct action, sorted by success rate ascending.
- Return type:
- Raises:
ValueError – If no outcomes have been recorded.
Submodules
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[source]
Bases:
objectGoogle 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)[source]
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_service_account()[source]
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)[source]
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 keywordstoday,yesterday, orNdaysAgo). 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.row_limit (int) – Maximum rows to retrieve
- Returns:
Pandas DataFrame with GA4 data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate 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:
- get_ua_data(view_id, start_date, end_date, metrics, dimensions=None, max_results=100000)[source]
Retrieve data from Universal Analytics.
- Parameters:
- Returns:
Pandas DataFrame with UA data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA relative keyword.- Return type:
- save_as_pandas(df, output_path, format='parquet')[source]
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)[source]
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')[source]
Batch retrieve GA data for all accounts associated with a client.
- Parameters:
- Returns:
Dictionary with retrieval results
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_account_profile(client_id, ga_property_id, account_type='ga4', credentials_file=None)[source]
Create a Google Analytics account profile linked to a client.
- Parameters:
- Returns:
GA account profile dictionary
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_connector_from_1password(item_title='Google Analytics Service Account - Multi-Client Reporter')[source]
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')[source]
Create a GoogleAnalyticsConnector using OAuth2 authentication.
- Parameters:
- Returns:
GoogleAnalyticsConnector instance configured for OAuth2 auth
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_connector_with_service_account(service_account_data=None)[source]
Create a GoogleAnalyticsConnector using service account authentication.
- siege_utilities.analytics.google_analytics.list_ga_accounts_for_client(client_id, config_directory=None)[source]
List all GA accounts associated with a client.
- siege_utilities.analytics.google_analytics.load_ga_account_profile(account_id, config_directory=None)[source]
Load GA account profile from JSON file.
- siege_utilities.analytics.google_analytics.save_ga_account_profile(profile, config_directory=None)[source]
Save GA account profile to JSON file.
Google Docs write service.
Provides functions for creating documents, inserting text, tables, images, and performing batch updates via the Docs API v1.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_docs import (
create_document, insert_text, insert_table,
)
client = GoogleWorkspaceClient.from_service_account() doc_id = create_document(client, “Meeting Notes”) insert_text(client, doc_id, “Agendan”, bold=True) insert_table(client, doc_id, rows=3, cols=2)
- siege_utilities.analytics.google_docs.batch_update(client, document_id, requests)[source]
Execute a batch of Docs API requests.
Delegates to
GoogleWorkspaceClient.batch_update_document().
- siege_utilities.analytics.google_docs.copy_document(client, document_id, title=None)[source]
Copy a document via the Drive API.
- siege_utilities.analytics.google_docs.create_document(client, title, folder_id=None)[source]
Create a new Google Doc and return its document ID.
- siege_utilities.analytics.google_docs.get_document(client, document_id)[source]
Fetch full document metadata and content structure.
- siege_utilities.analytics.google_docs.insert_image(client, document_id, image_uri, index=1, width=None, height=None)[source]
Insert an image from a URL.
The URI must be publicly accessible or a Google Drive file.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.insert_paragraph(client, document_id, text, index=1, heading=None)[source]
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", orNonefor normal text).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.insert_table(client, document_id, rows, cols, index=1)[source]
Insert an empty table at the given index.
- siege_utilities.analytics.google_docs.insert_text(client, document_id, text, index=1, bold=False, italic=False, font_size=None)[source]
Insert text at a given index in a document.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
document_id (str) – Target document ID.
text (str) – Text to insert.
index (int) – Character index (1-based; 1 = start of document body).
bold (bool) – Apply bold formatting.
italic (bool) – Apply italic formatting.
font_size (int | None) – Font size in points (
None= default).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.read_document_text(client, document_id)[source]
Read the full plain-text content of a document.
Extracts text from all structural elements (paragraphs, tables, etc.).
- siege_utilities.analytics.google_docs.replace_text(client, document_id, find, replace_with, match_case=True)[source]
Find and replace text throughout a document.
Google Sheets write service.
Provides functions for creating spreadsheets, writing data, managing tabs (sheets), and performing batch updates via the Sheets API v4.
All functions accept a GoogleWorkspaceClient for authentication.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_sheets import (
create_spreadsheet, write_dataframe, append_rows,
)
client = GoogleWorkspaceClient.from_service_account() spreadsheet_id = create_spreadsheet(client, “Q1 Report”) write_dataframe(client, spreadsheet_id, df)
- siege_utilities.analytics.google_sheets.add_sheet(client, spreadsheet_id, title)[source]
Add a new tab/sheet to an existing spreadsheet.
- siege_utilities.analytics.google_sheets.append_rows(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Append rows after existing data in a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_sheets.batch_update(client, spreadsheet_id, requests)[source]
Execute a batch of Sheets API requests.
Delegates to
GoogleWorkspaceClient.batch_update_spreadsheet().
- siege_utilities.analytics.google_sheets.copy_spreadsheet(client, spreadsheet_id, title=None)[source]
Copy an entire spreadsheet via the Drive API.
- siege_utilities.analytics.google_sheets.create_spreadsheet(client, title, sheet_names=None, folder_id=None)[source]
Create a new Google Spreadsheet and return its ID.
- Parameters:
- Returns:
The spreadsheet ID string.
- Return type:
- siege_utilities.analytics.google_sheets.get_spreadsheet_metadata(client, spreadsheet_id)[source]
Fetch spreadsheet metadata (title, sheets, locale, etc.).
- siege_utilities.analytics.google_sheets.read_dataframe(client, spreadsheet_id, range_='Sheet1', has_header=True)[source]
Read a range from a spreadsheet into a pandas DataFrame.
- siege_utilities.analytics.google_sheets.read_values(client, spreadsheet_id, range_)[source]
Read values from a range.
- siege_utilities.analytics.google_sheets.write_dataframe(client, spreadsheet_id, df, sheet_name='Sheet1', include_header=True, start_cell='A1')[source]
Write a pandas DataFrame to a sheet.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_sheets.write_values(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Write a 2-D list of values to a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
Google Slides write service.
Provides functions for creating presentations, adding slides, inserting text and images, and performing batch updates via the Slides API v1.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_slides import (
create_presentation, add_blank_slide, insert_text,
)
client = GoogleWorkspaceClient.from_service_account() pres_id = create_presentation(client, “Q1 Report”) slide_id = add_blank_slide(client, pres_id) insert_text(client, pres_id, slide_id, “Hello World”)
- siege_utilities.analytics.google_slides.add_blank_slide(client, presentation_id, layout='BLANK', insertion_index=None)[source]
Add a blank slide to a presentation.
- Parameters:
- Returns:
The new slide’s object ID.
- Return type:
- siege_utilities.analytics.google_slides.batch_update(client, presentation_id, requests)[source]
Execute a batch of Slides API requests.
Delegates to
GoogleWorkspaceClient.batch_update_presentation().
- siege_utilities.analytics.google_slides.copy_presentation(client, presentation_id, title=None)[source]
Copy an entire presentation via the Drive API.
- siege_utilities.analytics.google_slides.create_argument_slide(client, presentation_id, argument, slide_index=None, folder_id=None, publish_figures=True)[source]
Add one slide for an Argument.
- Layout is derived from argument.layout:
“full_width” → title / narrative / figure stacked vertically “side_by_side” → title top; narrative left, figure right
A figure is inserted through
temporary_public_figure(): it is uploaded tofolder_id, shared anyone-with-the-link while Slides copies it, then deleted from Drive.- Parameters:
client – Authenticated GoogleWorkspaceClient.
presentation_id (str) – Target presentation ID.
argument – Argument dataclass instance.
slide_index (int | None) – Insertion position. None → append at end.
folder_id (str | None) – Drive folder for the temporary figure file. None → the root of the authenticated account’s My Drive.
publish_figures (bool) – If False, the figure is not uploaded, no anyone-with-the-link permission is created, and the slide is built without it (a WARNING is logged).
- Returns:
Slide object ID.
- Return type:
- siege_utilities.analytics.google_slides.create_presentation(client, title, folder_id=None)[source]
Create a new Google Slides presentation and return its ID.
- siege_utilities.analytics.google_slides.create_report_from_arguments(client, title, arguments, folder_id=None, theme_presentation_id=None, publish_figures=True)[source]
Create a complete presentation from a list of Arguments.
- Public exposure of figures:
The Slides API can only insert an image from a URL that is reachable without authentication. With
publish_figures=True(the default), each chart or map is therefore uploaded to Drive (intofolder_idwhen given), shared anyone-with-the-link (reader) for the duration of itscreateImagerequest, and then deleted from Drive, including when the insert fails. While it exists, anyone holding the URL can fetch the image without signing in. If the delete fails, the file ID is logged at ERROR and the file must be removed manually. Passpublish_figures=Falseto never create an anyone permission; figures are then left out of the slides.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
title (str) – Presentation title.
arguments (List) – List of Argument dataclass instances (one slide each).
folder_id (str | None) – Optional Drive folder to place the presentation, and the temporary figure files, in.
theme_presentation_id (str | None) – If given, copies this presentation first (preserves master slides / theme).
publish_figures (bool) – If False, figures are not uploaded or published and are omitted from the slides.
- Returns:
Presentation ID of the newly created report.
- Return type:
- siege_utilities.analytics.google_slides.create_textbox(client, presentation_id, slide_id, text, left=100, top=100, width=400, height=50)[source]
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:
- siege_utilities.analytics.google_slides.get_presentation(client, presentation_id)[source]
Fetch full presentation metadata.
- siege_utilities.analytics.google_slides.insert_image(client, presentation_id, slide_id, image_url, left=100, top=100, width=400, height=300)[source]
Insert an image onto a slide from a URL.
The URL must be publicly accessible or a Google Drive file URL.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
presentation_id (str) – Target presentation ID.
slide_id (str) – Slide object ID.
image_url (str) – Public URL of the image.
left (float) – Position in points.
top (float) – Position in points.
width (float) – Size in points.
height (float) – Size in points.
- Returns:
The image object ID.
- Return type:
- siege_utilities.analytics.google_slides.insert_text(client, presentation_id, object_id, text, insertion_index=0)[source]
Insert text into an existing shape or text box.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_slides.temporary_public_figure(client, figure, filename, folder_id=None)[source]
Upload a figure, publish it for the duration of the block, then delete it.
Slides
createImagefetches the image once, at request time, from a URL that must be reachable without authentication, and stores its own copy. This context manager uploads the figure tofolder_id, grants an anyone-with-the-link reader permission, yields the public URL, and in afinallydeletes the Drive file (which removes the permission too), whether or not the block raised.If the delete fails, the file ID is logged at ERROR. The delete error is raised only when the block itself succeeded; an error from the block is never masked by a cleanup failure.
- Parameters:
- Yields:
The public URL of the uploaded image.
- Return type:
- siege_utilities.analytics.google_slides.upload_figure_to_drive(client, figure, filename, folder_id=None)[source]
Save a matplotlib Figure to a temp PNG, upload to Google Drive, return URL.
Warning
The uploaded file is shared anyone-with-the-link (reader) via
GoogleWorkspaceClient.public_urland is left in Drive. Nothing revokes that permission or deletes the file. For inserting a figure into a slide, usetemporary_public_figure(), which deletes the Drive file once the insert has run.- Parameters:
- Returns:
Public URL string for the uploaded image (suitable for insert_image).
- Return type:
Google Workspace base client for Docs, Sheets, and Slides write APIs.
Provides shared authentication and service-building logic reused by the Sheets, Slides, and Docs service modules.
Authentication follows the same patterns as GoogleAnalyticsConnector: - OAuth2 (interactive flow with token file) - Service Account (from 1Password or file) - Explicit Credentials object
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient
# OAuth2 with token file client = GoogleWorkspaceClient.from_oauth(
client_id=”…”, client_secret=”…”, token_file=”token.json”,
)
# Service Account from 1Password client = GoogleWorkspaceClient.from_service_account()
# Get a specific API service sheets = client.sheets_service() docs = client.docs_service() slides = client.slides_service()
- class siege_utilities.analytics.google_workspace.GoogleWorkspaceClient[source]
Bases:
objectAuthenticated client that builds Google API service objects.
Do not instantiate directly — use the
from_oauth()orfrom_service_account()class methods.- classmethod from_oauth(client_id, client_secret, token_file=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', scopes=None)[source]
Authenticate via OAuth2 interactive flow.
If token_file exists and contains a valid/refreshable token, no browser interaction is needed.
- classmethod from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account='TLTQ3ANAABGCNEK7KIAOTDNK2Q', token_file=None, scopes=None)[source]
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:
- classmethod from_service_account(service_account_data=None, service_account_file=None, scopes=None)[source]
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().
- classmethod from_credentials(credentials)[source]
Wrap an already-authenticated
google.auth.credentials.Credentials.- Return type:
- classmethod from_account(account, person=None, scopes=None)[source]
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:
- classmethod from_registry(registry, google_account_id=None, person=None, scopes=None)[source]
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:
- property credentials
- batch_update_presentation(presentation_id, requests)[source]
Execute a batch of Slides API requests.
- static spreadsheet_url(spreadsheet_id)[source]
Return the live Google Sheets URL for a spreadsheet ID.
- static presentation_url(presentation_id)[source]
Return the live Google Slides URL for a presentation ID.
- static file_url(file_id, mime_type=None)[source]
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.
- copy_file(file_id, title=None)[source]
Copy a Drive file (spreadsheet, doc, presentation) and return the new ID.
Share a Drive file with a user.
- upload_file(local_path, name=None, mime_type=None, folder_id=None)[source]
Upload a local file to Drive and return its file ID.
- Parameters:
- Returns:
The new Drive file’s ID.
- Return type:
Snowflake data warehouse connector for Siege Utilities.
Provides seamless integration with Snowflake for data warehousing and analytics.
- class siege_utilities.analytics.snowflake_connector.SnowflakeConnector[source]
Bases:
objectSnowflake data warehouse connector with advanced features.
- __init__(account=None, user=None, password=None, warehouse=None, database=None, schema=None, role=None, config_file=None)[source]
Initialize Snowflake connector.
- Parameters:
account (str | None) – Snowflake account identifier
user (str | None) – 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()[source]
Establish connection to Snowflake.
- Raises:
ConnectionError – If the connection attempt fails.
- Return type:
None
- execute_query(query, params=None)[source]
Execute a SQL query.
- Parameters:
- Returns:
Query results as list of tuples
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If query execution fails.
- Return type:
- execute_ddl(ddl_statement)[source]
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)[source]
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)[source]
Download data from Snowflake as pandas DataFrame.
- Parameters:
- 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)[source]
Get information about a Snowflake table.
- Parameters:
- Returns:
Dictionary with table information
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- list_tables(database=None, schema=None)[source]
List all tables in a database/schema.
- Parameters:
- Returns:
List of table names
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- siege_utilities.analytics.snowflake_connector.get_snowflake_connector(config_file=None)[source]
Get Snowflake connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.snowflake_connector.upload_to_snowflake(df, table_name, config_file=None, **kwargs)[source]
Convenience function to upload DataFrame to Snowflake.
- siege_utilities.analytics.snowflake_connector.download_from_snowflake(query, config_file=None, **kwargs)[source]
Convenience function to download DataFrame from Snowflake.
- siege_utilities.analytics.snowflake_connector.execute_snowflake_query(query, config_file=None, **kwargs)[source]
Convenience function to execute Snowflake query.
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[source]
Bases:
objectFacebook Business connection and data retrieval manager.
- __init__(access_token, app_id=None, app_secret=None)[source]
Initialize Facebook Business connector.
- get_ad_insights(ad_account_id, start_date, end_date, fields=None, breakdowns=None)[source]
Retrieve ad insights from Facebook Ads.
- Parameters:
- Returns:
Pandas DataFrame with ad insights
- Return type:
- get_page_insights(page_id, start_date, end_date, metrics=None)[source]
Retrieve page insights from Facebook Pages.
- get_business_insights(business_id, start_date, end_date, metrics=None)[source]
Retrieve business insights from Facebook Business Manager.
- save_as_pandas(df, output_path, format='parquet')[source]
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)[source]
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')[source]
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:
- siege_utilities.analytics.facebook_business.create_facebook_account_profile(client_id, fb_account_id, account_type='ad_account', access_token=None)[source]
Create a Facebook account profile linked to a client.
- siege_utilities.analytics.facebook_business.list_facebook_accounts_for_client(client_id, config_directory=None)[source]
List all Facebook accounts associated with a client.
- siege_utilities.analytics.facebook_business.load_facebook_account_profile(account_id, config_directory=None)[source]
Load Facebook account profile from JSON file.
- siege_utilities.analytics.facebook_business.save_facebook_account_profile(profile, config_directory=None)[source]
Save Facebook account profile to JSON file.
Vista Social (https://vistasocial.com) is a social-media management
platform that aggregates engagement, posts, and audience metrics across
Facebook, Instagram, LinkedIn, X, TikTok, YouTube, and other channels.
This module provides a thin connector that auths against the Vista
Social REST API and exposes a few common reporting calls in the same
shape as the other connectors in this package
(FacebookBusinessConnector, SnowflakeConnector).
Why “thin”
Vista Social’s REST API documentation requires an authenticated account; the public website does not enumerate endpoint URLs. This connector ships with:
The base
_requesthelper that handles auth headers, retries, and error normalization – fully tested.Concrete
account,profiles,posts,analyticsmethods that wrap likely endpoint paths (/v1/accounts,/v1/profiles,/v1/posts,/v1/analytics). These are informed by the publicly visible URL structure on the Vista Social web app and are documented as “best-known” rather than spec-confirmed.
When you obtain real Vista Social API documentation (request access at
api@vistasocial.com), verify the endpoints against the docs and adjust
this module. The _request plumbing is correct regardless of which
endpoints exist.
The intended consumer is siege_utilities.reporting, which will
embed the returned analytics into a combined GA4 + social-media PDF
report (the second half of #306). That work lives in a follow-up PR.
- exception siege_utilities.analytics.vista_social.VistaSocialError[source]
Bases:
RuntimeErrorBase class for all Vista Social connector failures.
- exception siege_utilities.analytics.vista_social.VistaSocialAuthError[source]
Bases:
VistaSocialErrorAuthentication failed (401/403). Token is missing, wrong, or expired.
- exception siege_utilities.analytics.vista_social.VistaSocialRateLimitError[source]
Bases:
VistaSocialErrorAPI rate limit hit (429). Caller can back off and retry later.
- class siege_utilities.analytics.vista_social.VistaSocialResponse[source]
Bases:
objectStructured response from a Vista Social API call.
The connector returns this rather than the raw
requests.Responseso consumers don’t need to know about the underlying HTTP library.- 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).
- class siege_utilities.analytics.vista_social.VistaSocialConnector[source]
Bases:
objectBearer-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:
VistaSocialAuthErroron 401/403,VistaSocialRateLimitErroron 429,VistaSocialErroron 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)[source]
- list_accounts()[source]
Return the list of accounts this token can access.
Wraps
GET /v1/accounts. Returns the decodeddatalist from the response, or an empty list if the API returned nothing. Raises on auth / rate-limit / network failures (no silent empty-on-error).
- list_profiles(account_id)[source]
Return social profiles connected to account_id.
Wraps
GET /v1/accounts/{account_id}/profiles.
- get_account_analytics(account_id, *, start_date, end_date, metrics=None)[source]
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:
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)[source]
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.