import os
from datetime import date, timedelta
from urllib.parse import urlparse
from profound import Profound
client = Profound(api_key=os.environ["PROFOUND_API_KEY"])
# What to fetch — replace with your own values.
CATEGORY_NAME = "<your-category-name>"
DAYS = 7
INCLUSIVE_END = date(2026, 5, 11) # the last day of your current window
def normalize(host_or_url):
"""Strip scheme, www, and trailing slash so domains compare cleanly."""
p = urlparse(host_or_url if "://" in host_or_url else f"https://{host_or_url}")
return (p.hostname or "").replace("www.", "").rstrip("/")
def get_owned_domains(category_id):
"""All domains across assets flagged is_owned in the category."""
out = set()
for a in client.organizations.categories.assets(category_id):
if not a.is_owned:
continue
if a.website:
out.add(normalize(a.website))
for d in a.alternate_domains or []:
out.add(normalize(d))
return out
def get_owned_citation_share(category_id, owned_domains, start, end):
"""Sum citation_share across all rows whose root_domain is one of yours."""
res = client.reports.citations(
category_id=category_id,
start_date=start.isoformat(),
end_date=end.isoformat(),
metrics=["count", "citation_share"],
dimensions=["root_domain"],
filters=[{"field": "prompt_type", "operator": "is", "value": "visibility"}],
)
m = res.info.query["metrics"]
d = res.info.query["dimensions"]
i_share = m.index("citation_share")
i_domain = d.index("root_domain")
return sum(
row.metrics[i_share]
for row in res.data
if normalize(row.dimensions[i_domain]) in owned_domains
)
def current_and_prior_windows(inclusive_end, days):
"""Two (start, end_exclusive) pairs of equal length, back-to-back."""
current = (
inclusive_end - timedelta(days=days - 1),
inclusive_end + timedelta(days=1), # +1 day → exclusive end
)
prior = (current[0] - timedelta(days=days), current[0])
return current, prior
# Helper — translate the category name to the UUID the report API expects.
def find_category_id(name):
"""Return the UUID of the category whose name matches (case-insensitive)."""
for c in client.organizations.categories.list():
if c.name.lower() == name.lower():
return c.id
raise ValueError(f"No category named {name!r}")
# Resolve name → ID, then run both windows.
category_id = find_category_id(CATEGORY_NAME)
owned = get_owned_domains(category_id)
current, prior = current_and_prior_windows(INCLUSIVE_END, DAYS)
current_share = get_owned_citation_share(category_id, owned, *current)
prior_share = get_owned_citation_share(category_id, owned, *prior)
delta_pp = (current_share - prior_share) * 100
print(f"Citation Share: {current_share:.1%} ({delta_pp:+.1f} pp vs prev period)")
