"""
Bot Or Not (BON) - influencer due-diligence app.
Reads the dataset from the HF dataset repo and the winning embedding model from HF.
Pipeline: classifier score + retrieval (3 similar) + neighbor-vote diagnosis + bot-ring check + generated memo.
CPU friendly. No em-dashes or en-dashes anywhere in the output.
"""
import os, re, json, time, random
from collections import Counter
import numpy as np
import pandas as pd
import faiss
import joblib
import torch
import gradio as gr
from huggingface_hub import hf_hub_download
from sentence_transformers import SentenceTransformer
from transformers import AutoModelForCausalLM, AutoTokenizer
# ----------------------------------------------------------------------------
# Config
# ----------------------------------------------------------------------------
DATASET_REPO = "mayacheruty/BON-Synthetic-Influencers" # read dataset live from HF
DATASET_FILE = "bon_dataset.csv"
EMB_MODEL = "BAAI/bge-small-en-v1.5" # winning embedding model, loaded from HF
GEN_MODEL = os.environ.get("BON_GEN_MODEL", "Qwen/Qwen2.5-1.5B-Instruct") # better instruction following
EMB_FILE = "bon_embeddings.npy" # these four files sit in the Space repo
IDX_FILE = "bon_faiss.index"
DOC_FILE = "bon_docs.parquet"
CLF_FILE = "bon_classifier.pkl"
K_DETECT = 10 # neighbors used for the fraud vote
TIGHT = 0.98 # tight similarity threshold for coordinated-ring detection
RING_MIN = 15 # a query is "in a ring" if it has this many near-identical neighbors
# ----------------------------------------------------------------------------
# Load artifacts and models once at startup (never per request)
# ----------------------------------------------------------------------------
torch.set_num_threads(max(1, os.cpu_count() or 1)) # use all vCPUs on the CPU Space
print("Loading dataset, docs, classifier ...")
csv_path = hf_hub_download(repo_id=DATASET_REPO, filename=DATASET_FILE, repo_type="dataset")
raw = pd.read_csv(csv_path)
docs = pd.read_parquet(DOC_FILE) # profile_id, doc
FEATURES14 = ["follower_count", "following_count", "posts_count", "account_age_months",
"avg_likes", "avg_comments", "avg_views", "engagement_rate", "like_comment_ratio",
"follower_following_ratio", "view_follower_ratio", "growth_spike_score",
"audience_geo_mismatch", "audience_real_ratio"]
try:
bundle = joblib.load(CLF_FILE)
clf, CLF_FEATURES = bundle["model"], bundle["features"]
except Exception as e: # sklearn version drift, etc.
print("Classifier pickle unavailable (", e, "), retraining from the dataset.")
from sklearn.ensemble import RandomForestClassifier
CLF_FEATURES = FEATURES14
clf = RandomForestClassifier(n_estimators=200, random_state=42).fit(raw[CLF_FEATURES], raw["is_fraud"])
meta = docs.merge(raw, on="profile_id", how="left").reset_index(drop=True)
arch = meta["archetype"].values
fraud = meta["is_fraud"].values
print("Loading embedding model from HF ...")
embedder = SentenceTransformer(EMB_MODEL, device="cpu") # CPU Space, no accelerator available
# Embed the dataset with the SAME model used for queries, so both live in one vector space.
# (Loading a saved .npy from a different model/version is what caused zero-similarity matches.)
print("Embedding the dataset (one time at startup) ...")
emb = embedder.encode(meta["doc"].tolist(), batch_size=64, convert_to_numpy=True,
normalize_embeddings=True, device="cpu").astype("float32")
index = faiss.IndexFlatIP(emb.shape[1]); index.add(emb)
assert len(meta) == index.ntotal == emb.shape[0], "row counts do not match"
# 2D projection for the map (PCA transforms new queries too)
from sklearn.decomposition import PCA
print("Projecting embeddings to 2D (PCA) for the map ...")
_pca = PCA(n_components=2, random_state=42).fit(emb)
_c2 = _pca.transform(emb)
_cmin = _c2.min(axis=0); _cspan = (_c2.max(axis=0) - _cmin) + 1e-9
coords_norm = (_c2 - _cmin) / _cspan
def project(vec):
return (_pca.transform(vec) - _cmin) / _cspan
_rng = np.random.default_rng(0)
_bg = _rng.choice(len(coords_norm), size=min(1400, len(coords_norm)), replace=False)
BON_MAP = [[round(float(coords_norm[i, 0]), 4), round(float(coords_norm[i, 1]), 4), int(fraud[i])]
for i in _bg]
print("Loading generation model from HF ...")
tok = AutoTokenizer.from_pretrained(GEN_MODEL)
gen = AutoModelForCausalLM.from_pretrained(GEN_MODEL, torch_dtype=torch.float32,
low_cpu_mem_usage=True) # CPU, fixed RAM budget
gen.eval()
print("Ready.")
# ----------------------------------------------------------------------------
# Feature engineering (identical to the Recommendation notebook)
# ----------------------------------------------------------------------------
def bucket(r):
parts = []
er = r["engagement_rate"]
parts.append("almost no engagement" if er < 1 else "low engagement" if er < 3
else "moderate engagement" if er < 8 else "high engagement")
g = r["audience_geo_mismatch"]
if g > 0.5: parts.append("audience mostly in unrelated regions")
elif g > 0.3: parts.append("noticeable audience geo mismatch")
rr = r["audience_real_ratio"]
parts.append("mostly fake-looking followers" if rr < 0.5 else "mixed follower quality" if rr < 0.75
else "mostly real followers")
if r["follower_following_ratio"] > 80: parts.append("follows almost nobody despite a huge following")
if r["like_comment_ratio"] > 120: parts.append("many likes but very few comments")
if r["growth_spike_score"] > 0.6: parts.append("sudden follower spikes")
return ", ".join(parts)
def make_doc(r):
return (f"{r['niche']} influencer. {bucket(r)}. "
f"Bio: {r['bio']} Comments: {r['sample_comments']} Audience: {r['audience_note']}")
def red_flags(r):
f = []
if r["engagement_rate"] < 1: f.append("Engagement rate is near zero.")
elif r["engagement_rate"] < 3: f.append("Engagement rate is low for the follower tier.")
if r["audience_geo_mismatch"] > 0.5: f.append("Most of the audience sits in unrelated regions.")
elif r["audience_geo_mismatch"] > 0.3: f.append("Noticeable audience geography mismatch.")
if r["audience_real_ratio"] < 0.5: f.append("A large share of followers look fake.")
elif r["audience_real_ratio"] < 0.75: f.append("Follower quality is mixed.")
if r["follower_following_ratio"] > 80: f.append("Follows almost nobody despite a large following.")
if r["like_comment_ratio"] > 120: f.append("Many likes but very few comments.")
if r["growth_spike_score"] > 0.6: f.append("Sudden follower spikes in its history.")
return f
# ----------------------------------------------------------------------------
# Analysis: classifier + retrieval + ring check
# ----------------------------------------------------------------------------
def recommended_actions(verdict, fair_ceiling):
if verdict == "WALK AWAY":
return ["Do not proceed with a paid partnership.",
"If you still consider it, require an independent audience audit first.",
"Do not send product or advance payment."]
if verdict == "NEGOTIATE DOWN":
return [f"Pay only for genuine reach; a fair ceiling is about ${fair_ceiling:,} for one post.",
"Request an audience audit or analytics screenshots before you sign.",
"Start with one paid trial post, not a full campaign."]
return ["Reasonable to proceed at market rate.",
"Confirm deliverables, usage rights, and the posting schedule.",
"Track engagement on the sponsored post to verify performance."]
def value_lines(verdict, genuine_reach, fair_ceiling, followers):
# keep the ROI framing consistent with the recommendation the customer receives
if verdict == "WALK AWAY":
return [f"Genuine reach: only about {genuine_reach:,} real followers out of {followers:,} total.",
"Recommended spend: $0. The audience is largely inauthentic, so there is no fair price for a post."]
if verdict == "NEGOTIATE DOWN":
return [f"Genuine reach: about {genuine_reach:,} real followers; the rest looks inauthentic.",
f"Pay only for genuine reach. A fair ceiling is about ${fair_ceiling:,} for one post, below the likely asking price."]
return [f"Genuine reach: about {genuine_reach:,} real followers.",
f"Fair value for one sponsored post: about ${fair_ceiling:,}. Reasonable to pay around this rate."]
def analyze(row):
doc = make_doc(row)
q = embedder.encode([doc], convert_to_numpy=True, normalize_embeddings=True, device="cpu").astype("float32")
sims, nn = index.search(q, 61)
neigh, s = list(nn[0]), sims[0]
neigh_k = neigh[:K_DETECT]
fraud_types = [arch[n] for n in neigh_k if fraud[n] == 1]
likely = Counter(fraud_types).most_common(1)[0][0] if fraud_types else "none"
n_sim = len(neigh_k)
n_sim_fraud = int(sum(int(fraud[n]) for n in neigh_k)) # case-based evidence over the 10 nearest
density = int((s[:60] > TIGHT).sum()) # query is not in the index, so no self match
in_ring = density >= RING_MIN
feats = np.array([[float(row[f]) for f in CLF_FEATURES]])
clf_prob = float(clf.predict_proba(feats)[0, 1]) # numeric classifier
neighbor_rate = n_sim_fraud / max(1, n_sim) # case-based evidence from retrieval
prob = 0.5 * clf_prob + 0.5 * neighbor_rate # combine the two so the verdict is consistent
qv = q[0]
similar = []
for rank, n in enumerate(neigh[:3]):
cos = float(np.dot(qv, emb[n])) # true cosine (query and dataset both normalized)
similar.append({"archetype": str(arch[n]), "fraud": int(fraud[n]),
"sim": round(max(0.0, cos), 3), "niche": str(meta.iloc[n]["niche"]),
"x": round(float(coords_norm[n, 0]), 4), "y": round(float(coords_norm[n, 1]), 4)})
qxy = project(q)[0] # the query's real position in the same 2D space
authentic_share = max(0.0, 1.0 - prob) # same authenticity basis as the trust score and verdict
genuine_reach = int(row["follower_count"] * authentic_share)
fair_ceiling = int(round(genuine_reach / 1000.0 * 20)) # rough guide: about $20 per 1,000 genuine followers
trust = int(round(100 * (1 - prob)))
if in_ring or prob >= 0.6: verdict, risk = "WALK AWAY", "high"
elif prob >= 0.35: verdict, risk = "NEGOTIATE DOWN", "medium"
else: verdict, risk = "PAY", "low"
return {"trust": trust, "verdict": verdict, "risk": risk, "niche": row["niche"],
"platform": row.get("platform", "General"),
"fraud_probability": round(prob, 2), "likely_fraud_type": likely,
"in_ring": in_ring, "density": density,
"red_flags": red_flags(row), "similar": similar,
"n_similar": n_sim, "n_similar_fraud": n_sim_fraud,
"actions": recommended_actions(verdict, fair_ceiling),
"genuine_reach": genuine_reach, "fair_ceiling": fair_ceiling,
"value_lines": value_lines(verdict, genuine_reach, fair_ceiling, int(row["follower_count"])),
"qxy": [round(float(qxy[0]), 4), round(float(qxy[1]), 4)]}
# ----------------------------------------------------------------------------
# Generation (streamed), with the em-dash / en-dash cleaner from Phase 4
# ----------------------------------------------------------------------------
SYSTEM = ("You are BON, an influencer fraud analyst. Write a short memo to a BRAND advising whether to pay "
"an influencer. Address the brand, never the influencer. No greeting, no signature, no markdown, "
"no asterisks, no bold. Three to four short sentences. Use ONLY the facts given, including the exact "
"trust score, and do not invent numbers. Your recommendation MUST match the given verdict and must "
"not contradict it anywhere in the memo. End with the recommendation in capitals: PAY, NEGOTIATE "
"DOWN, or WALK AWAY. Do not use em-dashes or en-dashes.")
_CLAUSE_DASHES = ["—", "–", "―", "‒"]
_HYPHENISH = {"−": "-", "‑": "-"}
_QUOTES = {"“": '"', "”": '"', "‘": "'", "’": "'", "…": "..."}
def humanize(t):
for b, g in _QUOTES.items(): t = t.replace(b, g)
for b, g in _HYPHENISH.items(): t = t.replace(b, g)
for d in _CLAUSE_DASHES: t = re.sub(rf"\s*{d}\s*", ", ", t)
t = re.sub(r"\s+([,.;:!?])", r"\1", t)
t = re.sub(r",\s*,", ",", t)
t = re.sub(r"\s+", " ", t).strip()
return t
_LABELS = (r"(?:platform|verdict\s*to\s*state|verdict|risk\s*level|risk|similar\s*cases|closest\s*cases|"
r"trust\s*score|likely\s*(?:issue|fraud\s*type)|coordinated\s*bot\s*ring|red\s*flags|facts|memo|"
r"pay\s*/\s*walk[- ]?away|recommendation)")
def finalize_memo(memo, verdict):
memo = memo.replace("**", "").replace("*", "").replace("#", "").replace("`", "")
memo = re.sub(r"(?<=[A-Z])-(?=[A-Z])", "", memo) # NEGOTIA-TE -> NEGOTIATE
memo = re.sub(r"(?i)\bwalk\s*aways\b", "walk away", memo) # WALK AWAYS -> walk away
memo = re.sub(r"(?i)\bnegotiates\s*down\b", "negotiate down", memo)
memo = re.sub(rf"(?is)\b{_LABELS}\s*:\s*[^.]*\.?", " ", memo) # strip any echoed "Label: value." parts
memo = re.sub(r"\s+", " ", memo).strip().strip(".").strip()
memo = (memo + ".") if memo else "This profile needs a closer look before you commit."
return (memo + " Recommendation: " + verdict + ".").strip() # one clean, canonical recommendation
def build_prompt(a):
flags = "; ".join(a["red_flags"]) or "none"
ring = "yes" if a["in_ring"] else "no"
return ("Write a three sentence due-diligence note to a brand about whether to pay this influencer. "
"Write plain prose only. Do not use labels, lists, markdown, or headings, and do not repeat the "
"findings word for word. Base it on these findings: "
f"trust score {a['trust']} out of 100; risk {a['risk']}; "
f"{a.get('n_similar_fraud', 0)} of {a.get('n_similar', 10)} most similar past profiles are "
f"fraudulent; likely issue {a['likely_fraud_type']}; coordinated bot ring {ring}; "
f"red flags {flags}. In the three sentences, say plainly what the audience quality means for the "
f"brand's campaign spend. End with exactly: Recommendation: {a['verdict']}")
def generate_memo(a):
msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": build_prompt(a)}]
text = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
inp = tok(text, return_tensors="pt")
with torch.inference_mode():
out = gen.generate(**inp, max_new_tokens=150, do_sample=True, temperature=0.3,
top_p=0.9, repetition_penalty=1.2, no_repeat_ngram_size=3,
pad_token_id=tok.eos_token_id)
memo = tok.decode(out[0][inp["input_ids"].shape[1]:], skip_special_tokens=True)
return finalize_memo(humanize(memo), a["verdict"])
_PRETTY = {"authentic_organic": "Authentic, organic creator", "micro_real": "Genuine micro-creator",
"viral_legit": "Real viral creator", "lightly_boosted": "Lightly boosted",
"bought_followers": "Bought followers", "click_farm_audience": "Click-farm audience",
"bought_engagement": "Bought engagement (fake likes)", "engagement_pod": "Engagement pod",
"ai_persona": "AI-generated persona"}
def _ascii(s):
return str(s).encode("latin-1", "ignore").decode("latin-1")
def make_pdf(a, memo):
from fpdf import FPDF
from datetime import date
pdf = FPDF(format="A4"); pdf.set_auto_page_break(True, margin=20); pdf.add_page()
W, M, ink, mut = pdf.w, 18, (34, 44, 64), (120, 130, 150)
def L(txt, h=6, size=11, style="", rgb=ink):
pdf.set_x(M); pdf.set_font("Helvetica", style, size); pdf.set_text_color(*rgb)
pdf.multi_cell(W - 2 * M, h, _ascii(txt))
def section(t):
pdf.ln(3); pdf.set_x(M); pdf.set_font("Helvetica", "B", 12); pdf.set_text_color(20, 90, 150)
pdf.multi_cell(W - 2 * M, 7, _ascii(t))
pdf.set_draw_color(205, 218, 236); pdf.set_line_width(0.3); y = pdf.get_y(); pdf.line(M, y, W - M, y); pdf.ln(2)
# header band
pdf.set_fill_color(11, 21, 36); pdf.rect(0, 0, W, 26, style="F")
pdf.set_xy(M, 8); pdf.set_font("Helvetica", "B", 17); pdf.set_text_color(255, 255, 255); pdf.cell(20, 9, "BON")
pdf.set_xy(M + 21, 9); pdf.set_font("Helvetica", "", 12); pdf.set_text_color(150, 200, 240); pdf.cell(70, 7, "Investigator Report")
pdf.set_xy(W - M - 45, 10); pdf.set_font("Helvetica", "", 9); pdf.set_text_color(180, 200, 220)
pdf.cell(45, 6, _ascii(date.today().isoformat()), align="R")
# verdict badge
vc = {"WALK AWAY": (214, 74, 90), "NEGOTIATE DOWN": (210, 150, 50), "PAY": (45, 170, 120)}.get(a["verdict"], (90, 110, 140))
pdf.set_fill_color(*vc); pdf.rect(M, 34, W - 2 * M, 16, style="F")
pdf.set_xy(M + 5, 37); pdf.set_font("Helvetica", "B", 14); pdf.set_text_color(255, 255, 255); pdf.cell(110, 10, _ascii(f"Verdict: {a['verdict']}"))
pdf.set_xy(W - M - 60, 38); pdf.set_font("Helvetica", "B", 12); pdf.set_text_color(255, 255, 255); pdf.cell(55, 8, _ascii(f"Trust {a['trust']} / 100"), align="R")
pdf.set_y(57)
section("Analyst memo"); L(memo)
section("Recommended actions")
for x in a.get("actions", []): L(f"- {x}")
section("Comparable cases")
nn, nf = a.get("n_similar", 10), a.get("n_similar_fraud", 0)
L(f"Compared to 10,000 labeled cases, {nf} of the {nn} most similar are fraudulent. Closest examples:")
for c in a["similar"]:
lbl = "likely fake" if c["fraud"] else "genuine"
L(f"- {_PRETTY.get(c['archetype'], c['archetype'])} ({lbl}, {round(c['sim'] * 100)}% match)")
section("Estimated value")
for line in a.get("value_lines", []):
L(line)
section("Red flags")
for f in (a["red_flags"] or ["none"]): L(f"- {f}")
pdf.ln(4)
L("BON is a decision aid based on behavioral signals, not a verdict on a real person. Value figures are "
"illustrative estimates, not financial advice.", 5, 9, "I", mut)
path = "/tmp/BON_report.pdf"; pdf.output(path); return path
def fallback_memo(a):
lead = {"WALK AWAY": "This profile shows strong signs of an inauthentic audience.",
"NEGOTIATE DOWN": "This profile looks partly real but padded.",
"PAY": "This profile looks authentic."}.get(a["verdict"], "")
flags = " ".join(a["red_flags"]) if a["red_flags"] else "No strong red flags were found."
ring = " It also sits inside a coordinated bot ring." if a["in_ring"] else ""
return humanize(f"{lead} Trust score {a['trust']} out of 100, with a fraud probability of "
f"{a['fraud_probability']:.2f}. {flags}{ring} Recommendation: {a['verdict']}.")
# ----------------------------------------------------------------------------
# Gradio entry point: stream data JSON, then the memo
# ----------------------------------------------------------------------------
# Only the simple, public metrics a brand can actually get. Everything else is derived or optional.
INPUT_FIELDS = ["platform", "niche", "follower_count", "following_count", "posts_count",
"avg_likes", "avg_comments", "bio",
"account_age_months", "avg_views",
"audience_real_ratio", "audience_geo_mismatch", "growth_spike_score",
"sample_comments", "post_caption", "audience_note"]
def _num(x, d=0.0):
try:
f = float(x)
return max(0.0, f)
except (TypeError, ValueError):
return d
def build_row(v):
foll = _num(v["follower_count"]); following = _num(v["following_count"])
likes = _num(v["avg_likes"]); comments = _num(v["avg_comments"]); views = _num(v["avg_views"])
return {
"platform": v.get("platform") or "General",
"niche": v["niche"] or "lifestyle",
"follower_count": foll, "following_count": following,
"posts_count": _num(v["posts_count"]),
"account_age_months": _num(v["account_age_months"], 24.0), # neutral default if unknown
"avg_likes": likes, "avg_comments": comments, "avg_views": views,
# derived ratios (a brand never enters these directly)
"engagement_rate": (likes + comments) / foll * 100 if foll > 0 else 0.0,
"like_comment_ratio": likes / comments if comments > 0 else likes,
"follower_following_ratio": foll / following if following > 0 else foll,
"view_follower_ratio": views / foll if (views > 0 and foll > 0) else 1.0, # neutral if no video views
# optional audit signals (neutral defaults if left alone)
"growth_spike_score": min(1.0, _num(v["growth_spike_score"], 0.2)),
"audience_geo_mismatch": min(1.0, _num(v["audience_geo_mismatch"], 0.2)),
"audience_real_ratio": min(1.0, _num(v["audience_real_ratio"], 0.8)),
"bio": v["bio"] or "", "sample_comments": v["sample_comments"] or "",
"post_caption": v["post_caption"] or "", "audience_note": v["audience_note"] or "",
}
def investigate(*vals):
v = dict(zip(INPUT_FIELDS, vals))
a = analyze(build_row(v))
try:
memo = generate_memo(a)
except Exception as e:
print("generation failed, using fallback:", e)
memo = fallback_memo(a)
try:
pdf = make_pdf(a, memo)
except Exception as e:
print("pdf failed:", e); pdf = None
return json.dumps({**a, "done": True}), memo, pdf # data + memo + downloadable report
# ----------------------------------------------------------------------------
# Quick Starters (3 one-click examples)
# ----------------------------------------------------------------------------
PRESETS = {
"clickfarm": {
"platform": "Instagram",
"niche": "crypto", "follower_count": 220000, "following_count": 410, "posts_count": 60,
"avg_likes": 350, "avg_comments": 6, "account_age_months": 8, "avg_views": 120000,
"audience_real_ratio": 0.28, "audience_geo_mismatch": 0.8, "growth_spike_score": 0.82,
"bio": "Crypto signals daily. DM for VIP access. Guaranteed profits.",
"sample_comments": "amazing signals || to the moon || best group || just joined",
"post_caption": "New signal is live, do not miss it", "audience_note": "global crypto traders"},
"authentic": {
"platform": "Instagram",
"niche": "fitness", "follower_count": 18000, "following_count": 620, "posts_count": 340,
"avg_likes": 1600, "avg_comments": 140, "account_age_months": 41, "avg_views": 22000,
"audience_real_ratio": 0.93, "audience_geo_mismatch": 0.15, "growth_spike_score": 0.12,
"bio": "Coach sharing real training and honest progress.",
"sample_comments": "this helped me so much || tried it today || love your form tips",
"post_caption": "Week 6 of the program, here is what changed", "audience_note": "mostly local, engaged"},
"boosted": {
"platform": "Instagram",
"niche": "beauty", "follower_count": 75000, "following_count": 900, "posts_count": 210,
"avg_likes": 2100, "avg_comments": 60, "account_age_months": 22, "avg_views": 40000,
"audience_real_ratio": 0.68, "audience_geo_mismatch": 0.32, "growth_spike_score": 0.5,
"bio": "Beauty and skincare, sharing what works.",
"sample_comments": "obsessed || need this || where is it from || nice",
"post_caption": "My current everyday routine", "audience_note": "mixed regions"},
}
# text variations so each example run of a persona reads a little differently
VARIANTS = {
"clickfarm": {
"bio": ["Crypto signals daily. DM for VIP access. Guaranteed profits.",
"Daily crypto calls. Join the VIP group. Profits guaranteed.",
"Free crypto signals. DM to join our winning group today."],
"sample_comments": ["amazing signals || to the moon || best group || just joined",
"to the moon || huge gains || best signals || joined now",
"profit daily || legit group || just joined || lets go"]},
"authentic": {
"bio": ["Coach sharing real training and honest progress.",
"Strength coach. Real programs, honest results, no hype.",
"Helping people train smart. Real progress, no shortcuts."],
"sample_comments": ["this helped me so much || tried it today || love your form tips",
"so helpful || tried this today || your tips changed my training",
"great advice || started this week || form tips are gold"]},
"boosted": {
"bio": ["Beauty and skincare, sharing what works.",
"Skincare and beauty. Honest reviews of what actually works.",
"Beauty tips and product picks that hold up."],
"sample_comments": ["obsessed || need this || where is it from || nice",
"love this || need it || where to buy || so pretty",
"obsessed || adding to cart || where from || gorgeous"]},
}
# keep the authenticity signal inside each persona's band so the type never drifts
REAL_BANDS = {"clickfarm": (0.20, 0.38), "authentic": (0.88, 0.97), "boosted": (0.60, 0.75)}
def preset(name):
p = dict(PRESETS[name])
def jit(x, frac=0.15):
return max(0, round(x * (1 + random.uniform(-frac, frac))))
for k in ["follower_count", "following_count", "posts_count", "avg_likes",
"avg_comments", "avg_views", "account_age_months"]:
if isinstance(p.get(k), (int, float)):
p[k] = jit(p[k])
for k in ["audience_real_ratio", "audience_geo_mismatch", "growth_spike_score"]:
if k in p:
p[k] = round(min(1.0, max(0.0, p[k] + random.uniform(-0.05, 0.05))), 2)
lo, hi = REAL_BANDS.get(name, (0.0, 1.0)) # clamp the key authenticity signal to the band
p["audience_real_ratio"] = round(min(hi, max(lo, p["audience_real_ratio"])), 2)
for k, opts in VARIANTS.get(name, {}).items():
p[k] = random.choice(opts)
return [p[k] for k in INPUT_FIELDS]
# ----------------------------------------------------------------------------
# UI: CSS + JS injected once via head; animation is client side and data driven
# ----------------------------------------------------------------------------
_here = os.path.dirname(os.path.abspath(__file__))
def _read(name):
p = os.path.join(_here, name)
return open(p, encoding="utf-8").read() if os.path.exists(p) else ""
HEAD = (""
+ f"")
with gr.Blocks(title="Bot Or Not", head=HEAD, theme=gr.themes.Base()) as demo:
gr.HTML(""
"
"
"
Every influencer tells a story. BON helps you discover whether it's real.
"
"
BON, the AI Investigator
"
"
"
"
Start an Investigation
"
"
Investigate any creator across Instagram, TikTok, or YouTube
"
"
New here? Click a Quick Starter above to run a full '
'example, then edit the numbers. Enter only what you can see on the public profile. '
'Everything else is optional.
')
comps = {}
with gr.Row():
comps["platform"] = gr.Dropdown(["General", "Instagram", "TikTok", "YouTube"],
value="General", label="platform")
comps["niche"] = gr.Dropdown(["lifestyle", "crypto", "fitness", "beauty", "fashion", "tech",
"travel", "food", "gaming"], value="lifestyle", label="niche")
with gr.Row():
comps["follower_count"] = gr.Number(label="followers", value=0, minimum=0)
comps["following_count"] = gr.Number(label="accounts they follow", value=0, minimum=0)
with gr.Row():
comps["posts_count"] = gr.Number(label="total posts", value=0, minimum=0)
comps["avg_likes"] = gr.Number(label="avg likes per post (estimate)", value=0, minimum=0)
comps["avg_comments"] = gr.Number(label="avg comments per post (estimate)", value=0, minimum=0)
comps["bio"] = gr.Textbox(label="bio (optional)", lines=2, placeholder="Paste the profile bio")
# The Advanced panel was removed for reliability. These extra signals are kept as permanently
# hidden inputs so the model and the Quick Starters still receive the full feature set, but
# there is no collapsible panel to open, so nothing here can freeze or trap the page.
with gr.Column(visible=False):
comps["account_age_months"] = gr.Number(label="account age in months (if known)", value=0, minimum=0)
comps["avg_views"] = gr.Number(label="avg views per post (video, e.g. TikTok or Reels)", value=0, minimum=0)
comps["audience_real_ratio"] = gr.Slider(0, 1, value=0.8, step=0.01, label="share of real followers",
info="0 to 1. 0.9 means about 90 percent of followers look genuine.")
comps["audience_geo_mismatch"] = gr.Slider(0, 1, value=0.2, step=0.01, label="audience geo mismatch",
info="0 to 1. Higher means more followers in unrelated countries.")
comps["growth_spike_score"] = gr.Slider(0, 1, value=0.2, step=0.01, label="growth spike score",
info="0 to 1. Higher means sudden, unnatural jumps in followers.")
comps["sample_comments"] = gr.Textbox(label="sample comments (optional)", lines=2,
placeholder="great || love this || where can I buy")
comps["post_caption"] = gr.Textbox(label="post caption (optional)", lines=1,
placeholder="New drop is live")
comps["audience_note"] = gr.Textbox(label="audience note (optional)", lines=1,
placeholder="mostly US, ages 18 to 24")
run_btn = gr.Button("Investigate", variant="primary", elem_id="bon-investigate")
gr.HTML('') # animation stage renders here, below the form
report_dl = gr.DownloadButton("Export PDF report", elem_id="bon-dl")
data_box = gr.Textbox(visible=False, elem_id="bon-data")
memo_box = gr.Textbox(visible=False, elem_id="bon-memo")
ordered = [comps[k] for k in INPUT_FIELDS]
# start the animation on click, then run the streaming analysis
run_btn.click(None, None, None, js="() => window.BON && window.BON.begin()")
run_btn.click(investigate, ordered, [data_box, memo_box, report_dl])
# feed real data and streamed memo into the animation
data_box.change(None, data_box, None, js="(v) => window.BON && window.BON.onData(v)")
memo_box.change(None, memo_box, None, js="(v) => window.BON && window.BON.onMemo(v)")
# landing: "Analyze Manually" opens the existing workflow; "Instant Creator Scan" opens a coming-soon modal
manual_btn.click(lambda: gr.update(visible=True), None, manual_acc).then(
None, None, None,
js="""() => setTimeout(function(){
var e = document.getElementById('bon-form');
if (!e) return;
e.classList.remove('bon-reveal'); void e.offsetWidth; e.classList.add('bon-reveal');
e.scrollIntoView({behavior:'smooth', block:'start'});
}, 130)""")
scan_btn.click(None, None, None, js="() => window.bonShowScanModal && window.bonShowScanModal()")
# quick starters fill the form, then auto-run
for b, key in [(starter1, "clickfarm"), (starter2, "authentic"), (starter3, "boosted")]:
b.click(lambda k=key: preset(k), None, ordered).then(
None, None, None, js="() => document.getElementById('bon-investigate').click()")
if __name__ == "__main__":
demo.queue().launch()