Sales_Toolkit / app.py
github-actions[bot]
sync: automatic content update from github
bc7640d
import os
import json
import pandas as pd
import streamlit as st
import snowflake.connector
from cryptography.hazmat.primitives import serialization
from comscore_site_list import render as render_comscore_site_list
from media_plan_templater import render as render_media_plan_templater
st.set_page_config(layout="wide", page_title="Sales Toolkit", page_icon="πŸ“Š")
# ─── Snowflake Helpers ───────────────────────────────────────
def get_conn():
raw_key = os.getenv("snowflake_private_key")
priv_key = serialization.load_pem_private_key(raw_key.encode(), password=None)
return snowflake.connector.connect(
user=os.getenv("snowflake_user"),
account=os.getenv("snowflake_account_identifier"),
private_key=priv_key,
role=os.getenv("snowflake_role"),
warehouse=os.getenv("snowflake_warehouse"),
database=os.getenv("snowflake_database"),
)
@st.cache_data
def get_lookup_table():
conn = get_conn()
cs = conn.cursor()
database = os.getenv("snowflake_database") # e.g. ANALYTICS
schema = "SIGMA_SCRATCH"
view = "VIEW_NEW_DATASET_FROM_SQL_3B3605DCE9F84DBC8F8508F33C4364FF"
cs.execute(
f"""
SELECT
"ENTITY" AS entity,
"SITE ID" AS site_id,
"SITE NAME" AS site_name,
"STATUS" AS status,
"AD OPTIONS" AS ad_options
FROM {database}.{schema}.{view}
"""
)
cols = [d[0].lower() for d in cs.description]
rows = cs.fetchall()
cs.close()
conn.close()
return pd.DataFrame(rows, columns=cols)
def extract_pmp(ad_opts):
"""Parse the JSON in `ad_options` and return the top-level 'pmp' boolean."""
try:
obj = json.loads(ad_opts)
return obj.get("pmp")
except Exception:
return None
# ─── App Layout ───────────────────────────────────────────────
st.title("Sales Toolkit")
tab1, tab2 = st.tabs(["πŸ“Š Comscore Site List", "Media Plan Templater"])
with tab1:
render_comscore_site_list(get_conn, get_lookup_table, extract_pmp)
with tab2:
render_media_plan_templater(get_conn)