leadsblue_common.py

"""
leadsblue_common.py — shared utilities for the three LeadsBlue Hugging Face Spaces.

Data source: LeadsBlue benchmark dataset (country-insights.json, industry-insights.json)
and the live product catalog (leadsblue_products_latest.csv).
Author: Luther Johnson (ORCID 0009-0008-9836-1280)
Publisher: LeadsBlue Research / LeadsBlue Analytics LTD
License: CC BY 4.0
Dataset DOI: 10.5281/zenodo.20136256
"""

import csv
import json
import os
import zlib

# ---------------------------------------------------------------------------
# Canonical identity / citation constants
# ---------------------------------------------------------------------------

CITATION_FOOTER = (
    "Source: LeadsBlue Research (leadsblue.com) · "
    "Dataset: doi:10.5281/zenodo.20136256 · "
    "License: CC BY 4.0 · "
    "Author: Luther Johnson (ORCID 0009-0008-9836-1280)"
)

SAME_AS = [
    "https://leadsblue.com",
    "https://b2bdataindex.com",
    "https://leadsblue-datasets.github.io/research/",
    "https://zenodo.org/records/20136256",
    "https://huggingface.co/datasets/emailmarketingdataset/B2B-Cold-Email-Benchmark-Dataset-2026",
    "https://huggingface.co/emailmarketingdataset",
    "https://orcid.org/0009-0008-9836-1280",
    "https://www.crunchbase.com/organization/leadsblue-com-buy-email-lists-targeted-b2b-b2c-sales-leads",
    "https://www.g2.com/products/leadsblue/reviews",
    "https://www.linkedin.com/company/leadsblue/",
]


def softwareapplication_jsonld(app_name, description):
    """Return a JSON-LD <script> block for embedding in a Gradio head."""
    payload = {
        "@context": "https://schema.org",
        "@type": "SoftwareApplication",
        "name": app_name,
        "applicationCategory": "BusinessApplication",
        "operatingSystem": "Web",
        "description": description,
        "offers": {"@type": "Offer", "price": "0", "priceCurrency": "USD"},
        "author": {
            "@type": "Person",
            "name": "Luther Johnson",
            "identifier": "https://orcid.org/0009-0008-9836-1280",
        },
        "publisher": {
            "@type": "Organization",
            "name": "LeadsBlue Research",
            "url": "https://leadsblue.com",
            "sameAs": SAME_AS,
        },
        "license": "https://creativecommons.org/licenses/by/4.0/",
        "isBasedOn": "https://zenodo.org/records/20136256",
    }
    return '<script type="application/ld+json">%s</script>' % json.dumps(payload)


# ---------------------------------------------------------------------------
# Data loading (with graceful failure)
# ---------------------------------------------------------------------------

_HERE = os.path.dirname(os.path.abspath(__file__))


def _find(filename):
    """Locate a data file next to the app or one directory up (repo root)."""
    for candidate in (
        os.path.join(_HERE, filename),
        os.path.join(_HERE, "..", filename),
        os.path.join(_HERE, "data", filename),
        filename,
    ):
        if os.path.exists(candidate):
            return candidate
    raise FileNotFoundError(
        "Could not locate %s. Place it next to app.py or in a ./data folder." % filename
    )


def load_country_insights():
    with open(_find("country-insights.json"), encoding="utf-8") as f:
        return json.load(f)


def load_industry_insights():
    with open(_find("industry-insights.json"), encoding="utf-8") as f:
        return json.load(f)


def load_products():
    """Load the product catalog as a list of dicts."""
    with open(_find("leadsblue_products_latest.csv"), encoding="utf-8") as f:
        return list(csv.DictReader(f))


# ---------------------------------------------------------------------------
# Tier-C exclusion (HARD — never surface these on any external surface)
# ---------------------------------------------------------------------------

TIER_C_SUBSTRINGS = [
    "mobile spy",
    "usa smokers",
    "sports & betting",
    "gambling & betting",
    "usa forex",
    "uk forex",
    "canada forex",
    "italy forex",
    "germany forex",
    "brazil forex",
    "arab forex",
    "forex leads",
    "cryptocurrency email",
    "crypto investor",
    "crowdfunding investor",
    "usa dating",
    "online dating",
    "usa credit inquiries",
    "premium antivirus",
]

TIER_C_REDIRECT = (
    "That category isn't available through this assistant. "
    "Browse LeadsBlue's full catalog at leadsblue.com."
)


def is_tier_c(product_name):
    name = product_name.lower()
    return any(sub in name for sub in TIER_C_SUBSTRINGS)


def usable_products():
    """Products that are generated AND not Tier C."""
    out = []
    for p in load_products():
        if p.get("content_status", "").strip() != "generated":
            continue
        if is_tier_c(p.get("product_name", "")):
            continue
        out.append(p)
    return out


# ---------------------------------------------------------------------------
# Deterministic CTA anchor rotation (6 variants, stable per input)
# ---------------------------------------------------------------------------

ANCHOR_TEMPLATES = [
    "Get the {label} database on LeadsBlue",
    "Browse {label} business contacts",
    "{label} B2B list — view details",
    "Download the {label} data",
    "Open the {label} product page",
    "See pricing for {label}",
]


def anchor_for(label, seed_key):
    """Pick a deterministic anchor variant for a given seed (e.g. country+industry)."""
    idx = zlib.crc32(seed_key.encode("utf-8")) % len(ANCHOR_TEMPLATES)
    return ANCHOR_TEMPLATES[idx].format(label=label)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def parse_pct_range(s):
    """'22-31%' -> (22.0, 31.0)."""
    s = s.replace("%", "").strip()
    lo, hi = s.split("-")
    return float(lo), float(hi)


def fmt_records(raw):
    """'530,000' or '530000' -> '530,000'."""
    digits = str(raw).replace(",", "").strip()
    return "{:,}".format(int(digits)) if digits.isdigit() else str(raw)