import os import logging import datetime from typing import List, Dict, Any # --- MOCK IMPORTS FOR AIVA'S INTERNAL SYSTEMS --- # In a real AIVA system, these would be actual modules managing configuration, # skill graphs, and knowledge base updates. class MockAIVAConfig: YOUTUBE_API_KEY = os.environ.get("AIVA_YOUTUBE_API_KEY", "MOCK_YOUTUBE_API_KEY") # Example: A list of YouTube channels or specific video IDs AIVA should monitor MONITORED_YOUTUBE_SOURCES = [ {"type": "channel", "id": "UC-lHJZR3Gqxm24_Vd_AJ5Yw"}, # Example channel ID {"type": "video", "id": "dQw4w9WgXcQ"} # Example video ID (Rick Astley - Never Gonna Give You Up) ] # Define a threshold for how far back to look for new videos (e.g., 24 hours) MONITORING_LOOKBACK_HOURS = 24 class MockAIVASkillsGraph: _skills_database = {} _insights_database = {} @classmethod def add_skill(cls, skill_name: str, source_url: str, confidence: float = 1.0, context: str = ""): if skill_name not in cls._skills_database: cls._skills_database[skill_name] = [] cls._skills_database[skill_name].append({ "source": source_url, "timestamp": datetime.datetime.utcnow().isoformat(), "confidence": confidence, "context": context }) logging.info(f"[AIVA Skills Graph] Added skill: '{skill_name}' from {source_url}") @classmethod def add_insight(cls, insight_text: str, source_url: str, confidence: float = 1.0, context: str = ""): if insight_text not in cls._insights_database: cls._insights_database[insight_text] = [] cls._insights_database[insight_text].append({ "source": source_url, "timestamp": datetime.datetime.utcnow().isoformat(), "confidence": confidence, "context": context }) logging.info(f"[AIVA Skills Graph] Added insight: '{insight_text}' from {source_url}") @classmethod def get_all_skills(cls): return cls._skills_database @classmethod def get_all_insights(cls): return cls._insights_database # --- MOCK EXTERNAL LIBRARIES --- # In a real system, you would install and import 'youtube-transcript-api' # and potentially a more sophisticated NLP library like spaCy or NLTK. class MockYouTubeTranscriptApi: @staticmethod def get_transcript(video_id: str, languages: List[str] = ['en']) -> List[Dict[str, Any]]: # Simulate fetching a transcript mock_transcripts = { "dQw4w9WgXcQ": [ {"text": "We're no strangers to love", "start": 0.0, "duration": 3.0}, {"text": "You know the rules and so do I", "start": 3.5, "duration": 3.0}, {"text": "A full commitment's what I'm thinking of", "start": 7.0, "duration": 4.0}, {"text": "You wouldn't get this from any other guy", "start": 11.5, "duration": 4.0}, {"text": "I just wanna tell you how I'm feeling", "start": 16.0, "duration": 3.0}, {"text": "Gotta make you understand", "start": 19.5, "duration": 2.0}, {"text": "Never gonna give you up", "start": 22.0, "duration": 2.0}, {"text": "Never gonna let you down", "start": 24.5, "duration": 2.0}, {"text": "Never gonna run around and desert you", "start": 27.0, "duration": 3.0}, {"text": "Never gonna make you cry", "start": 30.5, "duration": 2.0}, {"text": "Never gonna say goodbye", "start": 33.0, "duration": 2.0}, {"text": "Never gonna tell a lie and hurt you", "start": 35.5, "duration": 3.0} ], "example_tech_video_id": [ {"text": "Today we discuss advanced Python programming techniques.", "start": 0.0, "duration": 5.0}, {"text": "Machine learning models require careful data preprocessing.", "start": 6.0, "duration": 6.0}, {"text": "Understanding neural networks is crucial for AI development.", "start": 13.0, "duration": 7.0}, {"text": "The future of quantum computing presents new challenges.", "start": 21.0, "duration": 6.0}, {"text": "Leadership in technology often involves agile methodologies.", "start": 28.0, "duration": 7.0}, {"text": "Key takeaway: continuous learning is vital for innovation.", "start": 36.0, "duration": 7.0} ] } if video_id in mock_transcripts: logging.info(f"[Mock Transcript API] Retrieved transcript for {video_id}") return mock_transcripts[video_id] else: logging.warning(f"[Mock Transcript API] No transcript found for {video_id}. Simulating a generic one.") return [ {"text": "This video contains valuable information about skill development and future trends.", "start": 0.0, "duration": 10.0}, {"text": "Problem-solving strategies are discussed alongside new approaches to data analysis.", "start": 11.0, "duration": 12.0}, {"text": "A critical challenge in AI is ethical deployment, requiring careful consideration.", "start": 24.0, "duration": 13.0} ] class MockYouTubeDataAPI: def __init__(self, api_key: str): self.api_key = api_key def get_channel_uploads(self, channel_id: str, published_after: datetime.datetime) -> List[Dict[str, Any]]: # Simulate fetching recent videos from a channel logging.info(f"[Mock YouTube Data API] Fetching uploads for channel {channel_id} after {published_after}") if channel_id == "UC-lHJZR3Gqxm24_Vd_AJ5Yw": # Example channel # Return a mock video that would be new return [ { "id": "example_tech_video_id", "title": "Advanced Python & AI Development", "published_at": (datetime.datetime.utcnow() - datetime.timedelta(hours=12)).isoformat() + "Z", "url": "https://www.youtube.com/watch?v=example_tech_video_id" } ] return [] # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class YouTubeIntelligencePipeline: SKILL_KEYWORDS = { "programming": ["python", "java", "c++", "coding", "development", "software engineering", "algorithm"], "data analysis": ["data science", "analytics", "machine learning", "ai", "deep learning", "statistics", "model", "neural network"], "design": ["ui/ux", "graphic design", "product design", "figma", "adobe xd", "user experience"], "leadership": ["management", "team building", "strategy", "decision making", "agile", "scrum"], "communication": ["presentation", "public speaking", "negotiation", "storytelling"], "finance": ["investing", "stocks", "economics", "budgeting", "market"], "quantum computing": ["quantum computing", "qubit", "quantum physics", "quantum algorithm"] } INSIGHT_TRIGGERS = [ "future of", "trend in", "breakthrough", "paradigm shift", "new approach", "critical challenge", "key takeaway", "implications for", "innovation", "ethical deployment" ] def __init__(self, config=MockAIVAConfig, skills_graph=MockAIVASkillsGraph, youtube_transcript_api=MockYouTubeTranscriptApi, youtube_data_api_client=None): self.config = config self.skills_graph = skills_graph self.transcript_api = youtube_transcript_api self.youtube_data_api = youtube_data_api_client or MockYouTubeDataAPI(self.config.YOUTUBE_API_KEY) self.logger = logging.getLogger(self.__class__.__name__) def _get_video_transcript(self, video_id: str) -> str: """Fetches the full text transcript for a given YouTube video ID.""" try: transcript_list = self.transcript_api.get_transcript(video_id) full_transcript = " ".join([entry['text'] for entry in transcript_list]) self.logger.info(f"Successfully retrieved transcript for video ID: {video_id}") return full_transcript except Exception as e: self.logger.error(f"Failed to get transcript for video ID {video_id}: {e}") return "" def _extract_skills_and_insights(self, text: str) -> Dict[str, List[str]]: """Extracts skills and key insights from the given text using keyword matching.""" extracted_skills = set() extracted_insights = set() text_lower = text.lower() # Skill extraction for skill_category, keywords in self.SKILL_KEYWORDS.items(): for keyword in keywords: if keyword in text_lower: extracted_skills.add(skill_category) # Optionally, add the specific keyword as a sub-skill or context # extracted_skills.add(keyword) # if we want more granular skills # Insight extraction for trigger in self.INSIGHT_TRIGGERS: if trigger in text_lower: # For insights, we'll try to capture a surrounding phrase # This is a very basic implementation; a real NLP would use sentence parsing start_index = text_lower.find(trigger) if start_index != -1: # Capture a snippet around the trigger snippet_start = max(0, start_index - 50) snippet_end = min(len(text_lower), start_index + len(trigger) + 100) snippet = text[snippet_start:snippet_end].strip() # Basic cleaning: remove excessive whitespace, trim snippet = ' '.join(snippet.split()) extracted_insights.add(f"...{snippet}...") self.logger.info(f"Extracted {len(extracted_skills)} skills and {len(extracted_insights)} insights.") return {"skills": list(extracted_skills), "insights": list(extracted_insights)} def _process_video(self, video_id: str, video_url: str): """Processes a single video: fetches transcript, extracts, and updates AIVA.""" self.logger.info(f"Processing video: {video_url}") transcript_text = self._get_video_transcript(video_id) if not transcript_text: self.logger.warning(f"Skipping video {video_id} due to empty transcript.") return extracted_data = self._extract_skills_and_insights(transcript_text) for skill in extracted_data["skills"]: self.skills_graph.add_skill(skill, source_url=video_url, context=transcript_text[:200]) # Add first 200 chars as context for insight in extracted_data["insights"]: self.skills_graph.add_insight(insight, source_url=video_url, context=transcript_text[:200]) # Add first 200 chars as context self.logger.info(f"Completed processing for video: {video_url}") def run_daily_pipeline(self): """Main function to run the daily YouTube intelligence pipeline.""" self.logger.info("Starting YouTube Intelligence Pipeline for AIVA.") processed_count = 0 new_skills_count = 0 new_insights_count = 0 # Calculate the 'published_after' timestamp for monitoring new content published_after = datetime.datetime.utcnow() - datetime.timedelta(hours=self.config.MONITORING_LOOKBACK_HOURS) videos_to_process = [] for source in self.config.MONITORED_YOUTUBE_SOURCES: if source["type"] == "channel": # Use YouTube Data API to fetch recent videos from the channel recent_channel_videos = self.youtube_data_api.get_channel_uploads(source["id"], published_after) for video in recent_channel_videos: videos_to_process.append({ "id": video["id"], "url": video["url"] }) elif source["type"] == "video": # For specific video IDs, we just add them directly # In a real scenario, you might check if already processed or if content changed videos_to_process.append({ "id": source["id"], "url": f"https://www.youtube.com/watch?v={source["id"]}" }) if not videos_to_process: self.logger.info("No new YouTube content to process today.") return for video_info in videos_to_process: video_id = video_info["id"] video_url = video_info["url"] initial_skills_count = len(self.skills_graph.get_all_skills()) initial_insights_count = len(self.skills_graph.get_all_insights()) self._process_video(video_id, video_url) processed_count += 1 new_skills_count += (len(self.skills_graph.get_all_skills()) - initial_skills_count) new_insights_count += (len(self.skills_graph.get_all_insights()) - initial_insights_count) self.logger.info(f"YouTube Intelligence Pipeline finished. Processed {processed_count} videos.") self.logger.info(f"Added {new_skills_count} new skills and {new_insights_count} new insights to AIVA's knowledge.") self.logger.info("AIVA's Skills Graph (current snapshot):\n" + f"Skills: {list(self.skills_graph.get_all_skills().keys())}\n" + f"Insights: {list(self.skills_graph.get_all_insights().keys())}") self.logger.info("YouTube Intelligence Pipeline successfully completed for Queen AIVA.") # Example of how to run the pipeline (for testing/demonstration) if __name__ == "__main__": # Ensure the mock YouTube API key is set or provide a real one if testing live os.environ["AIVA_YOUTUBE_API_KEY"] = "YOUR_YOUTUBE_API_KEY_HERE" # Replace with a real key if you want to test with real data # Instantiate and run the pipeline pipeline = YouTubeIntelligencePipeline() pipeline.run_daily_pipeline() print("\n--- AIVA's Current Skills & Insights ---") print("Skills:") for skill, sources in MockAIVASkillsGraph.get_all_skills().items(): print(f" - {skill} (from {len(sources)} sources)") print("\nInsights:") for insight, sources in MockAIVASkillsGraph.get_all_insights().items(): print(f" - {insight} (from {len(sources)} sources)")