Arquitecto de Pipelines de Recomendacion

# cultiva_feed/pipeline.py
# Feed "Para Ti" — CULTIVA IA · Arquitecto de Pipelines de Recomendacion
# Patron: Source → Hydrator → Filter → Scorer → Selector → SideEffect

from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
from typing import Protocol, Sequence
from fastapi import FastAPI, Depends

# ── Tipos base ───────────────────────────────────────────────────────────────

@dataclass
class SkillCandidate:
    slug: str
    servicio: str
    tipo: str
    nivel: str
    bucket: str
    precio_eur: float
    acceso: str = "free"
    # hydrated fields
    usage_count: int = 0
    context_overlap: float = 0.0
    global_popularity: int = 0
    # scores
    p_use: float = 0.0
    p_share: float = 0.0
    p_skip: float = 0.0
    p_archive: float = 0.0
    final_score: float = 0.0

@dataclass
class FeedContext:
    user_id: str
    project_context: str   # e.g. "SEO", "CRO", "Automatizacion"
    session_ts: float
    user_nivel: str = "intermedio"
    plan: str = "premium"

# ── Protocolos (interfaces) ───────────────────────────────────────────────────

class Source(Protocol):
    async def fetch(self, ctx: FeedContext) -> list[SkillCandidate]: ...

class Hydrator(Protocol):
    async def hydrate(self, cands: list[SkillCandidate], ctx: FeedContext) -> None: ...

class Filter(Protocol):
    def apply(self, cands: list[SkillCandidate], ctx: FeedContext) -> list[SkillCandidate]: ...

class Scorer(Protocol):
    def score(self, cands: list[SkillCandidate], ctx: FeedContext) -> None: ...

class SideEffect(Protocol):
    async def run(self, top_k: list[SkillCandidate], ctx: FeedContext) -> None: ...

# ── Pipeline principal ────────────────────────────────────────────────────────

@dataclass
class FeedPipeline:
    sources:      list[Source]
    hydrators:    list[Hydrator]
    filters:      list[Filter]
    scorers:      list[Scorer]
    side_effects: list[SideEffect]
    top_k:        int = 10
    max_per_svc:  int = 3   # diversidad: max skills por servicio

    async def run(self, ctx: FeedContext) -> list[SkillCandidate]:
        # 1. SOURCE — paralelo
        batches = await asyncio.gather(*[s.fetch(ctx) for s in self.sources])
        cands = dedupe(c for b in batches for c in b)

        # 2. HYDRATOR — paralelo
        await asyncio.gather(*[h.hydrate(cands, ctx) for h in self.hydrators])

        # 3. FILTER — secuencial (cheap → expensive)
        for f in self.filters:
            cands = f.apply(cands, ctx)

        # 4. SCORER — secuencial (cada scorer ve los anteriores)
        for sc in self.scorers:
            sc.score(cands, ctx)

        # 5. SELECTOR — sort + top-K + mix estratificado
        top_k = self._select(cands)

        # 6. SIDE EFFECTS — fire-and-forget, nunca bloquea
        for se in self.side_effects:
            asyncio.create_task(se.run(top_k, ctx))   # no await

        return top_k

    def _select(self, cands: list[SkillCandidate]) -> list[SkillCandidate]:
        sorted_c = sorted(cands, key=lambda c: c.final_score, reverse=True)
        result, svc_count = [], {}
        for c in sorted_c:
            if len(result) == self.top_k: break
            if svc_count.get(c.servicio, 0) < self.max_per_svc:
                result.append(c)
                svc_count[c.servicio] = svc_count.get(c.servicio, 0) + 1
        return result

# ── Scorer multi-accion ───────────────────────────────────────────────────────

@dataclass
class MultiActionScorer:
    w_use:     float =  1.0
    w_share:   float =  0.5
    w_skip:    float = -0.3
    w_archive: float = -0.5

    def score(self, cands: list[SkillCandidate], ctx: FeedContext) -> None:
        for c in cands:
            c.p_use     = _predict_use(c, ctx)
            c.p_share   = _predict_share(c, ctx)
            c.p_skip    = _predict_skip(c, ctx)
            c.p_archive = _predict_archive(c, ctx)
            c.final_score = (
                self.w_use     * c.p_use
                + self.w_share   * c.p_share
                + self.w_skip    * c.p_skip
                + self.w_archive * c.p_archive
            )

# ── FastAPI endpoint ──────────────────────────────────────────────────────────

app = FastAPI(title="CULTIVA Feed API")

@app.get("/feed/{user_id}")
async def get_feed(user_id: str, project: str = "general"):
    ctx = FeedContext(user_id=user_id, project_context=project,
                      session_ts=time())
    top_k = await pipeline.run(ctx)
    return [asdict(c) for c in top_k]