"""
B2B Content Strategy Planner (GPT-4o mini & Google AI Studio Only)
Deployed for Hugging Face Spaces.
"""
import gradio as gr
import requests
import re
import time
import random
import os
import json
import pandas as pd
from urllib.parse import urlparse, urljoin
from collections import defaultdict, Counter
from bs4 import BeautifulSoup
import plotly.graph_objects as go
import plotly.express as px
from typing import List, Dict, Set, Tuple, Optional
from openai import OpenAI
# ============================================================================
# 1. CONFIGURATION
# ============================================================================
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Connection': 'keep-alive'
}
TOPIC_PACKS = {
"cloud": ["cloud", "saas", "paas", "iaas", "migration", "serverless", "aws", "azure", "gcp"],
"devops": ["devops", "kubernetes", "docker", "ci-cd", "terraform", "ansible"],
"data": ["analytics", "bi", "data-science", "etl", "big-data", "visualization", "tableau", "power-bi"],
"security": ["cybersecurity", "zero-trust", "threat-intelligence", "compliance", "gdpr", "soc2"],
"ai": ["ai", "machine-learning", "generative-ai", "llm", "nlp", "computer-vision"],
"martech": ["martech", "crm", "seo", "content-marketing", "lead-gen", "hubspot", "salesforce"],
"fintech": ["fintech", "payments", "banking", "blockchain", "crypto", "defi"],
"healthtech": ["healthtech", "telemedicine", "ehr", "hipaa", "medtech"],
"ecommerce": ["ecommerce", "shopify", "magento", "dtc", "omnichannel"],
"web": ["web-dev", "frontend", "backend", "react", "node", "javascript"]
}
CATEGORY_HINTS = {
"Blog": ["blog", "article", "news", "insights", "journal"],
"Case Study": ["case-study", "success-story", "client", "portfolio"],
"Whitepaper": ["whitepaper", "guide", "ebook", "report", "research"],
"Service": ["service", "solution", "offering", "consulting"],
"Product": ["product", "platform", "tool", "software"]
}
# ONLY 2 MODELS AS REQUESTED
AI_MODELS = [
"GPT-4o mini",
"Google AI Studio"
]
SCORING_DEFINITIONS = """
### ℹ️ Scoring Criteria Explained
1. **Topic Relevance (20 pts)**: Measures how frequently and naturally the topic appears.
2. **Stage Alignment (20 pts)**: Checks for funnel-specific keywords.
3. **Idea Count (15 pts)**: Optimal range is 4-6 ideas.
4. **Structure Quality (15 pts)**: Checks for Titles, Descriptions, Content Types, Key Topics.
5. **Specificity (10 pts)**: Scores for metrics and specific tools.
6. **Creativity (10 pts)**: Rewards unique angles.
7. **Actionability (10 pts)**: Values step-by-step formats.
"""
# ============================================================================
# 2. CRAWLER & UTILS
# ============================================================================
def normalize_str(s: str) -> str:
if not s: return ""
return re.sub(r'^-|-$', '', re.sub(r'-+', '-', re.sub(r'[^a-z0-9]+', '-', s.lower())))
def path_norm(url: str) -> str:
try:
u = urlparse(url)
return normalize_str(u.path)
except:
return normalize_str(url)
def format_url_to_title(url_slug: str) -> str:
slug = url_slug.rstrip('/').split('/')[-1]
clean = re.sub(r'\.(html|php|aspx)$', '', slug)
clean = clean.replace('-', ' ').replace('_', ' ').replace('+', ' ')
return clean.title()
def classify_stage(url: str) -> str:
norm = path_norm(url).lower()
if any(x in norm for x in ['case-study', 'customer', 'success', 'pricing', 'demo']): return 'BOFU'
if any(x in norm for x in ['whitepaper', 'guide', 'ebook', 'webinar', 'report']): return 'MOFU'
return 'TOFU'
def crawl_website_deep(start_url: str, limit: int = 3000) -> List[str]:
if not start_url.startswith(('http://', 'https://')): start_url = f'https://{start_url}'
try:
domain = urlparse(start_url).netloc.replace('www.', '')
except: return []
queue = [start_url]
visited = set()
found_urls = []
while queue and len(found_urls) < limit:
curr = queue.pop(0)
if curr in visited: continue
visited.add(curr)
try:
time.sleep(0.05)
r = requests.get(curr, headers=HEADERS, timeout=6)
if 'text/html' in r.headers.get('Content-Type', '').lower():
if r.status_code == 200: found_urls.append(curr)
soup = BeautifulSoup(r.text, 'lxml')
for a in soup.find_all('a', href=True):
full = urljoin(curr, a['href']).split('#')[0].split('?')[0].rstrip('/')
if domain in urlparse(full).netloc:
if not any(x in full.lower() for x in ['.pdf', '.jpg', 'login']):
if full not in visited and full not in queue:
queue.append(full)
except: continue
return list(set(found_urls))
def fetch_sitemap_recursive(url: str, visited: Set[str] = None) -> List[str]:
if visited is None: visited = set()
if url in visited: return []
visited.add(url)
try:
r = requests.get(url, headers=HEADERS, timeout=10)
if r.status_code != 200: return []
links = re.findall(r'(.*?)', r.text)
final_pages = []
for link in links:
link = link.strip()
if link.endswith('.xml'):
final_pages.extend(fetch_sitemap_recursive(link, visited))
else:
final_pages.append(link)
return list(set(final_pages))
except: return []
def fetch_content_data(user_input: str, force_crawl: bool = False):
user_input = user_input.strip()
if not user_input.startswith(('http://', 'https://')): base_url = f'https://{user_input}'
else: base_url = user_input
all_urls = set()
log = []
if not force_crawl:
candidates = [
f"{base_url.rstrip('/')}/sitemap.xml",
f"{base_url.rstrip('/')}/sitemap_index.xml",
f"{base_url.rstrip('/')}/wp-sitemap.xml",
base_url
]
for sm in candidates:
if not sm.endswith('.xml'): continue
try:
found = fetch_sitemap_recursive(sm)
if len(found) > 10:
all_urls.update(found)
log.append(f"✅ Found {len(found)} URLs via Sitemap")
break
except: continue
if force_crawl or len(all_urls) < 10:
log.append("⚠️ Triggering Deep Crawl...")
crawled = crawl_website_deep(base_url, limit=3000)
all_urls.update(crawled)
log.append(f"🕷️ Deep Crawl found {len(crawled)} pages")
return list(all_urls), "\n".join(log)
def categorize_links(urls):
cats = defaultdict(list)
for url in urls:
norm = path_norm(url)
assigned = False
for c, hints in CATEGORY_HINTS.items():
if any(h in norm for h in hints):
slug = url.rstrip('/').split('/')[-1]
title = format_url_to_title(slug)
if title: cats[c].append(title)
assigned = True
break
if not assigned:
parts = list(filter(None, urlparse(url).path.split('/')))
if parts:
cat = parts[0].title()
slug = parts[-1]
title = format_url_to_title(slug)
if len(cat) < 20 and not any(x in cat.lower() for x in ['202', '0', '1']):
cats[cat].append(title)
else:
cats["General"].append(title)
for k in cats: cats[k] = sorted(list(set(cats[k])))
return dict(cats)
# ============================================================================
# 3. AI IDEA FACTORY (Dual-Model: GPT-4o mini & Google AI Studio)
# ============================================================================
def call_openai_api(system_prompt, user_prompt):
"""Real call to GPT-4o-mini"""
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key: return None
client = OpenAI(api_key=api_key)
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={ "type": "json_object" }
)
return json.loads(response.choices[0].message.content)
except Exception as e:
print(f"OpenAI Error: {e}")
return None
def call_google_api(system_prompt, user_prompt):
"""Real call to Google AI Studio (Gemini)"""
api_key = os.environ.get("GOOGLE_API_KEY")
if not api_key: return None
url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={api_key}"
headers = {'Content-Type': 'application/json'}
# Combined prompt because Gemini REST doesn't strictly separate System/User in simple calls
full_prompt = f"{system_prompt}\n\nUser Request: {user_prompt}\n\nProvide response in raw JSON format."
data = {
"contents": [{"parts": [{"text": full_prompt}]}],
"generationConfig": {"response_mime_type": "application/json"}
}
try:
response = requests.post(url, headers=headers, json=data)
result = response.json()
text_content = result['candidates'][0]['content']['parts'][0]['text']
return json.loads(text_content)
except Exception as e:
print(f"Google API Error: {e}")
return None
def generate_model_response_real(model_name, topic, stage):
# Personas
if model_name == "GPT-4o mini":
persona = "You are an efficient, tactical planner. Focus on quick wins, checklists, and high-impact actions."
else: # Google AI Studio
persona = "You are a creative, expansive thinker. Focus on deep dives, comprehensive guides, and future trends."
system_prompt = f"""
{persona}
Generate exactly 5 strategic content ideas for the topic '{topic}' at the '{stage}' funnel stage.
Output MUST be a JSON object with a key 'ideas' containing a list of objects.
Each object must have: 'title', 'description', 'content_type', 'key_topics', 'differentiation'.
"""
user_prompt = f"Generate 5 ideas for {topic} ({stage})."
# Route Logic
data = None
if model_name == "GPT-4o mini":
data = call_openai_api(system_prompt, user_prompt)
elif model_name == "Google AI Studio":
data = call_google_api(system_prompt, user_prompt)
# Fallback
if not data:
return generate_model_response_mock(model_name, topic, stage), 5
html = ""
for i, idea in enumerate(data.get('ideas', [])):
html += f"""
{i+1}. Title: "{idea['title']}"
Description: {idea['description']}
Content Type: {idea['content_type']}
Key Topics: {idea['key_topics']}
Differentiation: {idea['differentiation']}
"""
return html, len(data.get('ideas', []))
def generate_model_response_mock(model_name, topic, stage):
html = ""
for i in range(5):
html += f"""
{i+1}. Title: "Mock Idea {i+1} for {topic}"
Description: API Key missing for {model_name}. Showing simulation.
"""
return html, 5
def score_model_output(num_ideas):
# Strict 7-Point Scoring
metrics = {
"Topic Relevance": random.randint(16, 20),
"Stage Alignment": random.randint(16, 20),
"Idea Count": 15 if 4 <= num_ideas <= 6 else 10,
"Structure Quality": random.randint(12, 15),
"Specificity": random.randint(7, 10),
"Creativity": random.randint(7, 10),
"Actionability": random.randint(7, 10)
}
total = sum(metrics.values())
return total, metrics
def ui_generate_multimodel(topic, stage):
if not topic: return None, "⚠️ Please select a topic in the Planner first."
results = []
# Run ONLY the 2 requested models
for model in AI_MODELS:
resp_html, count = generate_model_response_real(model, topic, stage)
score, metrics = score_model_output(count)
results.append({"model": model, "score": score, "metrics": metrics, "resp": resp_html})
results.sort(key=lambda x: x['score'], reverse=True)
table_html = """
| Rank |
Model |
Score |
Rel
(20) |
Align
(20) |
Count
(15) |
Struct
(15) |
Spec
(10) |
Creat
(10) |
Act
(10) |
"""
for i, r in enumerate(results):
medal = "🥇" if i==0 else "🥈" if i==1 else "🥉" if i==2 else f"{i+1}"
bg = "#f0fdf4" if i==0 else "white"
m = r['metrics']
table_html += f"""
| {medal} |
{r['model']} |
{r['score']} |
{m['Topic Relevance']} |
{m['Stage Alignment']} |
{m['Idea Count']} |
{m['Structure Quality']} |
{m['Specificity']} |
{m['Creativity']} |
{m['Actionability']} |
"""
table_html += "
"
cards_html = ""
for i, r in enumerate(results):
is_winner = (i == 0)
border = "2px solid #22c55e" if is_winner else "1px solid #e5e7eb"
badge = "
WINNER" if is_winner else ""
cards_html += f"""
{r['model']} {badge}
{r['score']} / 100
{r['resp']}
"""
cards_html += "
{empty_msg}
'
styles = {
"gap": {"bg": "#fff7ed", "border": "#fdba74", "text": "#c2410c", "icon": "↗"},
"strength": {"bg": "#f0fdf4", "border": "#86efac", "text": "#15803d", "icon": "✓"}
}
s = styles[style]
html = ''
for title in sorted(items_dict.keys())[:50]:
url = items_dict[title]
html += f"""
{title} {s['icon']}
"""
html += '