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 | 49db80a641b10ab62d16cc7643bd0b367f93951c |
| Branch | codex/build-online-github-ani |
| Path or archive member | services/seeder_api/app.py |
| Original SHA-256 | 41d57e60483f3c58bb59e7c1fe21049c57bcb1e80a1615d0c5222bbf13ae4897 |
| Git blob SHA | 089ca89b40c5ba7fad9921efe40fa62958000b06 |
| Reading-copy SHA-256 | 41d57e60483f3c58bb59e7c1fe21049c57bcb1e80a1615d0c5222bbf13ae4897 |
Recovered source transcript
Download complete source text ↓312 lines · 9.9 KB
#!/usr/bin/env python3
import os
import time
from fastapi import FastAPI
from pydantic import BaseModel
from .intelligence import (
CampaignMetrics,
CampaignScore,
GrowthExperiment,
GoogleCampaignPerformance,
MarketingIntelligenceEngine,
MarketingCampaign,
PromptDiagnostics,
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"]
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
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,
)
@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)
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],
)