448 lines
16 KiB
Python
448 lines
16 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, UploadFile, File
|
||
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 Luna, der freundliche Social-Media-Coach von Ban Yaro (banyaro.app). "
|
||
"Du begleitest eine junge Content-Creatorin (15 Jahre) dabei, für die Hunde-App "
|
||
"auf TikTok und Instagram zu wachsen. "
|
||
"Erkläre immer kurz WARUM etwas funktioniert — sie soll lernen, nicht nur kopieren. "
|
||
"Ton: locker, ermutigend, auf Augenhöhe — nie von oben herab. "
|
||
"Zielgruppe der App: Hundebesitzer in DACH, 20–45 Jahre, Fokus Frauen. "
|
||
"Antworte immer auf Deutsch."
|
||
)
|
||
|
||
_PROMPT_SUGGESTIONS = '''\
|
||
Schlage 5 aktuelle, kreative Content-Ideen für Ban Yaro (Hunde-App) vor.
|
||
Heute: {datum} ({saison}).
|
||
Bereits verwendete Themen (nicht wiederholen):
|
||
{used_topics}
|
||
|
||
Antworte NUR als JSON-Array:
|
||
[
|
||
{{
|
||
"thema": "Konkretes Thema für einen Post (1 Satz)",
|
||
"warum": "Kurze Erklärung warum das gerade gut funktioniert (1-2 Sätze, als Coach)",
|
||
"format": "reel|post|story|carousel",
|
||
"platform": "tiktok|instagram|both",
|
||
"emoji": "1 passendes Emoji"
|
||
}}
|
||
]
|
||
Misch Formate und Plattformen. Aktuelle Trends, Saisonales und Evergreen-Themen kombinieren.
|
||
'''
|
||
|
||
_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>,
|
||
"coaching": "2-3 Sätze für die Creatorin: Warum dieser Hook? Warum diese Hashtags? Was macht diesen Post stark? Locker und ermutigend formuliert."
|
||
}}
|
||
'''
|
||
|
||
_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/suggestions — KI schlägt 5 Themen vor
|
||
# ------------------------------------------------------------------
|
||
@router.get("/suggestions")
|
||
async def get_suggestions(user=Depends(require_social_media)):
|
||
from datetime import date
|
||
import calendar
|
||
|
||
today = date.today()
|
||
month = today.month
|
||
saison = (
|
||
"Frühling" if month in (3,4,5) else
|
||
"Sommer" if month in (6,7,8) else
|
||
"Herbst" if month in (9,10,11) else "Winter"
|
||
)
|
||
prompt = _PROMPT_SUGGESTIONS.format(
|
||
datum=today.strftime("%d.%m.%Y"),
|
||
saison=saison,
|
||
used_topics=_used_topics(20),
|
||
)
|
||
try:
|
||
raw = await _ki_complete(prompt)
|
||
import re
|
||
m = re.search(r"\[[\s\S]+\]", raw)
|
||
data = json.loads(m.group(0)) if m else []
|
||
return data
|
||
except Exception as e:
|
||
logger.error("Suggestions-Fehler: %s", e)
|
||
raise HTTPException(500, f"KI-Fehler: {e}")
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 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, coaching)
|
||
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, data.get("coaching"),
|
||
),
|
||
)
|
||
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]
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /api/social/stats — Level / XP
|
||
# ------------------------------------------------------------------
|
||
_LEVELS = [
|
||
(0, 50, "🌱 Rookie", "Anfängerin"),
|
||
(50, 150, "✏️ Creator", "Content Creatorin"),
|
||
(150, 400, "🌟 Influencer", "Influencerin"),
|
||
(400, 800, "🏆 Pro Creator", "Profi"),
|
||
(800, 99999,"👑 Star", "Social Media Star"),
|
||
]
|
||
|
||
@router.get("/stats")
|
||
async def get_stats(user=Depends(require_social_media)):
|
||
with db() as conn:
|
||
generated = conn.execute(
|
||
"SELECT COUNT(*) FROM social_content WHERE source='generated'"
|
||
).fetchone()[0]
|
||
published = conn.execute(
|
||
"SELECT COUNT(*) FROM social_content WHERE status='published'"
|
||
).fetchone()[0]
|
||
evaluated = conn.execute(
|
||
"SELECT COUNT(*) FROM social_content WHERE source='user'"
|
||
).fetchone()[0]
|
||
five_star = conn.execute(
|
||
"SELECT COUNT(*) FROM social_content WHERE ai_score=5"
|
||
).fetchone()[0]
|
||
|
||
xp = generated * 10 + published * 20 + evaluated * 5 + five_star * 15
|
||
lvl = _LEVELS[0]
|
||
for l in _LEVELS:
|
||
if xp >= l[0]:
|
||
lvl = l
|
||
idx = _LEVELS.index(lvl)
|
||
nxt = _LEVELS[idx + 1] if idx < len(_LEVELS) - 1 else None
|
||
|
||
return {
|
||
"xp": xp, "generated": generated, "published": published,
|
||
"evaluated": evaluated, "five_star": five_star,
|
||
"level": lvl[2], "level_desc": lvl[3],
|
||
"xp_current_min": lvl[0], "xp_next": nxt[0] if nxt else xp,
|
||
"next_level": nxt[2] if nxt else None,
|
||
}
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /api/social/media — Medien-Upload (Foto/Video)
|
||
# ------------------------------------------------------------------
|
||
@router.post("/media")
|
||
async def upload_media(
|
||
file: UploadFile = File(...),
|
||
user=Depends(require_social_media),
|
||
):
|
||
import shutil, uuid, os
|
||
|
||
MEDIA_DIR = os.getenv("MEDIA_DIR", "/data/media")
|
||
social_dir = os.path.join(MEDIA_DIR, "social")
|
||
os.makedirs(social_dir, exist_ok=True)
|
||
|
||
ext = os.path.splitext(file.filename or "")[-1].lower() or ".jpg"
|
||
fname = f"{uuid.uuid4().hex}{ext}"
|
||
path = os.path.join(social_dir, fname)
|
||
|
||
with open(path, "wb") as f:
|
||
shutil.copyfileobj(file.file, f)
|
||
|
||
return {"url": f"/media/social/{fname}", "filename": fname}
|