""" Hype Generator Service for OpenTriage. Generates shareable social media posts for milestones and achievements. """ import logging from typing import Dict, Any, Optional, List from datetime import datetime, timezone from pydantic import BaseModel from config.settings import settings logger = logging.getLogger(__name__) class Milestone(BaseModel): """A milestone or achievement to celebrate.""" milestone_type: str # pr_merged, first_contribution, streak, trophy, etc. user_id: str username: str repo_name: Optional[str] = None value: Optional[int] = None # e.g., number of PRs description: str = "" trophy_name: Optional[str] = None achieved_at: datetime = None class HypePost(BaseModel): """A generated social media post.""" platform: str # linkedin, twitter/x content: str hashtags: List[str] = [] mentioned_users: List[str] = [] stats_included: Dict[str, Any] = {} generated_at: datetime = None class HypeGeneratorService: """ Service for generating celebratory social media posts. Features: - LinkedIn post generation - Twitter/X post generation - Stats-based content - Hashtag suggestions """ def __init__(self): self.linkedin_char_limit = 3000 self.twitter_char_limit = 280 self.default_hashtags = { "linkedin": ["#OpenSource", "#Developer", "#GitHub", "#Coding", "#TechCommunity"], "twitter": ["#OpenSource", "#100DaysOfCode", "#GitHub", "#DevCommunity"] } async def generate_linkedin_post(self, milestone: Milestone) -> HypePost: """ Generate a LinkedIn-style celebratory post. Args: milestone: The milestone to celebrate Returns: HypePost for LinkedIn """ content = await self._generate_linkedin_content(milestone) hashtags = self._get_hashtags_for_milestone(milestone, "linkedin") return HypePost( platform="linkedin", content=content, hashtags=hashtags, mentioned_users=[milestone.username], stats_included=await self._get_user_stats(milestone.username), generated_at=datetime.now(timezone.utc) ) async def generate_twitter_post(self, milestone: Milestone) -> HypePost: """ Generate a Twitter/X celebration post. Args: milestone: The milestone to celebrate Returns: HypePost for Twitter """ content = await self._generate_twitter_content(milestone) hashtags = self._get_hashtags_for_milestone(milestone, "twitter") return HypePost( platform="twitter", content=content, hashtags=hashtags, mentioned_users=[milestone.username], stats_included={}, generated_at=datetime.now(timezone.utc) ) async def _generate_linkedin_content(self, milestone: Milestone) -> str: """Generate LinkedIn post content.""" try: from openai import OpenAI client = OpenAI( base_url="https://openrouter.ai/api/v1", api_key=settings.OPENROUTER_API_KEY ) stats = await self._get_user_stats(milestone.username) prompt = f"""Generate a celebratory LinkedIn post for an open source contribution milestone. Make it professional yet engaging. Milestone Type: {milestone.milestone_type} Username: @{milestone.username} Repository: {milestone.repo_name or 'Various projects'} Description: {milestone.description} Value: {milestone.value} User Stats: - Total PRs: {stats.get('total_prs', 0)} - Total Contributions: {stats.get('total_contributions', 0)} - Current Streak: {stats.get('current_streak', 0)} days Guidelines: - Start with an emoji and celebratory opener - Mention the specific achievement - Include 2-3 sentences about the journey/growth - End with encouragement for others - Keep under 300 words - Don't include hashtags in the text Do NOT use placeholder brackets like [username] - use the actual values provided.""" response = client.chat.completions.create( model="cohere/rerank-4-pro", messages=[ {"role": "system", "content": "You are a social media content creator specializing in tech and open source."}, {"role": "user", "content": prompt} ], max_tokens=400, temperature=0.8 ) return response.choices[0].message.content.strip() except Exception as e: logger.error(f"AI generation failed: {e}") return self._get_fallback_linkedin(milestone) async def _generate_twitter_content(self, milestone: Milestone) -> str: """Generate Twitter/X post content.""" try: from openai import OpenAI client = OpenAI( base_url="https://openrouter.ai/api/v1", api_key=settings.OPENROUTER_API_KEY ) prompt = f"""Generate a celebratory tweet for an open source achievement. Milestone: {milestone.milestone_type} Username: @{milestone.username} Repo: {milestone.repo_name or 'open source'} Achievement: {milestone.description} Requirements: - Must be under 200 characters (leave room for hashtags) - Start with emoji - Be genuine and exciting - No hashtags in the text - Use actual values, not placeholders""" response = client.chat.completions.create( model="cohere/rerank-4-pro", messages=[ {"role": "user", "content": prompt} ], max_tokens=100, temperature=0.9 ) content = response.choices[0].message.content.strip() # Ensure it fits Twitter limit if len(content) > 200: content = content[:197] + "..." return content except Exception as e: logger.error(f"AI generation failed: {e}") return self._get_fallback_twitter(milestone) def _get_fallback_linkedin(self, milestone: Milestone) -> str: """Fallback LinkedIn content.""" templates = { "first_pr": f"""๐ŸŽ‰ Celebrating a special milestone! I just merged my first Pull Request on {milestone.repo_name or 'an open source project'}! It may seem like a small step, but every contribution to open source starts somewhere. The journey of a thousand commits begins with a single PR! To anyone thinking about contributing to open source: take that leap! The community is welcoming, and every bit counts. Special thanks to the maintainers for their guidance and patience. ๐Ÿ™ Here's to many more contributions! ๐Ÿ’ช""", "streak": f"""๐Ÿ”ฅ {milestone.value}-Day Contribution Streak! Consistency is key, and I'm proud to share that I've maintained a {milestone.value}-day contribution streak on GitHub! Whether it's code, documentation, or reviews - showing up every day has been transformative. Small daily actions lead to big results. What habits are you building? Drop them below! ๐Ÿ‘‡""", "default": f"""๐ŸŽŠ Milestone Unlocked! Excited to share: {milestone.description} Open source has been an incredible journey of learning, collaboration, and growth. Every contribution, no matter how small, makes a difference. Grateful for this amazing community! ๐ŸŒŸ""" } return templates.get(milestone.milestone_type, templates["default"]) def _get_fallback_twitter(self, milestone: Milestone) -> str: """Fallback Twitter content.""" templates = { "first_pr": f"๐ŸŽ‰ Just merged my first PR to {milestone.repo_name or 'open source'}! The journey begins!", "streak": f"๐Ÿ”ฅ {milestone.value}-day contribution streak! Consistency pays off!", "trophy": f"๐Ÿ† Unlocked: {milestone.trophy_name}! {milestone.description}", "default": f"โœจ {milestone.description}" } return templates.get(milestone.milestone_type, templates["default"]) def _get_hashtags_for_milestone(self, milestone: Milestone, platform: str) -> List[str]: """Get relevant hashtags for the milestone.""" hashtags = self.default_hashtags.get(platform, []).copy() # Add milestone-specific hashtags if milestone.milestone_type == "streak": hashtags.append("#ConsistencyIsKey") elif milestone.milestone_type == "first_pr": hashtags.append("#FirstPR") hashtags.append("#FirstContribution") elif milestone.milestone_type == "trophy": hashtags.append("#Achievement") # Add repo-specific if available if milestone.repo_name: repo_tag = milestone.repo_name.split('/')[-1].replace('-', '') if len(repo_tag) <= 15: hashtags.append(f"#{repo_tag}") return hashtags[:5] # Limit to 5 hashtags async def _get_user_stats(self, username: str) -> Dict[str, Any]: """Get user statistics for content.""" from config.database import db stats = { "total_prs": 0, "total_issues": 0, "total_contributions": 0, "current_streak": 0, "repos_contributed_to": 0 } try: # Count PRs stats["total_prs"] = await db.issues.count_documents({ "authorName": username, "isPR": True }) # Count issues stats["total_issues"] = await db.issues.count_documents({ "authorName": username, "isPR": False }) stats["total_contributions"] = stats["total_prs"] + stats["total_issues"] # Get unique repos pipeline = [ {"$match": {"authorName": username}}, {"$group": {"_id": "$repoName"}}, {"$count": "count"} ] result = await db.issues.aggregate(pipeline).to_list(1) stats["repos_contributed_to"] = result[0]["count"] if result else 0 # Get streak try: from services.gamification_engine import gamification_engine streak = await gamification_engine.get_user_streak(username) stats["current_streak"] = streak.current_streak except Exception: pass except Exception as e: logger.error(f"Error getting user stats: {e}") return stats async def generate_stats_image(self, username: str) -> Optional[bytes]: """ Generate a stats image for sharing. Returns PNG bytes of a stats card. Note: Full implementation would use PIL or similar. This returns a placeholder SVG for now. """ stats = await self._get_user_stats(username) svg = f''' @{username}'s Open Source Journey Pull Requests: {stats['total_prs']} Issues: {stats['total_issues']} Current Streak: {stats['current_streak']} days ๐Ÿ”ฅ Repos: {stats['repos_contributed_to']} Generated by OpenTriage ''' return svg.encode('utf-8') async def generate_impact_summary( self, pr_title: str, pr_body: str, repo_name: str, files_changed: int = 0, additions: int = 0, deletions: int = 0 ) -> str: """ Generate a short, motivating impact summary for a merged PR. Args: pr_title: Title of the merged PR pr_body: Description/body of the PR repo_name: Repository name (owner/repo) files_changed: Number of files changed additions: Lines added deletions: Lines deleted Returns: A short, motivating impact summary string """ try: from openai import OpenAI client = OpenAI( base_url="https://openrouter.ai/api/v1", api_key=settings.OPENROUTER_API_KEY ) prompt = f"""Generate a SHORT, punchy impact summary for a merged pull request. This will be shown in a celebration popup. Make it feel impactful and motivating! PR Title: {pr_title} PR Description: {pr_body[:500] if pr_body else 'No description'} Repository: {repo_name} Stats: {files_changed} files changed, +{additions} additions, -{deletions} deletions Guidelines: - Keep it to ONE sentence, max 100 characters - Start with an emoji - Use specific, quantifiable impact when possible - Be creative with metaphors (e.g., "saved 5 hours of manual work", "improved speed by 2x") - Make the contributor feel like a hero - If unclear what the PR does, estimate impact based on scope Examples: "๐Ÿš€ You just made the app 30% faster for 10,000 users!" "๐Ÿ’ก Your fix will save developers 5 hours of debugging!" "๐Ÿ”’ You patched a security gap protecting 50k users!" "โœจ Your feature unlocks new possibilities for creators!" Generate ONE impact summary:""" response = client.chat.completions.create( model="cohere/rerank-4-pro", messages=[ {"role": "user", "content": prompt} ], max_tokens=60, temperature=0.9 ) summary = response.choices[0].message.content.strip() # Clean up any quotes summary = summary.strip('"\'') # Ensure reasonable length if len(summary) > 150: summary = summary[:147] + "..." return summary except Exception as e: logger.error(f"AI impact summary generation failed: {e}") return self._get_fallback_impact_summary(files_changed, additions, deletions) def _get_fallback_impact_summary( self, files_changed: int = 0, additions: int = 0, deletions: int = 0 ) -> str: """Fallback impact summary when AI fails.""" total_changes = additions + deletions if total_changes > 500: return "๐Ÿš€ You just shipped a massive improvement to the codebase!" elif total_changes > 100: return "๐Ÿ’ช Your contribution makes a real difference!" elif deletions > additions: return "๐Ÿงน You cleaned up the codebase - less is more!" elif files_changed > 5: return "๐ŸŒŸ Your changes touch multiple parts of the project!" else: return "โœจ Every contribution counts - you're making an impact!" # Singleton instance hype_generator_service = HypeGeneratorService()