""" FomoFeed - Notification Generator AI Generates personalized push notification texts using Turkish GPT-2 """ from fastapi import FastAPI, HTTPException from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForCausalLM import torch import random from datetime import datetime import uvicorn app = FastAPI(title="FomoFeed Notification Generator", version="1.0.0") # Load Turkish GPT-2 model (lightweight) MODEL_NAME = "ytu-ce-cosmos/turkish-gpt2" tokenizer = None model = None def load_model(): global tokenizer, model try: tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) model = AutoModelForCausalLM.from_pretrained(MODEL_NAME) model.eval() print(f"✅ Model loaded: {MODEL_NAME}") except Exception as e: print(f"⚠️ Model load failed: {e}") print("Using template-based fallback") # Load model at startup load_model() class UserProfile(BaseModel): user_id: int username: str is_creator: bool = False top_tags: list[str] = [] last_post_days_ago: int = 0 avg_engagement: float = 0.0 follower_count: int = 0 optimal_hour: int = 19 class NotificationRequest(BaseModel): profile: UserProfile notification_type: str # "optimal_time", "trending_tag", "low_activity", "milestone" language: str = "tr" class NotificationResponse(BaseModel): title: str body: str emoji: str confidence: float # Template-based fallback (when AI model unavailable) TEMPLATES_TR = { "optimal_time_creator": [ "✨ En iyi paylaşım zamanın! {hour}:00'da takipçilerin en aktif. İçerik paylaşmaya ne dersin?", "🎯 Şu an içerik paylaşmak için ideal zaman! Takipçilerin seni bekliyor.", "⚡ Heyecan verici bir şey mi paylaşacaksın? Tam zamanı! {hour}:00 civarı en aktif saatiniz.", ], "optimal_time_user": [ "🔥 Yeni içerikler seni bekliyor! Keşfet sayfasında neler var bakalım?", "✨ Takip ettiğin kişiler yeni paylaşımlar yaptı. Göz atmaya ne dersin?", "📱 Feed'in dolup taşıyor! Kaçırdığın harika içerikler var.", ], "trending_tag_creator": [ "🔥 #{tag} trende! Bu konuda paylaşım yapma fırsatını kaçırma.", "⚡ Takipçilerin #{tag} hakkında içerik bekliyor. Paylaşmaya hazır mısın?", "✨ #{tag} şu an çok popüler. Senin de fikrini paylaşmanın tam zamanı!", ], "low_activity_creator": [ "🎨 {days} gündür paylaşım yapmadın. Takipçilerin seni özlemiş olabilir!", "💭 Bugün neler yapıyorsun? Takipçilerinle paylaş!", "✨ Bir süredir sessizsin. Yeni bir paylaşımla geri dönmeye ne dersin?", ], "low_activity_user": [ "👋 Seni görmeyeli uzun zaman oldu! Feed'de seni bekleyen harika içerikler var.", "🌟 FomoFeed'de neler döndüğünü merak ediyor musun? Hemen keşfet!", "✨ {days} gündür girmemişsin. Kaçırdığın içerikler var!", ], "milestone_creator": [ "🎉 Tebrikler! {milestone} takipçiye ulaştın! Başarılarının devamını diliyoruz.", "⭐ Harika gidiyorsun! {engagement} etkileşim aldın bu hafta.", "🔥 İçeriklerinin kalitesi göz dolduruyor! Devam et!", ] } TEMPLATES_EN = { "optimal_time_creator": [ "✨ Perfect time to post! Your followers are most active at {hour}:00. Ready to share?", "🎯 Now is the ideal time to post! Your audience is waiting.", "⚡ Got something exciting to share? Perfect timing! Peak activity around {hour}:00.", ], "optimal_time_user": [ "🔥 New content is waiting for you! Check what's new on your feed.", "✨ People you follow posted new content. Want to check it out?", "📱 Your feed is full! Don't miss great content.", ], "trending_tag_creator": [ "🔥 #{tag} is trending! Don't miss the opportunity to post about it.", "⚡ Your followers want content about #{tag}. Ready to share?", "✨ #{tag} is hot right now. Perfect time to share your thoughts!", ], "low_activity_creator": [ "🎨 You haven't posted in {days} days. Your followers might miss you!", "💭 What are you up to today? Share with your followers!", "✨ You've been quiet lately. How about a comeback post?", ], "low_activity_user": [ "👋 Long time no see! Great content is waiting for you on the feed.", "🌟 Curious what's happening on FomoFeed? Discover now!", "✨ You haven't visited in {days} days. You're missing out!", ], "milestone_creator": [ "🎉 Congratulations! You reached {milestone} followers! Keep it up!", "⭐ You're doing great! {engagement} engagements this week.", "🔥 Your content quality is impressive! Keep going!", ] } def generate_with_ai(prompt: str, profile: UserProfile) -> str: """ Generate notification text using Turkish GPT-2 """ if model is None or tokenizer is None: return None try: # Prepare prompt full_prompt = f"Kişiselleştirilmiş bildirim: {prompt}" # Generate inputs = tokenizer(full_prompt, return_tensors="pt", max_length=100, truncation=True) with torch.no_grad(): outputs = model.generate( inputs["input_ids"], max_length=60, num_return_sequences=1, temperature=0.8, do_sample=True, top_p=0.9, pad_token_id=tokenizer.eos_token_id ) # Decode generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True) # Extract notification (remove prompt) notification = generated_text.replace(full_prompt, "").strip() # Validate length (20-80 chars) if 20 <= len(notification) <= 80: return notification return None except Exception as e: print(f"AI generation error: {e}") return None def generate_template_based(request: NotificationRequest) -> dict: """ Fallback template-based generation """ profile = request.profile ntype = request.notification_type lang = request.language templates = TEMPLATES_TR if lang == "tr" else TEMPLATES_EN # Select template category if ntype == "optimal_time": key = "optimal_time_creator" if profile.is_creator else "optimal_time_user" elif ntype == "trending_tag": key = "trending_tag_creator" elif ntype == "low_activity": key = "low_activity_creator" if profile.is_creator else "low_activity_user" elif ntype == "milestone": key = "milestone_creator" else: key = "optimal_time_user" # Get random template template = random.choice(templates.get(key, templates["optimal_time_user"])) # Fill variables tag = profile.top_tags[0] if profile.top_tags else "fomo" body = template.format( hour=profile.optimal_hour, tag=tag, days=profile.last_post_days_ago, milestone=profile.follower_count, engagement=int(profile.avg_engagement * 10) ) # Title based on type if lang == "tr": titles = { "optimal_time": "⏰ İdeal Paylaşım Zamanı", "trending_tag": f"🔥 Trend: #{tag}", "low_activity": "👋 Seni Özledik", "milestone": "🎉 Yeni Başarı" } else: titles = { "optimal_time": "⏰ Perfect Posting Time", "trending_tag": f"🔥 Trending: #{tag}", "low_activity": "👋 We Miss You", "milestone": "🎉 New Milestone" } title = titles.get(ntype, titles["optimal_time"]) # Emoji mapping emoji_map = { "optimal_time": "⏰", "trending_tag": "🔥", "low_activity": "👋", "milestone": "🎉" } return { "title": title, "body": body, "emoji": emoji_map.get(ntype, "📱"), "confidence": 0.7 } @app.get("/") def root(): return { "service": "FomoFeed Notification Generator", "status": "active", "model": "turkish-gpt2" if model else "template-based", "version": "1.0.0" } @app.get("/health") def health(): return { "status": "healthy", "model_loaded": model is not None, "timestamp": datetime.now().isoformat() } @app.post("/generate", response_model=NotificationResponse) def generate_notification(request: NotificationRequest): """ Generate personalized notification text """ try: # Try AI generation first ai_result = None if model and request.language == "tr": prompt = f"Kullanıcı {request.profile.username} için {request.notification_type} bildirimi" ai_result = generate_with_ai(prompt, request.profile) if ai_result: return NotificationResponse( title=f"✨ {request.profile.username}", body=ai_result, emoji="✨", confidence=0.85 ) # Fallback to templates result = generate_template_based(request) return NotificationResponse(**result) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/batch_generate") def batch_generate(requests: list[NotificationRequest]): """ Generate multiple notifications at once """ try: results = [] for req in requests: result = generate_template_based(req) results.append({ "user_id": req.profile.user_id, **result }) return {"notifications": results} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)