327 lines
11 KiB
Python
327 lines
11 KiB
Python
"""
|
||
BAN YARO — Social Media Manager
|
||
KI-gestützte Content-Erstellung für TikTok & Instagram.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from typing import Optional
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel
|
||
|
||
from auth import get_current_user, require_social_media
|
||
from database import db
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
|
||
_PLATFORMS = {"tiktok", "instagram", "both"}
|
||
_FORMATS = {"reel", "story", "post", "carousel"}
|
||
_STATUSES = {"idea", "draft", "scheduled", "published", "archived"}
|
||
|
||
_SYSTEM = (
|
||
"Du bist ein erfahrener Social-Media-Stratege für Ban Yaro, "
|
||
"die deutschsprachige Hunde-App (banyaro.app). "
|
||
"Zielgruppe: Hundebesitzer in DACH, 20–45 Jahre, Fokus Frauen. "
|
||
"Ton: warmherzig, authentisch, informativ — nie aufdringlich werblich. "
|
||
"Antworte immer auf Deutsch."
|
||
)
|
||
|
||
_PROMPT_GENERATE = '''\
|
||
Erstelle einen vollständigen Social-Media-Content-Plan für folgenden Post.
|
||
|
||
Plattform: {platform}
|
||
Format: {format}
|
||
Thema: {topic}
|
||
{breed_info}
|
||
Bereits verwendete Themen (nicht wiederholen):
|
||
{used_topics}
|
||
|
||
Antworte NUR mit einem JSON-Objekt:
|
||
{{
|
||
"caption": "Post-Text (plattformgerecht, mit Emojis, max 2200 Zeichen für Instagram / 150 für TikTok)",
|
||
"hashtags": "kommagetrennte Hashtags ohne #, 5-15 Stück, mix aus groß/nische",
|
||
"hook": "Ersten 1-3 Sätze / Sekunden — sofortiger Aufmerksamkeitsfänger",
|
||
"cta": "Call-to-Action am Ende (Frage, Aufforderung zum Kommentieren/Teilen)",
|
||
"visual_brief": "Detaillierte Beschreibung was auf dem Foto/Video zu sehen sein soll (Motiv, Stimmung, Licht, Perspektive)",
|
||
"image_prompt": "Englischer DALL-E/Midjourney Prompt für ein passendes Illustration/Cartoon (wenn kein eigenes Foto verfügbar)",
|
||
"canva_notes": "Empfehlung für Canva: Template-Typ, Textüberlagerungen, Farbstimmung",
|
||
"script": {script_field},
|
||
"unsplash_query": "2-3 englische Suchbegriffe für Unsplash-Stockfotos",
|
||
"ai_score": <Zahl 1-5, geschätztes Engagement-Potenzial>
|
||
}}
|
||
'''
|
||
|
||
_SCRIPT_FIELD_REEL = '''"Videostruktur: Hook (0-3s): ... | Hauptteil (3-25s): Schritt 1... Schritt 2... | CTA (25-30s): ..."'''
|
||
_SCRIPT_FIELD_OTHER = "null"
|
||
|
||
_PROMPT_EVALUATE = '''\
|
||
Bewerte und verbessere diesen Social-Media-Entwurf für Ban Yaro (Hunde-App).
|
||
|
||
Plattform: {platform}
|
||
Format: {format}
|
||
Entwurf:
|
||
{draft}
|
||
|
||
Antworte NUR mit einem JSON-Objekt:
|
||
{{
|
||
"caption": "Verbesserter Post-Text",
|
||
"hashtags": "Optimierte Hashtags, kommagetrennt ohne #",
|
||
"hook": "Verbesserter Hook",
|
||
"cta": "Verbesserter CTA",
|
||
"visual_brief": "Empfehlung für passendes Bildmaterial",
|
||
"image_prompt": "DALL-E/Midjourney Prompt für Illustration",
|
||
"canva_notes": "Canva-Tipps",
|
||
"script": null,
|
||
"unsplash_query": "Unsplash-Suchbegriffe",
|
||
"ai_score": <1-5>,
|
||
"feedback": "Kurzes Feedback was gut war und was verbessert wurde (2-3 Sätze)"
|
||
}}
|
||
'''
|
||
|
||
|
||
class GenerateRequest(BaseModel):
|
||
platform: str = "both"
|
||
format: str = "post"
|
||
topic: str
|
||
breed_id: Optional[int] = None
|
||
|
||
|
||
class EvaluateRequest(BaseModel):
|
||
platform: str = "instagram"
|
||
format: str = "post"
|
||
draft: str
|
||
|
||
|
||
class StatusUpdate(BaseModel):
|
||
status: Optional[str] = None
|
||
scheduled_at: Optional[str] = None
|
||
published_at: Optional[str] = None
|
||
notes: Optional[str] = None
|
||
|
||
|
||
def _used_topics(limit: int = 30) -> str:
|
||
with db() as conn:
|
||
rows = conn.execute(
|
||
"SELECT topic FROM social_content ORDER BY created_at DESC LIMIT ?",
|
||
(limit,),
|
||
).fetchall()
|
||
if not rows:
|
||
return "(noch keine)"
|
||
return "\n".join(f"- {r['topic']}" for r in rows)
|
||
|
||
|
||
def _breed_info(breed_id: int | None) -> str:
|
||
if not breed_id:
|
||
return ""
|
||
with db() as conn:
|
||
r = conn.execute(
|
||
"SELECT name, beschreibung, temperament, groesse FROM wiki_rassen WHERE id=?",
|
||
(breed_id,),
|
||
).fetchone()
|
||
if not r:
|
||
return ""
|
||
parts = [f"Rasse: {r['name']}"]
|
||
if r["beschreibung"]:
|
||
parts.append(f"Info: {r['beschreibung'][:300]}")
|
||
if r["temperament"]:
|
||
parts.append(f"Temperament: {r['temperament']}")
|
||
return "\n".join(parts)
|
||
|
||
|
||
async def _ki_complete(prompt: str) -> str:
|
||
import ki as ki_module
|
||
return await ki_module.complete(
|
||
prompt,
|
||
system=_SYSTEM,
|
||
max_tokens=1200,
|
||
requires_premium=False,
|
||
)
|
||
|
||
|
||
def _parse_json(raw: str) -> dict:
|
||
import re
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
m = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", raw)
|
||
if m:
|
||
try:
|
||
return json.loads(m.group(1))
|
||
except json.JSONDecodeError:
|
||
pass
|
||
m = re.search(r"\{[\s\S]+\}", raw)
|
||
if m:
|
||
try:
|
||
return json.loads(m.group(0))
|
||
except json.JSONDecodeError:
|
||
pass
|
||
raise ValueError(f"Kein JSON in Antwort: {raw[:200]}")
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /api/social/content — alle Einträge
|
||
# ------------------------------------------------------------------
|
||
@router.get("/content")
|
||
async def list_content(
|
||
status: Optional[str] = None,
|
||
user=Depends(require_social_media),
|
||
):
|
||
with db() as conn:
|
||
if status:
|
||
rows = conn.execute(
|
||
"SELECT * FROM social_content WHERE status=? ORDER BY created_at DESC",
|
||
(status,),
|
||
).fetchall()
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT * FROM social_content ORDER BY created_at DESC LIMIT 200",
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /api/social/generate — KI-Content generieren
|
||
# ------------------------------------------------------------------
|
||
@router.post("/generate")
|
||
async def generate_content(req: GenerateRequest, user=Depends(require_social_media)):
|
||
if req.platform not in _PLATFORMS:
|
||
raise HTTPException(400, f"Ungültige Plattform: {req.platform}")
|
||
if req.format not in _FORMATS:
|
||
raise HTTPException(400, f"Ungültiges Format: {req.format}")
|
||
if not req.topic.strip():
|
||
raise HTTPException(400, "Thema darf nicht leer sein.")
|
||
|
||
script_field = _SCRIPT_FIELD_REEL if req.format == "reel" else _SCRIPT_FIELD_OTHER
|
||
prompt = _PROMPT_GENERATE.format(
|
||
platform=req.platform,
|
||
format=req.format,
|
||
topic=req.topic,
|
||
breed_info=_breed_info(req.breed_id),
|
||
used_topics=_used_topics(),
|
||
script_field=script_field,
|
||
)
|
||
|
||
try:
|
||
raw = await _ki_complete(prompt)
|
||
data = _parse_json(raw)
|
||
except Exception as e:
|
||
logger.error("Social-Media-Generierung fehlgeschlagen: %s", e)
|
||
raise HTTPException(500, f"KI-Fehler: {e}")
|
||
|
||
with db() as conn:
|
||
cur = conn.execute(
|
||
"""INSERT INTO social_content
|
||
(created_by, platform, format, topic, caption, hashtags,
|
||
visual_brief, image_prompt, canva_notes, script, hook, cta,
|
||
unsplash_query, ai_score, source, breed_id)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(
|
||
user["id"], req.platform, req.format, req.topic,
|
||
data.get("caption"), data.get("hashtags"),
|
||
data.get("visual_brief"), data.get("image_prompt"),
|
||
data.get("canva_notes"), data.get("script"),
|
||
data.get("hook"), data.get("cta"),
|
||
data.get("unsplash_query"), data.get("ai_score"),
|
||
"generated", req.breed_id,
|
||
),
|
||
)
|
||
entry_id = cur.lastrowid
|
||
|
||
with db() as conn:
|
||
row = conn.execute(
|
||
"SELECT * FROM social_content WHERE id=?", (entry_id,)
|
||
).fetchone()
|
||
result = dict(row)
|
||
result["feedback"] = data.get("feedback")
|
||
return result
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /api/social/evaluate — eigenen Entwurf bewerten + verbessern
|
||
# ------------------------------------------------------------------
|
||
@router.post("/evaluate")
|
||
async def evaluate_content(req: EvaluateRequest, user=Depends(require_social_media)):
|
||
if not req.draft.strip():
|
||
raise HTTPException(400, "Entwurf darf nicht leer sein.")
|
||
|
||
prompt = _PROMPT_EVALUATE.format(
|
||
platform=req.platform,
|
||
format=req.format,
|
||
draft=req.draft,
|
||
)
|
||
|
||
try:
|
||
raw = await _ki_complete(prompt)
|
||
data = _parse_json(raw)
|
||
except Exception as e:
|
||
raise HTTPException(500, f"KI-Fehler: {e}")
|
||
|
||
with db() as conn:
|
||
cur = conn.execute(
|
||
"""INSERT INTO social_content
|
||
(created_by, platform, format, topic, caption, hashtags,
|
||
visual_brief, image_prompt, canva_notes, script, hook, cta,
|
||
unsplash_query, ai_score, source, notes)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(
|
||
user["id"], req.platform, req.format,
|
||
req.draft[:80] + ("…" if len(req.draft) > 80 else ""),
|
||
data.get("caption"), data.get("hashtags"),
|
||
data.get("visual_brief"), data.get("image_prompt"),
|
||
data.get("canva_notes"), data.get("script"),
|
||
data.get("hook"), data.get("cta"),
|
||
data.get("unsplash_query"), data.get("ai_score"),
|
||
"user", data.get("feedback"),
|
||
),
|
||
)
|
||
entry_id = cur.lastrowid
|
||
|
||
with db() as conn:
|
||
row = conn.execute(
|
||
"SELECT * FROM social_content WHERE id=?", (entry_id,)
|
||
).fetchone()
|
||
result = dict(row)
|
||
result["feedback"] = data.get("feedback")
|
||
return result
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# PATCH /api/social/content/{id} — Status / Notizen aktualisieren
|
||
# ------------------------------------------------------------------
|
||
@router.patch("/content/{cid}")
|
||
async def update_content(cid: int, data: StatusUpdate, user=Depends(require_social_media)):
|
||
updates = {k: v for k, v in data.model_dump().items() if v is not None}
|
||
if not updates:
|
||
raise HTTPException(400, "Keine Felder zum Aktualisieren.")
|
||
if "status" in updates and updates["status"] not in _STATUSES:
|
||
raise HTTPException(400, f"Ungültiger Status: {updates['status']}")
|
||
|
||
cols = ", ".join(f"{k}=?" for k in updates)
|
||
values = list(updates.values()) + [cid]
|
||
with db() as conn:
|
||
conn.execute(f"UPDATE social_content SET {cols} WHERE id=?", values)
|
||
return {"ok": True}
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# DELETE /api/social/content/{id}
|
||
# ------------------------------------------------------------------
|
||
@router.delete("/content/{cid}", status_code=204)
|
||
async def delete_content(cid: int, user=Depends(require_social_media)):
|
||
with db() as conn:
|
||
conn.execute("DELETE FROM social_content WHERE id=?", (cid,))
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /api/social/breeds — Rassen für Dropdown
|
||
# ------------------------------------------------------------------
|
||
@router.get("/breeds")
|
||
async def social_breeds(user=Depends(require_social_media)):
|
||
with db() as conn:
|
||
rows = conn.execute(
|
||
"SELECT id, name FROM wiki_rassen WHERE ki_enriched=1 ORDER BY name ASC"
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|