""" Event Hashtag Generator - AI Chatbot for automatic hashtag generation Generates viral hashtags, keywords, and target audience insights from event data """ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional, List from datetime import datetime import os import json import re from huggingface_hub import InferenceClient import uvicorn # Initialize FastAPI app = FastAPI( title="Event Hashtag Generator API", description="AI-powered automatic hashtag and keyword generation for events", version="2.0.0" ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Hugging Face token hf_token = os.getenv("HUGGINGFACE_TOKEN") if hf_token: print("✓ Hugging Face token configured") else: print("⚠ Warning: No HUGGINGFACE_TOKEN found. Set it in environment variable.") # Pydantic models class EventHashtagRequest(BaseModel): event_name: str category: str short_description: str detailed_description: str max_hashtags: Optional[int] = 10 language: Optional[str] = "vi" hf_token: Optional[str] = None class EventHashtagResponse(BaseModel): event_name: str hashtags: List[str] keywords: List[str] target_audience: List[str] confidence_score: float generation_time: str model_used: str @app.get("/") async def root(): """API Information""" return { "status": "running", "service": "Event Hashtag Generator API", "version": "2.0.0", "description": "Generate hashtags, keywords, and target audience from event info", "endpoints": { "POST /generate-hashtags": { "description": "Generate viral hashtags for events", "request_body": { "event_name": "string - Tên sự kiện", "category": "string - Danh mục (âm nhạc, thể thao, công nghệ...)", "short_description": "string - Mô tả ngắn (1-2 câu)", "detailed_description": "string - Mô tả chi tiết", "max_hashtags": "integer (optional, default: 10)", "language": "string (optional, default: 'vi')", "hf_token": "string (optional)" } } } } def build_hashtag_prompt(event_name: str, category: str, short_desc: str, detailed_desc: str, max_hashtags: int, language: str) -> str: """Prompt chỉ tập trung vào hashtag, keywords và audience.""" lang_instruction = "tiếng Việt" if language == "vi" else "English" prompt = f"""Phân tích sự kiện sau và tạo ra các hashtag lan truyền mạnh mẽ, cùng với từ khóa và đối tượng mục tiêu. SỰ KIỆN: Tên: {event_name} Danh mục: {category} Mô tả ngắn: {short_desc} Mô tả chi tiết: {detailed_desc} YÊU CẦU: - Tạo tối đa {max_hashtags} hashtag độc đáo, dễ nhớ, dễ viral, liên quan đến sự kiện. - Mỗi hashtag phải bắt đầu bằng #. - Ngôn ngữ: {lang_instruction}. - Cung cấp thêm: - Danh sách từ khóa (keywords) liên quan đến sự kiện. - Danh sách đối tượng khán giả mục tiêu (target audience) phù hợp. - Không trả lời giải thích, chỉ xuất JSON. JSON OUTPUT: {{ "hashtags": ["#TênSựKiện", "#Hashtag2", "#Hashtag3"], "keywords": ["keyword1", "keyword2"], "target_audience": ["đối tượng 1", "đối tượng 2"] }} CHỈ TRẢ VỀ JSON, KHÔNG THÊM TEXT KHÁC. """ return prompt def parse_llm_response(response_text: str) -> dict: """Parse JSON từ model trả về.""" result = {"hashtags": [], "keywords": [], "target_audience": []} try: json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: data = json.loads(json_match.group(0)) result["hashtags"] = data.get("hashtags", []) result["keywords"] = data.get("keywords", []) result["target_audience"] = data.get("target_audience", []) print("✓ Parsed JSON successfully") else: print("⚠ No valid JSON found") except Exception as e: print(f"✗ Parsing error: {str(e)}") return result @app.post("/generate-hashtags", response_model=EventHashtagResponse) async def generate_hashtags(request: EventHashtagRequest): """Generate viral hashtags, keywords, and target audience for an event.""" try: start_time = datetime.utcnow() token = request.hf_token or hf_token if not token: raise HTTPException(status_code=401, detail="HUGGINGFACE_TOKEN required.") prompt = build_hashtag_prompt( request.event_name, request.category, request.short_description, request.detailed_description, request.max_hashtags, request.language ) client = InferenceClient(token=token) models_to_try = [ "mistralai/Mistral-7B-Instruct-v0.3", "microsoft/Phi-3-mini-4k-instruct", "meta-llama/Meta-Llama-3-8B-Instruct" ] llm_response = "" model_used = "" for model in models_to_try: try: print(f"Trying model: {model}") response = client.chat_completion( model=model, messages=[{"role": "user", "content": prompt}], max_tokens=800, temperature=0.6, ) llm_response = response.choices[0].message.content if llm_response and len(llm_response) > 20: model_used = model break except Exception as e: print(f"✗ Failed with {model}: {e}") continue if not llm_response: raise HTTPException(status_code=500, detail="All models failed to respond.") parsed = parse_llm_response(llm_response) # Fallback nếu model không trả được hashtag if not parsed["hashtags"]: print("⚠ Creating fallback hashtags") base = re.sub(r'[^a-zA-Z0-9 ]', '', request.event_name) words = base.split() parsed["hashtags"] = [f"#{w.capitalize()}" for w in words[:request.max_hashtags]] # Tính confidence đơn giản confidence = 0.3 * bool(parsed["hashtags"]) + 0.3 * bool(parsed["keywords"]) + 0.4 * bool(parsed["target_audience"]) end_time = datetime.utcnow() return EventHashtagResponse( event_name=request.event_name, hashtags=parsed["hashtags"][:request.max_hashtags], keywords=parsed["keywords"], target_audience=parsed["target_audience"], confidence_score=round(confidence, 2), generation_time=f"{(end_time - start_time).total_seconds():.2f}s", model_used=model_used.split("/")[-1], ) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") if __name__ == "__main__": uvicorn.run( "app:app", host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), reload=False, log_level="info", )