""" Bulk Domain Filtration Engine FastAPI + SQLite + Hybrid (Gemini API OR Local Algorithmic Logic) """ import os import re import json import sqlite3 import asyncio from pathlib import Path from typing import Optional, List, Dict, Any import httpx from dotenv import load_dotenv from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field load_dotenv() BASE_DIR = Path(__file__).parent DB_PATH = BASE_DIR / "domains.db" STATIC_DIR = BASE_DIR / "static" app = FastAPI(title="Bulk Domain Filtration Engine") # --------------------------------------------------------------------------- # Database # --------------------------------------------------------------------------- def db_conn(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): with db_conn() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS domains ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch INTEGER NOT NULL, domain TEXT NOT NULL, score INTEGER NOT NULL, accepted INTEGER NOT NULL, category TEXT, why TEXT, est_value TEXT, sale_probability TEXT, engine TEXT NOT NULL, raw_json TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() init_db() # --------------------------------------------------------------------------- # Keyword libraries for local logic # --------------------------------------------------------------------------- TECH_KEYWORDS = { "ai","ml","gpt","bot","cloud","data","cyber","quantum","crypto","block", "chain","meta","nano","bio","gene","health","med","care","fit","green", "eco","solar","auto","drive","robot","saas","app","dev","code","tech", "net","web","dig","smart","neural","brain","mind","vision","synth", "labs","logic","core","hub","stack","io","forge","mint","flow","link" } COMMERCIAL_KEYWORDS = { "pay","buy","shop","cart","market","trade","cash","bank","fund","invest", "capital","credit","loan","insure","broker","deal","sale","price","gold", "coin","wallet","finance","fin","biz","pro","premium","luxe","elite", "vault","asset","yield","earn","profit","stock","fx","forex","estate" } TREND_KEYWORDS = TECH_KEYWORDS | { "auto","drive","electric","volt","spark","future","next","gen","neo", "aero","space","orbit","fusion","zero","carbon","clean","organic" } BAD_PATTERNS = re.compile(r"[^a-z0-9\-]") TRIPLE_LETTER = re.compile(r"(.)\1\1") BAD_CLUSTERS = re.compile(r"[bcdfghjklmnpqrstvwxz]{4,}", re.I) VOWELS = set("aeiouy") def count_syllables(word: str) -> int: word = word.lower() if not word: return 0 count, prev = 0, False for ch in word: is_v = ch in VOWELS if is_v and not prev: count += 1 prev = is_v if word.endswith("e") and count > 1: count -= 1 return max(1, count) def split_two_words(sld: str) -> Optional[List[str]]: """Naive 2-word segmentation by finding a vowel/consonant break.""" common = ["baby","born","premier","wood","tech","hub","labs","cloud","data", "pay","shop","market","gold","smart","green","auto","health", "care","cyber","quantum","forge","stack","core","flow","link", "mint","logic","vision","neural","brain","bio","gene","meta"] for w in common: if sld.startswith(w) and len(sld) > len(w) + 2: return [w, sld[len(w):]] if sld.endswith(w) and len(sld) > len(w) + 2: return [sld[:-len(w)], w] return None # --------------------------------------------------------------------------- # Local Algorithmic Engine # --------------------------------------------------------------------------- def local_evaluate(domain: str) -> Dict[str, Any]: d = domain.strip().lower() if "." not in d: d += ".com" sld, _, tld = d.rpartition(".") scores: Dict[str, int] = {} reasons: List[str] = [] # 7. Zero red flags — hard reject red_flag = False if tld != "com" or "-" in sld or re.search(r"\d", sld) or BAD_PATTERNS.search(sld): red_flag = True scores["red_flags"] = 0 else: scores["red_flags"] = 10 L = len(sld) # 1. Ultra-short if L <= 3: scores["ultra_short"] = 10 elif L <= 5: scores["ultra_short"] = 10 elif L <= 8: scores["ultra_short"] = 7 else: scores["ultra_short"] = 3 # 2. Radio test radio = 10 if TRIPLE_LETTER.search(sld): radio -= 5 if BAD_CLUSTERS.search(sld): radio -= 5 if re.search(r"(ph|zz|xx|qk|xz)", sld): radio -= 2 scores["radio_test"] = max(0, radio) # 3. Commercial liquidity scores["commercial"] = 10 if any(k in sld for k in COMMERCIAL_KEYWORDS) else (7 if L <= 6 else 4) # 4. Two-word synergy parts = split_two_words(sld) scores["two_word"] = 10 if parts else (6 if L <= 8 else 3) # 5. Dopamine / syllables syl = count_syllables(sld) scores["dopamine"] = 10 if 2 <= syl <= 3 else (6 if syl == 1 or syl == 4 else 3) # 6. Future trend scores["trend"] = 10 if any(k in sld for k in TREND_KEYWORDS) else (5 if L <= 6 else 3) # 8. Trust / professionalism (CVCV-ish, or english-looking) cvcv = bool(re.fullmatch(r"([bcdfghjklmnpqrstvwxz][aeiou]){2,3}", sld)) scores["trust"] = 10 if cvcv else (8 if L <= 7 and syl <= 3 else 5) # 9. Pain-killer / SEO value scores["painkiller"] = 10 if any(k in sld for k in COMMERCIAL_KEYWORDS | TECH_KEYWORDS) else 5 # 10. Multi-purpose resilience generic = not any(k in sld for k in {"med","law","food","pet","car"}) scores["resilience"] = 10 if generic and L <= 8 else 6 if red_flag: # zero the trust-related ones to guarantee rejection for k in ("commercial","two_word","trust","painkiller","resilience","trend","dopamine","radio_test","ultra_short"): scores[k] = min(scores[k], 3) total = sum(scores.values()) # Category if parts and not red_flag: category = "Two-Word" elif any(k in sld for k in TECH_KEYWORDS | COMMERCIAL_KEYWORDS): category = "Brandable Keyword" else: category = "Dictionary" # Why why_bits = [] if scores["ultra_short"] >= 8: why_bits.append(f"tight {L}-letter length") if scores["trend"] >= 8: why_bits.append("aligns with next-gen tech trends") if scores["commercial"] >= 8: why_bits.append("high commercial liquidity keywords") if scores["radio_test"] >= 8: why_bits.append("passes the radio test") if not why_bits: why_bits.append("balanced linguistic profile with room to grow") why = "Strong " + ", ".join(why_bits) + "." # Value estimate if total >= 92: value, prob = "$8,500", "High" elif total >= 88: value, prob = "$4,500", "High" elif total >= 85: value, prob = "$2,500", "Medium" elif total >= 70: value, prob = "$900", "Medium" else: value, prob = "$150", "Low" return { "domain": d, "score": total, "accepted": total >= 85 and not red_flag, "category": category, "why": why, "est_value": value, "sale_probability": prob, "engine": "local", "breakdown": scores, } # --------------------------------------------------------------------------- # Gemini Engine # --------------------------------------------------------------------------- GEMINI_URL = ( "https://generativelanguage.googleapis.com/v1beta/models/" "gemini-1.5-flash:generateContent" ) GEMINI_PROMPT = """You are an elite domain broker. Evaluate the domain "{domain}" against these 10 commandments (10 points each, total 100): 1. Ultra-Short (4-8 chars ideal) 2. Radio Test (easy to spell on hearing) 3. Commercial Liquidity (premium market demand) 4. Two-Word Synergy (fuses like a corporate brand) 5. Dopamine / Catchiness (2-3 syllables) 6. Future Trend (AI, SaaS, GreenTech, Bio, Health) 7. Zero Red Flags (INSTANT ZERO if hyphens, digits, or non-.com) 8. Instant Trust (CVCV or real english words) 9. Pain-Killer Value (saves PPC spend) 10. Multi-Purpose Resilience (brand can pivot) Return STRICT JSON only, no markdown: {{"score": , "accepted": =85 AND no red flags>, "category": "Dictionary|Two-Word|Brandable Keyword", "why": "<1-2 sentence market angle>", "est_value": "$X,XXX", "sale_probability": "High|Medium|Low"}}""" async def gemini_evaluate(domain: str, api_key: str, client: httpx.AsyncClient) -> Dict[str, Any]: payload = { "contents": [{"parts": [{"text": GEMINI_PROMPT.format(domain=domain)}]}], "generationConfig": {"temperature": 0.3, "responseMimeType": "application/json"}, } try: r = await client.post( f"{GEMINI_URL}?key={api_key}", json=payload, timeout=30.0, ) r.raise_for_status() data = r.json() text = data["candidates"][0]["content"]["parts"][0]["text"] parsed = json.loads(text) return { "domain": domain, "score": int(parsed.get("score", 0)), "accepted": bool(parsed.get("accepted", False)), "category": parsed.get("category", "Brandable Keyword"), "why": parsed.get("why", ""), "est_value": parsed.get("est_value", "$0"), "sale_probability": parsed.get("sale_probability", "Low"), "engine": "gemini", } except Exception as e: # Fallback silently to local result = local_evaluate(domain) result["engine"] = f"local (gemini-fallback: {type(e).__name__})" return result # --------------------------------------------------------------------------- # Pre-filter: red flags before spending API # --------------------------------------------------------------------------- def has_red_flags(domain: str) -> bool: d = domain.strip().lower() if "." not in d: d += ".com" sld, _, tld = d.rpartition(".") return tld != "com" or "-" in sld or bool(re.search(r"\d", sld)) or bool(BAD_PATTERNS.search(sld)) # --------------------------------------------------------------------------- # API # --------------------------------------------------------------------------- class FilterRequest(BaseModel): batch: int = Field(..., ge=1) api_key: Optional[str] = "" domains: str class DomainRecord(BaseModel): domain: str score: int accepted: bool category: str why: str est_value: str sale_probability: str engine: str @app.post("/api/filter") async def filter_domains(req: FilterRequest): raw = [d.strip().lower() for d in re.split(r"[,\n\r]+", req.domains) if d.strip()] # dedupe, keep order seen, domains = set(), [] for d in raw: if d not in seen: seen.add(d); domains.append(d) if not domains: raise HTTPException(400, "No domains provided") api_key = (req.api_key or "").strip() or os.getenv("GEMINI_API_KEY", "").strip() results: List[Dict[str, Any]] = [] if api_key: async with httpx.AsyncClient() as client: tasks = [] for d in domains: if has_red_flags(d): # Skip API call, use local tasks.append(asyncio.to_thread(local_evaluate, d)) else: tasks.append(gemini_evaluate(d, api_key, client)) results = await asyncio.gather(*tasks) else: results = [local_evaluate(d) for d in domains] # persist with db_conn() as conn: for r in results: conn.execute( """INSERT INTO domains (batch, domain, score, accepted, category, why, est_value, sale_probability, engine, raw_json) VALUES (?,?,?,?,?,?,?,?,?,?)""", (req.batch, r["domain"], r["score"], int(r["accepted"]), r["category"], r["why"], r["est_value"], r["sale_probability"], r["engine"], json.dumps(r)), ) conn.commit() accepted = [r for r in results if r["accepted"] and r["score"] >= 85] accepted.sort(key=lambda x: -x["score"]) accepted = accepted[:10] rejected = [r for r in results if r not in accepted] # Build the exact plain-text block lines: List[str] = [] lines.append(f"### BATCH {req.batch} CRITICAL FILTRATION REPORT") lines.append("") lines.append("**🔴 REJECTED DOMAINS (Trash/No Real-World Market Value):**") lines.append(", ".join(r["domain"] for r in rejected) if rejected else "None") lines.append("") lines.append("**🟢 ACCEPTED ULTRA-PREMIUM DOMAINS (Strict Max 5-10 Domains, Score 85+/100):**") if not accepted: lines.append("ZERO DOMAINS QUALIFIED") else: for i, r in enumerate(accepted, 1): lines.append(f"{i}. **{r['domain']}** - Score: {r['score']}/100") lines.append(f" - **Category:** {r['category']}") lines.append(f" - **Why it Sells (Future-Proof Angle):** {r['why']}") lines.append(f" - **Value & Liquidity:** Estimated Value: {r['est_value']} | Sale Probability: {r['sale_probability']}") lines.append("") lines.append("=" * 50) lines.append("🟢 COPY-PASTE READY RAW DOMAINS (PLAIN TEXT ONLY)") lines.append("=" * 50) lines.append(",".join(r["domain"] for r in accepted)) lines.append("") lines.append("-" * 50) lines.append(f"BATCH {req.batch} COMPLETE. Please provide the next batch or type 'FINAL REPORT'.") report = "\n".join(lines) return { "report": report, "accepted": accepted, "rejected": rejected, "engine_used": "gemini" if api_key else "local", "batch": req.batch, } @app.get("/api/history") def history(batch: Optional[int] = None, limit: int = 200): q = "SELECT * FROM domains" args: tuple = () if batch is not None: q += " WHERE batch = ?" args = (batch,) q += " ORDER BY id DESC LIMIT ?" args = (*args, limit) with db_conn() as conn: rows = [dict(r) for r in conn.execute(q, args).fetchall()] return rows @app.get("/api/batches") def batches(): with db_conn() as conn: rows = conn.execute( "SELECT batch, COUNT(*) AS total, SUM(accepted) AS accepted FROM domains GROUP BY batch ORDER BY batch DESC" ).fetchall() return [dict(r) for r in rows] # --------------------------------------------------------------------------- # Static UI # --------------------------------------------------------------------------- app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @app.get("/") def index(): return FileResponse(STATIC_DIR / "index.html") if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)