from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException from sqlmodel import Session, select from app.api.deps import get_session from app.models.core import AnalyticsPostSnapshot, ContentScript, ExternalPost, Idea, Theme, VideoAsset, WebsiteCompanion from app.services.analytics_service import latest_snapshots from app.services.production_queue import add_recommendation router = APIRouter(prefix="/content", tags=["content"]) def _family_summary(session: Session, family: Theme) -> dict: videos = session.exec(select(VideoAsset).where(VideoAsset.topic_id == family.id)).all() scripts = session.exec(select(ContentScript).where(ContentScript.family_id == family.id)).all() ideas = session.exec(select(Idea).where(Idea.topic_id == family.id)).all() posts = session.exec(select(ExternalPost).where(ExternalPost.topic_id == family.id, ExternalPost.mapping_status != "ignored")).all() post_ids = {p.external_post_id for p in posts} video_ids = {v.id for v in videos} latest = latest_snapshots(session) family_snaps = [s for s in latest if (s.video_asset_id and s.video_asset_id in video_ids) or s.external_post_id in post_ids] views = sum(s.views or 0 for s in family_snaps) best = max(family_snaps, key=lambda s: s.views or 0, default=None) retention_values = [s.retention_proxy_pct for s in family_snaps if s.retention_proxy_pct is not None] subscribers = sum(s.subscribers_delta or 0 for s in family_snaps) guides = session.exec(select(WebsiteCompanion)).all() guide_video_ids = {g.video_asset_id for g in guides} return { "id": family.id, "title": family.title, "description": family.description, "audience": family.audience_pain, "promise": family.promise, "status": family.status, "priority": family.priority, "performance_score": family.performance_score, "produced_count": len(videos), "scripts_count": len(scripts), "ideas_count": len(ideas), "videos_with_analytics": len(family_snaps), "views_total": views, "avg_retention": round(sum(retention_values) / len(retention_values), 1) if retention_values else None, "subscriber_delta": subscribers, "website_guides": len([v for v in videos if v.id in guide_video_ids]), "best_video": best.title if best else None, "next_recommendation": _recommendation_for_family(family, views, best, len(ideas)), } def _recommendation_for_family(family: Theme, views: int, best: AnalyticsPostSnapshot | None, ideas_count: int) -> str: if best and (best.subscribers_delta or 0) > 0: return "Make another variant with the same promise." if best and (best.retention_proxy_pct or 0) > 10 and (best.views or 0) < 100: return "Try a clearer title or opening hook for this family." if views > 0 and ideas_count: return "Produce the highest-priority unmade variant next." return "Add one strong script idea and measure the first result." @router.get("/families") def families(session: Session = Depends(get_session)): rows = session.exec(select(Theme).order_by(Theme.priority.desc(), Theme.updated_at.desc())).all() items = [_family_summary(session, row) for row in rows] best = max(items, key=lambda item: item["views_total"], default=None) scripts = session.exec(select(ContentScript)).all() ideas = session.exec(select(Idea)).all() return { "summary": { "families_count": len(items), "scripts_produced": len([s for s in scripts if s.status in {"produced", "reviewed", "uploaded", "measured"}]), "ideas_waiting": len([i for i in ideas if i.status in {"idea", "needs_research"}]), "best_family": best["title"] if best else None, }, "items": sorted(items, key=lambda item: item["views_total"], reverse=True), } @router.get("/families/{family_id}") def family_detail(family_id: str, session: Session = Depends(get_session)): family = session.get(Theme, family_id) if not family: raise HTTPException(status_code=404, detail="Content family not found") scripts = session.exec(select(ContentScript).where(ContentScript.family_id == family_id).order_by(ContentScript.updated_at.desc())).all() ideas = session.exec(select(Idea).where(Idea.topic_id == family_id).order_by(Idea.priority.desc())).all() videos = session.exec(select(VideoAsset).where(VideoAsset.topic_id == family_id)).all() latest = {s.video_asset_id: s for s in latest_snapshots(session) if s.video_asset_id} return { "family": _family_summary(session, family), "scripts": [{**s.model_dump(mode="json"), "performance": latest.get(s.video_asset_id).model_dump(mode="json") if s.video_asset_id and latest.get(s.video_asset_id) else None} for s in scripts], "ideas": [i.model_dump(mode="json") for i in ideas], "videos": [v.model_dump(mode="json") for v in videos], } @router.get("/recommendations/next") def next_recommendation(session: Session = Depends(get_session)): families = session.exec(select(Theme).order_by(Theme.priority.desc())).all() if not families: return {"family": None, "idea": None, "script_id": None, "reason": ["No content families yet."], "confidence": "low", "recommended_action": "import_content_library"} summaries = [_family_summary(session, f) for f in families] chosen = max(summaries, key=lambda item: (item["views_total"], item["priority"])) family = session.get(Theme, chosen["id"]) idea = session.exec(select(Idea).where(Idea.topic_id == chosen["id"], Idea.status.in_(["idea", "needs_research"])).order_by(Idea.priority.desc())).first() script = session.exec(select(ContentScript).where(ContentScript.family_id == chosen["id"], ContentScript.status.in_(["idea", "scripted"])).order_by(ContentScript.updated_at.desc())).first() reason = [] best_snapshot = None if chosen["best_video"]: for snap in latest_snapshots(session): if snap.title == chosen["best_video"]: best_snapshot = snap break if chosen["views_total"] > 0: reason.append(f"{chosen['title']} is the strongest linked family right now with {chosen['views_total']} views.") if best_snapshot: reason.append(f"Best current video: {best_snapshot.title} with {best_snapshot.views or 0} views.") if (best_snapshot.views or 0) >= 40 and (best_snapshot.retention_proxy_pct or 0) < 10: reason.append("Views are strong but retention is modest: make the next version with a faster hook and shorter setup.") elif (best_snapshot.views or 0) < 50 and (best_snapshot.retention_proxy_pct or 0) >= 10: reason.append("Views are still low but retention is usable: improve packaging, title, or thumbnail before changing the idea.") if idea: reason.append("There is an unproduced idea ready for a follow-up variant.") if not reason: reason.append("This is the highest-priority family with available content work.") return { "family": family.title if family else chosen["title"], "family_id": chosen["id"], "idea": idea.title if idea else (script.title if script else None), "script_id": script.id if script else None, "reason": reason, "confidence": "medium" if chosen["views_total"] else "low", "recommended_action": "produce_variant" if idea or script else "add_idea", } @router.post("/recommendations/next/add-to-queue") def add_next_recommendation_to_queue(session: Session = Depends(get_session)): recommendation = next_recommendation(session) return add_recommendation(session, recommendation)