import json
import logging
import math
from datetime import datetime
from typing import Optional
from app.services.llm import call_llm_routed, call_llm_text_routed
logger = logging.getLogger(__name__)
EBAY_CATEGORIES = {
"startup_business": {
"id": 11759,
"name": "Businesses & Websites for Sale",
"subcategories": {
"internet_business": 11762,
"established_business": 11760,
"franchise": 11761,
},
},
"digital_products": {
"id": 31580,
"name": "Everything Else > Information Products",
"subcategories": {
"business_guides": 102479,
"how_to_guides": 102480,
"other": 31581,
},
},
"websites": {
"id": 67504,
"name": "Websites & Blogs",
"subcategories": {
"turnkey": 67505,
"custom_built": 67506,
"templates": 67507,
},
},
"domains": {
"id": 45058,
"name": "Domain Names",
"subcategories": {
"dot_com": 45059,
"other_tlds": 45060,
},
},
}
PRICING_TIERS = {
"starter": {
"label": "Starter Package",
"price_range": (27, 49),
"target_margin": 0.70,
"includes": [
"domain_analysis",
"brand_identity",
"basic_site_copy",
],
"ebay_position": "entry_level",
"description": "Domain + brand identity + basic site framework. Quick flip for buyers who want a head start.",
},
"professional": {
"label": "Professional Package",
"price_range": (97, 197),
"target_margin": 0.80,
"includes": [
"domain_analysis",
"brand_identity",
"full_site_copy",
"hero_images",
"sales_letter",
"theme_customization",
],
"ebay_position": "mid_market",
"description": "Complete website-ready package with professional copy, imagery, and brand kit.",
},
"premium": {
"label": "Premium Business-in-a-Box",
"price_range": (297, 497),
"target_margin": 0.85,
"includes": [
"domain_analysis",
"brand_identity",
"full_site_copy",
"hero_images",
"sales_letter",
"theme_customization",
"force_multiplier_docs",
"graphics_pack",
"deployment_ready",
],
"ebay_position": "premium",
"description": "Full business-in-a-box: website, 33 business documents, graphics pack, deployment-ready. Turnkey operation.",
},
"legendary": {
"label": "Legendary Enterprise Package",
"price_range": (997, 2497),
"target_margin": 0.90,
"includes": [
"domain_analysis",
"brand_identity",
"full_site_copy",
"hero_images",
"sales_letter",
"theme_customization",
"force_multiplier_docs",
"graphics_pack",
"deployment_ready",
"market_research",
"competitor_analysis",
"seo_strategy",
"ad_copy_suite",
"email_sequences",
],
"ebay_position": "elite",
"description": "Enterprise-grade business package with market intelligence, competitor analysis, SEO, ad copy, and email sequences. The ultimate turnkey business.",
},
}
PACKAGE_COMPONENTS = {
"domain_analysis": {"weight": 5, "label": "Domain Analysis & Valuation"},
"brand_identity": {"weight": 10, "label": "Brand Identity (Name, Tagline, Colors, Tone)"},
"basic_site_copy": {"weight": 8, "label": "Basic Site Copy (3-5 sections)"},
"full_site_copy": {"weight": 15, "label": "Full Site Copy (16 section types)"},
"hero_images": {"weight": 10, "label": "AI-Generated Hero Images"},
"sales_letter": {"weight": 12, "label": "Professional Sales Letter"},
"theme_customization": {"weight": 8, "label": "Visual Theme & Bling Engine"},
"force_multiplier_docs": {"weight": 18, "label": "33 Business Documents (6 Tiers)"},
"graphics_pack": {"weight": 10, "label": "AI Graphics Pack"},
"deployment_ready": {"weight": 4, "label": "FTP Deployment Ready"},
"market_research": {"weight": 8, "label": "Live Market Intelligence Report"},
"competitor_analysis": {"weight": 6, "label": "Competitor Landscape Analysis"},
"seo_strategy": {"weight": 5, "label": "SEO Strategy & Keywords"},
"ad_copy_suite": {"weight": 5, "label": "Ad Copy Suite (Social, PPC)"},
"email_sequences": {"weight": 4, "label": "Email Marketing Sequences"},
}
def score_package_completeness(package_data: dict) -> dict:
present = []
missing = []
checks = {
"domain_analysis": bool(package_data.get("domain") and package_data.get("niche")),
"brand_identity": bool(
package_data.get("brand_name")
and package_data.get("tagline")
and package_data.get("brand_colors")
),
"basic_site_copy": bool(
package_data.get("site_copy") and len(package_data.get("site_copy", {})) >= 3
),
"full_site_copy": bool(
package_data.get("site_copy") and len(package_data.get("site_copy", {})) >= 10
),
"hero_images": bool(package_data.get("hero_image_url") or package_data.get("hero_images")),
"sales_letter": bool(package_data.get("sales_letter") and len(str(package_data.get("sales_letter", ""))) > 200),
"theme_customization": bool(package_data.get("theme") or package_data.get("theme_config")),
"force_multiplier_docs": bool(
package_data.get("business_docs") and len(package_data.get("business_docs", [])) >= 10
),
"graphics_pack": bool(package_data.get("graphics_pack") and len(package_data.get("graphics_pack", [])) >= 1),
"deployment_ready": bool(package_data.get("deployed") or package_data.get("ftp_config")),
"market_research": bool(package_data.get("market_research")),
"competitor_analysis": bool(package_data.get("competitor_analysis")),
"seo_strategy": bool(package_data.get("seo_strategy")),
"ad_copy_suite": bool(package_data.get("ad_copy")),
"email_sequences": bool(package_data.get("email_sequences")),
}
total_weight = 0
earned_weight = 0
for comp_id, is_present in checks.items():
comp_meta = PACKAGE_COMPONENTS.get(comp_id, {"weight": 1, "label": comp_id})
total_weight += comp_meta["weight"]
if is_present:
earned_weight += comp_meta["weight"]
present.append({"id": comp_id, "label": comp_meta["label"], "weight": comp_meta["weight"]})
else:
missing.append({"id": comp_id, "label": comp_meta["label"], "weight": comp_meta["weight"]})
score = round((earned_weight / total_weight) * 100, 1) if total_weight > 0 else 0
recommended_tier = "starter"
for tier_id in ["legendary", "premium", "professional", "starter"]:
tier = PRICING_TIERS[tier_id]
tier_components = set(tier["includes"])
present_ids = {p["id"] for p in present}
if tier_components.issubset(present_ids):
recommended_tier = tier_id
break
return {
"score": score,
"earned_weight": earned_weight,
"total_weight": total_weight,
"present": present,
"missing": missing,
"recommended_tier": recommended_tier,
"recommended_price_range": PRICING_TIERS[recommended_tier]["price_range"],
"upgrade_path": _build_upgrade_path(present, missing),
}
def _build_upgrade_path(present: list, missing: list) -> list:
present_ids = {p["id"] for p in present}
path = []
for tier_id, tier in PRICING_TIERS.items():
tier_components = set(tier["includes"])
still_needed = tier_components - present_ids
if still_needed:
path.append({
"tier": tier_id,
"label": tier["label"],
"price_range": tier["price_range"],
"components_needed": [
{"id": c, "label": PACKAGE_COMPONENTS.get(c, {}).get("label", c)}
for c in still_needed
],
"components_count": len(still_needed),
})
return path
def get_ebay_category(niche: str, package_tier: str = "professional") -> dict:
niche_lower = niche.lower() if niche else ""
if any(kw in niche_lower for kw in ["website", "web", "blog", "online"]):
cat = EBAY_CATEGORIES["websites"]
subcat_key = "turnkey" if package_tier in ("premium", "legendary") else "custom_built"
elif any(kw in niche_lower for kw in ["domain", "url", "tld"]):
cat = EBAY_CATEGORIES["domains"]
subcat_key = "dot_com"
elif any(kw in niche_lower for kw in ["ebook", "course", "guide", "template", "digital"]):
cat = EBAY_CATEGORIES["digital_products"]
subcat_key = "business_guides"
else:
cat = EBAY_CATEGORIES["startup_business"]
subcat_key = "internet_business"
return {
"primary_category_id": cat["id"],
"primary_category_name": cat["name"],
"secondary_category_id": cat["subcategories"].get(subcat_key),
"suggested_ebay_title_prefix": _ebay_title_prefix(niche, package_tier),
"listing_type_recommendation": "Buy It Now" if package_tier in ("premium", "legendary") else "Auction + Buy It Now",
}
def _ebay_title_prefix(niche: str, tier: str) -> str:
tier_labels = {
"starter": "Starter",
"professional": "Professional",
"premium": "Premium Turnkey",
"legendary": "LEGENDARY Enterprise",
}
tier_label = tier_labels.get(tier, "Professional")
niche_short = niche[:40] if niche else "Online"
return f"{tier_label} {niche_short} Business-in-a-Box"
def calculate_listing_price(package_data: dict, tier: str = None) -> dict:
if not tier:
completeness = score_package_completeness(package_data)
tier = completeness["recommended_tier"]
tier_config = PRICING_TIERS.get(tier, PRICING_TIERS["professional"])
price_low, price_high = tier_config["price_range"]
niche_multiplier = 1.0
niche = (package_data.get("niche") or "").lower()
high_value_niches = ["saas", "software", "finance", "health", "legal", "real estate", "crypto", "ai"]
if any(kw in niche for kw in high_value_niches):
niche_multiplier = 1.3
completeness = score_package_completeness(package_data)
completeness_multiplier = 0.7 + (completeness["score"] / 100) * 0.3
adjusted_low = round(price_low * niche_multiplier * completeness_multiplier)
adjusted_high = round(price_high * niche_multiplier * completeness_multiplier)
suggested = round((adjusted_low + adjusted_high) / 2)
margin = tier_config["target_margin"]
estimated_cost = round(suggested * (1 - margin))
return {
"tier": tier,
"tier_label": tier_config["label"],
"base_range": (price_low, price_high),
"adjusted_range": (adjusted_low, adjusted_high),
"suggested_price": suggested,
"niche_multiplier": niche_multiplier,
"completeness_multiplier": round(completeness_multiplier, 2),
"completeness_score": completeness["score"],
"estimated_cost": estimated_cost,
"estimated_profit": suggested - estimated_cost,
"target_margin": margin,
}
async def generate_listing_copy(domain: str, niche: str, package_data: dict, tier: str = "professional") -> dict:
tier_config = PRICING_TIERS.get(tier, PRICING_TIERS["professional"])
pricing = calculate_listing_price(package_data, tier)
completeness = score_package_completeness(package_data)
category = get_ebay_category(niche, tier)
brand_name = package_data.get("brand_name", domain)
tagline = package_data.get("tagline", "")
present_components = [p["label"] for p in completeness["present"]]
prompt = f"""Generate a compelling eBay marketplace listing for a turnkey business-in-a-box package.
BUSINESS DETAILS:
- Domain: {domain}
- Niche: {niche}
- Brand Name: {brand_name}
- Tagline: {tagline}
- Package Tier: {tier_config['label']}
- Suggested Price: ${pricing['suggested_price']}
INCLUDED COMPONENTS ({len(present_components)} items):
{chr(10).join(f'- {c}' for c in present_components)}
PACKAGE COMPLETENESS: {completeness['score']}%
Generate the listing in this JSON structure:
{{
"title": "eBay listing title (max 80 chars, keyword-rich)",
"subtitle": "Optional subtitle (max 55 chars)",
"description_html": "Full HTML listing description with professional formatting, sections for What's Included, Why This Business, Getting Started, and FAQ",
"bullet_points": ["5-7 key selling points"],
"item_specifics": {{
"Business Type": "...",
"Industry": "...",
"Revenue Model": "...",
"Included Assets": "..."
}},
"search_keywords": ["10-15 relevant search terms"],
"condition_description": "Item condition note"
}}"""
system_prompt = (
"You are an expert eBay listing copywriter specializing in digital businesses and turnkey business packages. "
"Write compelling, trust-building copy that converts browsers into buyers. "
"Use proven eBay listing best practices: keyword-rich titles, scannable descriptions, clear value propositions. "
"Always respond with valid JSON."
)
try:
result = call_llm_routed("market_research", prompt, system_prompt=system_prompt, max_tokens=4096)
parsed = json.loads(result)
except (json.JSONDecodeError, Exception) as e:
logger.error(f"Listing copy generation failed: {e}")
parsed = _fallback_listing(domain, niche, tier_config, present_components, pricing)
parsed["pricing"] = pricing
parsed["category"] = category
parsed["completeness"] = {
"score": completeness["score"],
"recommended_tier": completeness["recommended_tier"],
}
return parsed
def _fallback_listing(domain: str, niche: str, tier_config: dict, components: list, pricing: dict) -> dict:
return {
"title": f"{tier_config['label']} - {niche} Business-in-a-Box | {domain}"[:80],
"subtitle": f"Complete Turnkey {niche} Business Package"[:55],
"description_html": f"{tier_config['label']}: {niche} Business Package
{tier_config['description']}
",
"bullet_points": components[:7],
"item_specifics": {
"Business Type": "Turnkey Online Business",
"Industry": niche,
"Revenue Model": "Multiple (Affiliate, Digital Products)",
"Included Assets": f"{len(components)} components",
},
"search_keywords": [niche, "business-in-a-box", "turnkey business", "online business", domain],
"condition_description": "Brand new, never operated business package",
}
async def run_market_research(domain: str, niche: str) -> dict:
prompt = f"""Research the current marketplace landscape for "{niche}" businesses, particularly as sold on eBay, Flippa, and similar platforms.
RESEARCH TARGET:
- Domain: {domain}
- Niche: {niche}
Provide analysis in this JSON structure:
{{
"niche_demand": {{
"score": 1-10,
"trend": "rising|stable|declining",
"explanation": "..."
}},
"competition": {{
"level": "low|medium|high|saturated",
"active_sellers": "estimated count or range",
"differentiation_opportunities": ["..."]
}},
"pricing_intelligence": {{
"similar_listings_range": "$X - $Y",
"average_selling_price": "$X",
"premium_packages_range": "$X - $Y",
"what_commands_premium": ["..."]
}},
"buyer_personas": [
{{
"type": "...",
"motivation": "...",
"price_sensitivity": "low|medium|high",
"key_decision_factors": ["..."]
}}
],
"marketplace_tips": ["3-5 actionable tips for selling this type of package"],
"seasonal_factors": "Any seasonality notes",
"recommended_platforms": ["ranked list of best platforms to sell this package"]
}}"""
system_prompt = (
"You are a marketplace intelligence analyst specializing in digital business sales, "
"eBay business listings, Flippa, and online business marketplaces. "
"Provide data-driven, actionable intelligence grounded in current market conditions. "
"Always respond with valid JSON."
)
try:
result = call_llm_routed("market_research", prompt, system_prompt=system_prompt, max_tokens=4096)
parsed = json.loads(result)
parsed["_meta"] = {
"source": "market_research_pipeline",
"generated_at": datetime.utcnow().isoformat(),
"domain": domain,
"niche": niche,
}
return parsed
except (json.JSONDecodeError, Exception) as e:
logger.error(f"Market research failed: {e}")
return {
"error": str(e),
"niche_demand": {"score": 5, "trend": "stable", "explanation": "Market research unavailable"},
"competition": {"level": "medium", "active_sellers": "unknown", "differentiation_opportunities": []},
"pricing_intelligence": {"similar_listings_range": "N/A", "average_selling_price": "N/A"},
"buyer_personas": [],
"marketplace_tips": ["Ensure package is complete before listing"],
"_meta": {
"source": "fallback",
"generated_at": datetime.utcnow().isoformat(),
"domain": domain,
"niche": niche,
},
}
def get_all_tiers() -> dict:
return {
tier_id: {
**tier,
"component_details": [
{
"id": comp_id,
"label": PACKAGE_COMPONENTS.get(comp_id, {}).get("label", comp_id),
"weight": PACKAGE_COMPONENTS.get(comp_id, {}).get("weight", 1),
}
for comp_id in tier["includes"]
],
}
for tier_id, tier in PRICING_TIERS.items()
}
def get_marketplace_summary(package_data: dict) -> dict:
completeness = score_package_completeness(package_data)
pricing = calculate_listing_price(package_data)
tier = completeness["recommended_tier"]
category = get_ebay_category(package_data.get("niche", ""), tier)
return {
"completeness": completeness,
"pricing": pricing,
"category": category,
"tiers": get_all_tiers(),
"package_value_summary": {
"components_present": len(completeness["present"]),
"components_total": len(PACKAGE_COMPONENTS),
"score_label": _score_label(completeness["score"]),
"ready_to_list": completeness["score"] >= 60,
},
}
def _score_label(score: float) -> str:
if score >= 90:
return "Legendary"
if score >= 75:
return "Premium"
if score >= 50:
return "Professional"
if score >= 25:
return "Starter"
return "Incomplete"