Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/AI_Brain / services/seeder_api/app.py |
| Date | Not specified in source |
| Repository | Janady13/AI_Brain |
| Revision | 840945d70b88674d8f40d330001cc63d871dc8e2 |
| Branch | codex/build-online-github-ani-nbgsix |
| Path or archive member | services/seeder_api/app.py |
| Original SHA-256 | 3b503cd13afe20aa046ed47338610684c7c66cb06a2d9f8d3cacea3e2e61b39d |
| Git blob SHA | 7cf53c89fff2555f84ef7eeeb5e8882f484e3629 |
| Reading-copy SHA-256 | 3b503cd13afe20aa046ed47338610684c7c66cb06a2d9f8d3cacea3e2e61b39d |
Recovered source transcript
Download complete source text ↓493 lines · 15.9 KB
#!/usr/bin/env python3
import os
import time
from fastapi import FastAPI
from pydantic import BaseModel
from .intelligence import (
AmazonProductRecommendation,
AWSRevenueRecommendation,
CampaignMetrics,
CampaignScore,
CJAffiliateRecommendation,
CryptoMarketSnapshot,
GrowthExperiment,
GoogleCampaignPerformance,
MarketingIntelligenceEngine,
MarketingCampaign,
PromptDiagnostics,
StripeMonetizationPlan,
default_marketing_engine,
)
APP_NAME = os.getenv("APP_NAME", "HiveSeedAPI")
START_TS = time.time()
app = FastAPI(title=APP_NAME)
class Health(BaseModel):
status: str
uptime_seconds: float
class Ready(BaseModel):
ready: bool
reason: str | None = None
class MarketingQuery(BaseModel):
prompt: str
limit: int = 3
class MarketingInsight(BaseModel):
channel: str
audience: str
offer: str
objective: str
rationale: str
class MarketingResponse(BaseModel):
query: str
recommendations: list[MarketingInsight]
insight: str
class StatisticalInsight(BaseModel):
channel: str
audience: str
offer: str
objective: str
expected_roi: float
expected_ctr: float
confidence: float
rationale: str
drivers: dict[str, float]
class PortfolioInsight(BaseModel):
expected_roi: float
expected_ctr: float
confidence: float
rationale: str
strongest_drivers: dict[str, float]
class StatisticalResponse(BaseModel):
query: str
recommendations: list[StatisticalInsight]
portfolio: PortfolioInsight
class DiagnosticResponse(BaseModel):
query: str
stage: str
urgency: str
focus: str
confidence: float
recommended_kpis: list[str]
opportunities: list[str]
risk_signals: list[str]
keyword_summary: dict[str, list[str]]
class ProcessRun(BaseModel):
process: str
recommendations: list[MarketingInsight]
summary: str
class ExperimentInsight(BaseModel):
name: str
hypothesis: str
primary_metric: str
secondary_metrics: list[str]
playbook: list[str]
confidence: float
class MarketingProcessesResponse(BaseModel):
query: str
rag: ProcessRun
egg: list[ExperimentInsight]
google_ads: list["GoogleAdsInsight"]
cj_affiliate: list["CJAffiliateInsight"]
crypto_markets: list["CryptoMarketInsight"]
stripe: list["StripePlanInsight"]
amazon_marketplace: list["AmazonMarketplaceInsight"]
aws_cloud: list["AWSRevenueInsight"]
class GoogleAdsInsight(BaseModel):
campaign_name: str
clicks: int
impressions: int
conversions: float
cost_micros: int
ctr: float
cpc: float
roas: float
source: str
notes: str | None = None
class CJAffiliateInsight(BaseModel):
program_name: str
advertiser_id: str
category: str
commission_rate: float
epc: float
seven_day_epc: float
projected_epc: float
projected_monthly_revenue: float
network_rank: int
payout_terms: str
landing_url: str
rationale: str
source: str
notes: str | None = None
class CryptoMarketInsight(BaseModel):
asset: str
price_usd: float | None
change_24h: float | None
market_cap: float | None
volume_24h: float | None
source: str
notes: str | None = None
class StripePlanInsight(BaseModel):
product_name: str
price_usd: float
billing_interval: str
currency: str
rationale: str
setup_steps: list[str]
source: str
notes: str | None = None
class AmazonMarketplaceInsight(BaseModel):
asin: str
title: str
category: str
price_usd: float
commission_rate: float
rating: float
reviews: int
projected_monthly_revenue: float
rationale: str
url: str
source: str
notes: str | None = None
class AWSRevenueInsight(BaseModel):
program_name: str
focus: str
description: str
revenue_motion: str
primary_service: str
secondary_services: list[str]
automation_play: str
projected_roi: float
source: str
notes: str | None = None
try: # Pydantic v2
GoogleAdsInsight.model_rebuild()
CJAffiliateInsight.model_rebuild()
CryptoMarketInsight.model_rebuild()
StripePlanInsight.model_rebuild()
AmazonMarketplaceInsight.model_rebuild()
AWSRevenueInsight.model_rebuild()
MarketingProcessesResponse.model_rebuild()
except AttributeError: # pragma: no cover - pydantic v1 fallback
MarketingProcessesResponse.update_forward_refs(
GoogleAdsInsight=GoogleAdsInsight,
CJAffiliateInsight=CJAffiliateInsight,
CryptoMarketInsight=CryptoMarketInsight,
StripePlanInsight=StripePlanInsight,
AmazonMarketplaceInsight=AmazonMarketplaceInsight,
AWSRevenueInsight=AWSRevenueInsight,
)
try: # Pydantic v2
MarketingProcessesResponse.model_rebuild()
except AttributeError: # pragma: no cover - pydantic v1 fallback
MarketingProcessesResponse.update_forward_refs(GoogleAdsInsight=GoogleAdsInsight)
@app.get("/healthz", response_model=Health)
def healthz() -> Health:
return Health(status="ok", uptime_seconds=time.time() - START_TS)
@app.get("/readyz", response_model=Ready)
def readyz() -> Ready:
# Minimal readiness: enumerations exist or DRY_RUN enabled
base = os.path.join(os.path.dirname(__file__), "..", "..", "daemon", "enumerations")
files = [
os.path.join(base, "math_model_500.txt"),
os.path.join(base, "agent_500.txt"),
os.path.join(base, "language_500.txt"),
os.path.join(base, "prophecy_500.txt"),
os.path.join(base, "marketing_500.txt"),
]
if all(os.path.isfile(p) for p in files):
return Ready(ready=True)
if os.getenv("DRY_RUN", "0") in {"1", "true", "TRUE", "yes", "on"}:
return Ready(ready=True, reason="DRY_RUN enabled")
return Ready(ready=False, reason="Enumerations missing; run make enumerations")
def _campaign_to_insight(engine: MarketingIntelligenceEngine, query: str, campaign: MarketingCampaign) -> MarketingInsight:
tokens = engine.keywords(query)
matches: list[str] = []
normalized = {
"channel": campaign.channel.lower(),
"audience": campaign.audience.lower(),
"offer": campaign.offer.lower(),
"objective": campaign.objective.lower(),
}
for token in tokens:
for feature, text in normalized.items():
if token in text and feature not in matches:
matches.append(feature)
rationale = "Matched: " + ", ".join(matches) if matches else "General fit from curated library"
return MarketingInsight(
channel=campaign.channel,
audience=campaign.audience,
offer=campaign.offer,
objective=campaign.objective,
rationale=rationale,
)
def _campaign_to_statistical_insight(
engine: MarketingIntelligenceEngine,
query: str,
scored: CampaignScore,
) -> StatisticalInsight:
matches: list[str] = []
for feature, count in scored.keyword_coverage.items():
if count:
matches.append(feature)
if matches:
rationale = f"Matched {', '.join(sorted(matches))} intent with statistical lift."
else:
rationale = "Benchmark-driven pick from marketing corpus."
strongest = sorted(scored.metrics.drivers.items(), key=lambda item: item[1], reverse=True)
if strongest:
rationale += f" Top driver: {strongest[0][0]} ROI {strongest[0][1]:.2f}."
return StatisticalInsight(
channel=scored.campaign.channel,
audience=scored.campaign.audience,
offer=scored.campaign.offer,
objective=scored.campaign.objective,
expected_roi=round(scored.metrics.expected_roi, 2),
expected_ctr=round(scored.metrics.expected_ctr, 4),
confidence=round(scored.metrics.confidence, 2),
rationale=rationale,
drivers={key: round(value, 2) for key, value in scored.metrics.drivers.items()},
)
def _portfolio_to_insight(query: str, portfolio: CampaignMetrics, tokens: list[str]) -> PortfolioInsight:
highlights = ", ".join(sorted(set(tokens))) or "broad"
strongest = sorted(portfolio.drivers.items(), key=lambda item: item[1], reverse=True)
driver_text = strongest[0][0] if strongest else "channel"
rationale = (
f"Statistical blend for '{query}' optimized for {highlights} focus; strongest lift from {driver_text}."
)
return PortfolioInsight(
expected_roi=round(portfolio.expected_roi, 2),
expected_ctr=round(portfolio.expected_ctr, 4),
confidence=round(portfolio.confidence, 2),
rationale=rationale,
strongest_drivers={key: round(value, 2) for key, value in portfolio.drivers.items()},
)
@app.post("/intelligence/marketing", response_model=MarketingResponse)
def marketing_intelligence(query: MarketingQuery) -> MarketingResponse:
engine = default_marketing_engine()
campaigns = engine.recommend(query.prompt, query.limit)
insights = [_campaign_to_insight(engine, query.prompt, campaign) for campaign in campaigns]
narrative = engine.summarize(query.prompt, campaigns)
return MarketingResponse(query=query.prompt, recommendations=insights, insight=narrative)
@app.post("/intelligence/marketing/statistical", response_model=StatisticalResponse)
def marketing_statistical_intelligence(query: MarketingQuery) -> StatisticalResponse:
engine = default_marketing_engine()
ranked = engine.recommend_statistical(query.prompt, query.limit)
insights = [
_campaign_to_statistical_insight(engine, query.prompt, scored) for scored in ranked
]
tokens = engine.keywords(query.prompt)
portfolio_metrics = engine.portfolio(query.prompt, ranked)
portfolio = _portfolio_to_insight(query.prompt, portfolio_metrics, tokens)
return StatisticalResponse(query=query.prompt, recommendations=insights, portfolio=portfolio)
def _diagnostics_payload(query: str, diagnostics: PromptDiagnostics) -> DiagnosticResponse:
return DiagnosticResponse(
query=query,
stage=diagnostics.stage,
urgency=diagnostics.urgency,
focus=diagnostics.focus,
confidence=diagnostics.confidence,
recommended_kpis=diagnostics.recommended_kpis,
opportunities=diagnostics.opportunities,
risk_signals=diagnostics.risk_signals,
keyword_summary=diagnostics.keyword_summary,
)
def _experiment_payload(experiment: GrowthExperiment) -> ExperimentInsight:
return ExperimentInsight(
name=experiment.name,
hypothesis=experiment.hypothesis,
primary_metric=experiment.primary_metric,
secondary_metrics=experiment.secondary_metrics,
playbook=experiment.playbook,
confidence=experiment.confidence,
)
def _google_ads_payload(performance: GoogleCampaignPerformance) -> "GoogleAdsInsight":
return GoogleAdsInsight(
campaign_name=performance.campaign_name,
clicks=performance.clicks,
impressions=performance.impressions,
conversions=performance.conversions,
cost_micros=performance.cost_micros,
ctr=performance.ctr,
cpc=performance.cpc,
roas=performance.roas,
source=performance.source,
notes=performance.notes,
)
def _cj_affiliate_payload(recommendation: CJAffiliateRecommendation) -> "CJAffiliateInsight":
return CJAffiliateInsight(
program_name=recommendation.program_name,
advertiser_id=recommendation.advertiser_id,
category=recommendation.category,
commission_rate=recommendation.commission_rate,
epc=recommendation.epc,
seven_day_epc=recommendation.seven_day_epc,
projected_epc=recommendation.projected_epc,
projected_monthly_revenue=recommendation.projected_monthly_revenue,
network_rank=recommendation.network_rank,
payout_terms=recommendation.payout_terms,
landing_url=recommendation.landing_url,
rationale=recommendation.rationale,
source=recommendation.source,
notes=recommendation.notes,
)
def _crypto_market_payload(snapshot: CryptoMarketSnapshot) -> "CryptoMarketInsight":
return CryptoMarketInsight(
asset=snapshot.asset,
price_usd=snapshot.price_usd,
change_24h=snapshot.change_24h,
market_cap=snapshot.market_cap,
volume_24h=snapshot.volume_24h,
source=snapshot.source,
notes=snapshot.notes,
)
def _stripe_plan_payload(plan: StripeMonetizationPlan) -> "StripePlanInsight":
return StripePlanInsight(
product_name=plan.product_name,
price_usd=plan.price_usd,
billing_interval=plan.billing_interval,
currency=plan.currency,
rationale=plan.rationale,
setup_steps=plan.setup_steps,
source=plan.source,
notes=plan.notes,
)
def _amazon_listing_payload(
recommendation: AmazonProductRecommendation,
) -> "AmazonMarketplaceInsight":
return AmazonMarketplaceInsight(
asin=recommendation.asin,
title=recommendation.title,
category=recommendation.category,
price_usd=recommendation.price_usd,
commission_rate=recommendation.commission_rate,
rating=recommendation.rating,
reviews=recommendation.reviews,
projected_monthly_revenue=recommendation.projected_monthly_revenue,
rationale=recommendation.rationale,
url=recommendation.url,
source=recommendation.source,
notes=recommendation.notes,
)
def _aws_revenue_payload(recommendation: AWSRevenueRecommendation) -> "AWSRevenueInsight":
return AWSRevenueInsight(
program_name=recommendation.program_name,
focus=recommendation.focus,
description=recommendation.description,
revenue_motion=recommendation.revenue_motion,
primary_service=recommendation.primary_service,
secondary_services=recommendation.secondary_services,
automation_play=recommendation.automation_play,
projected_roi=recommendation.projected_roi,
source=recommendation.source,
notes=recommendation.notes,
)
@app.post("/intelligence/marketing/diagnostics", response_model=DiagnosticResponse)
def marketing_diagnostics(query: MarketingQuery) -> DiagnosticResponse:
engine = default_marketing_engine()
diagnostics = engine.diagnose(query.prompt)
return _diagnostics_payload(query.prompt, diagnostics)
@app.post("/intelligence/marketing/processes", response_model=MarketingProcessesResponse)
def marketing_processes(query: MarketingQuery) -> MarketingProcessesResponse:
engine = default_marketing_engine()
rag_campaigns = engine.recommend(query.prompt, query.limit)
rag_insights = [
_campaign_to_insight(engine, query.prompt, campaign) for campaign in rag_campaigns
]
rag_summary = engine.summarize(query.prompt, rag_campaigns)
ranked = engine.recommend_statistical(query.prompt, query.limit)
egg_experiments = engine.expansion_grid(query.prompt, ranked)
google_ads_data = engine.google_ads_enrichment(query.prompt, ranked, query.limit)
cj_affiliate = engine.cj_affiliate_enrichment(query.prompt, ranked, query.limit)
crypto_markets = engine.crypto_market_enrichment(query.prompt, query.limit)
stripe_plans = engine.stripe_checkout_enrichment(query.prompt, ranked, query.limit)
amazon_listings = engine.amazon_marketplace_enrichment(query.prompt, ranked, query.limit)
aws_programs = engine.aws_cloud_enrichment(query.prompt, ranked, query.limit)
return MarketingProcessesResponse(
query=query.prompt,
rag=ProcessRun(process="rag", recommendations=rag_insights, summary=rag_summary),
egg=[_experiment_payload(experiment) for experiment in egg_experiments],
google_ads=[_google_ads_payload(performance) for performance in google_ads_data],
cj_affiliate=[_cj_affiliate_payload(rec) for rec in cj_affiliate],
crypto_markets=[_crypto_market_payload(snapshot) for snapshot in crypto_markets],
stripe=[_stripe_plan_payload(plan) for plan in stripe_plans],
amazon_marketplace=[_amazon_listing_payload(rec) for rec in amazon_listings],
aws_cloud=[_aws_revenue_payload(rec) for rec in aws_programs],
)