Janady13/AI_Brain / skynet/handlers.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / skynet/handlers.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	cdb497ece2d1a770ca8c83eb1de8923cfaa9eceb
Branch	feature/seeder-engine-tests
Path or archive member	skynet/handlers.py
Original SHA-256	85b4877cf4cbf45bbf6877e625168f8105accfd461f974e7371f9ba37d035fdd
Git blob SHA	5b32edd8447225d8f7583c0c9e5f3a689ecfaf6b
Reading-copy SHA-256	85b4877cf4cbf45bbf6877e625168f8105accfd461f974e7371f9ba37d035fdd

Original source ↗

Recovered source transcript

Download complete source text ↓140 lines · 5.2 KB

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
from typing import Any, Dict, List
from config.stripe_secrets import apply_env_from_stripe_db
import subprocess

try:
    from infra import omnicron_iface as omni  # type: ignore
except Exception:  # pragma: no cover
    class _Dummy:
        @staticmethod
        def emit_log(event: str, level: str | None = None, payload: Dict[str, Any] | None = None) -> None:
            pass
    omni = _Dummy()  # type: ignore


def handle_planner_run() -> Dict[str, Any]:
    # Placeholder integration; real planner wires into DB-backed planner when present
    return {"ok": True, "notes": "planner.run stub"}


def handle_alignment_evaluate() -> Dict[str, Any]:
    return {"ok": True, "notes": "alignment.evaluate stub"}


def handle_metabrain_consolidate() -> Dict[str, Any]:
    return {"ok": True, "notes": "metabrain.consolidate stub"}


def handle_executor_run() -> Dict[str, Any]:
    return {"ok": True, "notes": "executor.run stub"}


def handle_github_sync() -> Dict[str, Any]:
    # For this codebase, discovery/sync tooling is external; return stub and emit log
    omni.emit_log("github_sync_stub", payload={"note": "use external sync tooling to clone/update and ingest"})
    return {"ok": True}


# Revenue-oriented handlers — integrate with seeder_api marketing intelligence
def _marketing_engine():
    try:
        from services.seeder_api.intelligence import MarketingIntelligenceEngine, default_marketing_engine  # type: ignore
        return default_marketing_engine()
    except Exception:
        return None


def _mk_insight(channel: str, audience: str, offer: str, objective: str, rationale: str) -> Dict[str, Any]:
    return {"channel": channel, "audience": audience, "offer": offer, "objective": objective, "rationale": rationale}


def handle_marketing_publish() -> Dict[str, Any]:
    eng = _marketing_engine()
    if not eng:
        return {"ok": False, "error": "marketing_engine_unavailable"}
    # Produce a few recommendations and a summary narrative
    recs = eng.recommendations_for("AI revenue strategies", limit=3)
    out: List[Dict[str, Any]] = []
    for r in recs:
        out.append(_mk_insight(r.channel, r.audience, r.offer, r.objective, r.rationale))
    portfolio = eng.statistical_for("AI revenue strategies", limit=3)
    return {"ok": True, "recommendations": out, "portfolio": {
        "expected_roi": portfolio.portfolio.expected_roi,
        "expected_ctr": portfolio.portfolio.expected_ctr,
        "confidence": portfolio.portfolio.confidence,
        "rationale": portfolio.portfolio.rationale,
    }}


def handle_affiliate_publish() -> Dict[str, Any]:
    eng = _marketing_engine()
    if not eng:
        return {"ok": False, "error": "marketing_engine_unavailable"}
    recs = eng.recommendations_for("Affiliate monetization for AI tools", limit=3)
    out: List[Dict[str, Any]] = []
    for r in recs:
        out.append(_mk_insight(r.channel, r.audience, r.offer, r.objective, r.rationale))
    return {"ok": True, "recommendations": out}


def handle_commerce_cashapp_promote() -> Dict[str, Any]:
    # Simple integration: hand back a crafted promotion text; real posting would wire to channel APIs
    link = os.getenv("CASHAPP_LINK", "")
    promo = f"Support our work instantly with Cash App: {link}. Join the AI journey and unlock new revenue plays."
    return {"ok": True, "message": promo}


def handle_commerce_venmo_promote() -> Dict[str, Any]:
    link = os.getenv("VENMO_LINK", "")
    promo = f"Fuel AI growth — tip via Venmo: {link}. Every contribution accelerates the roadmap."
    return {"ok": True, "message": promo}


def handle_revenue_stripe_refresh_keys() -> Dict[str, Any]:
    applied = apply_env_from_stripe_db()
    return {"applied": bool(applied), **applied}


def handle_revenue_stripe_checkout() -> Dict[str, Any]:
    try:
        import stripe  # type: ignore
    except Exception as exc:
        return {"error": f"stripe_sdk_missing: {exc}"}
    apply_env_from_stripe_db()
    key = os.environ.get("STRIPE_SECRET_KEY")
    if not key:
        return {"error": "stripe_not_configured"}
    try:
        stripe.api_key = key  # type: ignore
        session = stripe.checkout.Session.create(  # type: ignore
            mode="payment",
            line_items=[{
                "price_data": {
                    "currency": "usd",
                    "product_data": {"name": "Skynet Test"},
                    "unit_amount": 500,
                },
                "quantity": 1,
            }],
            success_url="https://example.com/success",
            cancel_url="https://example.com/cancel",
        )
        url = getattr(session, "url", None) or session.get("url")
        return {"session_id": getattr(session, "id", None) or session.get("id"), "url": url}
    except Exception as exc:
        return {"error": str(exc)}


def handle_revenue_stripe_consume() -> Dict[str, Any]:
    """Spawn the stripe consumer in a background process."""
    try:
        proc = subprocess.Popen(["python3", "workers/stripe_consumer.py"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return {"ok": True, "pid": proc.pid}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}