# poe: name=Study-Notes-Generator-18 # poe: enable_network_access=true import io import json import re import requests import base64 import asyncio import time import concurrent.futures from poepython.model import _state # Configuration LOADING_FRAMES = ["📚", "📖", "📝", "✏️"] # ---------- UI executor ---------- UI_EXECUTOR = concurrent.futures.ThreadPoolExecutor(max_workers=1) def ui_call(state, fn, *args, **kwargs): """ Run any progress_msg.* call in the dedicated UI executor so that the event-loop thread never blocks on HTTPS. Keeps Poe context (_state) inside that thread. """ def _wrapped(): _state.set(state) return fn(*args, **kwargs) loop = asyncio.get_running_loop() return loop.run_in_executor(UI_EXECUTOR, _wrapped) def call_with_state(state, *args, **kwargs): """Wrapper to call poe.call with state context""" _state.set(state) return poe.call(*args, **kwargs) def parse_json_from_response(response_text, required_fields=None): """Extract and parse JSON from response.""" # Try to find ```json codeblock first match = re.search(r'```json\s*(\{.*?\})\s*```', response_text, re.DOTALL) if match: json_str = match.group(1).strip() else: # Search for JSON object anywhere in response (handles thinking blocks, raw JSON, etc.) match = re.search(r'\{.*\}', response_text, re.DOTALL) if match: json_str = match.group(0).strip() else: raise Exception(f"Could not find JSON in response\n{response_text}") try: parsed = json.loads(json_str) except json.JSONDecodeError as e: raise Exception(f"Failed to parse JSON: {e}\nJSON string: {json_str}") if required_fields: missing_fields = set(required_fields) - set(parsed.keys()) if missing_fields: raise Exception(f"Missing required fields: {', '.join(missing_fields)}") return parsed def generate_study_app_html(flashcards, mcqs, notes, title="Study Space"): """Generate an interactive HTML study app with Notion-esque design.""" # Base64 encode to avoid ANY escaping issues flashcards_b64 = base64.b64encode(json.dumps(flashcards).encode()).decode() mcqs_b64 = base64.b64encode(json.dumps(mcqs).encode()).decode() notes_b64 = base64.b64encode(json.dumps(notes).encode()).decode() title_b64 = base64.b64encode(title.encode()).decode() html = """ Study Space

📚

Study Space

📄 Study Notes

Read

Loading notes preview...

🎴 Flashcards

0 Cards

Review key concepts and definitions using active recall.

❓ Practice Quiz

0 Qs

Test your knowledge with multiple choice questions.

Card 1 of 0

Spacebar to flip • Arrows to navigate

Question 1/0 Score: 0

Final Score

0/0

""" # Replace placeholders with base64-encoded data html = html.replace('FLASHCARDS_B64', flashcards_b64) html = html.replace('MCQS_B64', mcqs_b64) html = html.replace('NOTES_B64', notes_b64) html = html.replace('TITLE_B64', title_b64) return html def extract_video_id(url): """Extract video ID from YouTube URL""" import re patterns = [ r'(?:youtube\.com\/watch\?v=|youtu\.be\/)([^&\n?]*)', r'youtube\.com\/embed\/([^&\n?]*)', r'youtube\.com\/v\/([^&\n?]*)' ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) return None def is_youtube_url(text): """Check if text contains a YouTube URL""" return text and ("youtube.com" in text or "youtu.be" in text) def fetch_youtube_transcript(video_id): """Fetch YouTube transcript using Invidious instances""" import json import urllib.request import urllib.error import re # Working public Invidious instances invidious_instances = [ "inv.nadeko.net", "yewtu.be", "invidious.nerdvpn.de", "inv.perditum.com", "invidious.f5.si" ] transcript_text = None video_title = None working_instance = None # Try each instance until one works for instance in invidious_instances: try: # Get video info including captions api_url = f"https://{instance}/api/v1/videos/{video_id}" req = urllib.request.Request(api_url) req.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36') with urllib.request.urlopen(req, timeout=20) as response: data = json.loads(response.read().decode()) video_title = data.get('title', 'Unknown') # Check for captions in the response captions = data.get('captions', []) if not captions: continue # Try to find English captions first, then any caption selected_caption = None # First pass: Look for English captions for caption in captions: label = caption.get('label', '').lower() lang_code = caption.get('languageCode', '').lower() if 'english' in label or lang_code.startswith('en'): selected_caption = caption # Prefer non-auto-generated if 'auto' not in label: break # Second pass: If no English, take first available if not selected_caption and captions: selected_caption = captions[0] if not selected_caption: continue # Get the caption URL caption_url = selected_caption.get('url') if not caption_url: continue # Make URL absolute if needed if caption_url.startswith('/'): caption_url = f"https://{instance}{caption_url}" # Fetch the actual caption content caption_req = urllib.request.Request(caption_url) caption_req.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36') with urllib.request.urlopen(caption_req, timeout=20) as caption_response: caption_content = caption_response.read().decode('utf-8', errors='ignore') # Parse the caption content (usually VTT format) lines = caption_content.split('\n') text_lines = [] for line in lines: line = line.strip() # Skip VTT headers, timestamps, and empty lines if (not line or line.startswith('WEBVTT') or line.startswith('Kind:') or line.startswith('Language:') or '-->' in line or (line.isdigit() and len(line) <= 4)): continue # Remove HTML/XML tags line = re.sub(r'<[^>]+>', '', line) if line and len(line) > 1: text_lines.append(line) transcript_text = ' '.join(text_lines) # Verify we got meaningful content if transcript_text and len(transcript_text) > 100: working_instance = instance break except urllib.error.HTTPError: continue except urllib.error.URLError: continue except Exception: continue # If we couldn't get a transcript from any instance if not transcript_text: raise Exception( f"Could not retrieve transcript for video {video_id}. " "This could be because:\n" "- The video doesn't have captions/subtitles\n" "- All Invidious instances are temporarily down\n" "- The video is age-restricted or private" ) return transcript_text, video_title def is_audio_file(filename): """Check if file is an audio file.""" audio_extensions = ['.mp3', '.wav', '.m4a', '.flac', '.ogg', '.aac', '.wma', '.opus'] filename_lower = filename.lower() return any(filename_lower.endswith(ext) for ext in audio_extensions) def is_video_file(filename): """Check if file is a video file.""" video_extensions = ['.mp4', '.mov', '.avi', '.mkv', '.webm', '.flv', '.wmv', '.m4v', '.mpeg', '.mpg'] filename_lower = filename.lower() return any(filename_lower.endswith(ext) for ext in video_extensions) def extract_audio_from_video(video_bytes): """Extract audio from video using ffmpeg.""" import tempfile import subprocess import os video_path = None audio_path = None try: # Create temp file for video with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as video_file: video_path = video_file.name video_file.write(video_bytes) # Create temp path for audio output audio_path = tempfile.mktemp(suffix='.wav') # Extract audio using ffmpeg # Use WAV format (universally supported, uncompressed) result = subprocess.run([ 'ffmpeg', '-loglevel', 'error', # Only show errors '-i', video_path, '-vn', # No video '-acodec', 'pcm_s16le', # PCM 16-bit little-endian (standard WAV) '-ar', '16000', # Sample rate 16kHz (good for speech) '-ac', '1', # Mono audio '-y', # Overwrite output file audio_path ], capture_output=True, text=True, check=False) # Check if ffmpeg succeeded if result.returncode != 0: error_msg = result.stderr.strip() if result.stderr else "Unknown error" # Print full error for debugging (won't be truncated in exception) with poe.start_message() as msg: msg.write(f"**ffmpeg error output:**\n```\n{error_msg}\n```") # Limit error message length for exception if len(error_msg) > 500: error_msg = error_msg[:500] + "..." raise Exception(f"ffmpeg audio extraction failed. See error output above for details.") # Check if audio file was created if not os.path.exists(audio_path): raise Exception("Audio file was not created - video may not have an audio track") if os.path.getsize(audio_path) == 0: raise Exception("Extracted audio file is empty - video may not have an audio track") # Read audio bytes with open(audio_path, 'rb') as f: audio_bytes = f.read() return audio_bytes finally: # Clean up temp files if video_path and os.path.exists(video_path): try: os.remove(video_path) except: pass if audio_path and os.path.exists(audio_path): try: os.remove(audio_path) except: pass def extract_text_from_attachment(attachment): """Extract text content from various file types.""" filename = attachment.name.lower() # Check if attachment is a URL (for HTML content) if attachment.url and (attachment.url.startswith('http://') or attachment.url.startswith('https://')) and attachment.content_type == 'text/html': # For URLs, fetch the HTML content directly response = requests.get(attachment.url, headers={'User-Agent': 'Mozilla/5.0'}) response.raise_for_status() from bs4 import BeautifulSoup soup = BeautifulSoup(response.content, 'html.parser') # Remove script and style elements for script in soup(["script", "style"]): script.decompose() # Get text and clean up whitespace text = soup.get_text() lines = (line.strip() for line in text.splitlines()) chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) text = '\n'.join(chunk for chunk in chunks if chunk) return text # Check if it's a video file - extract audio and transcribe if is_video_file(filename): # Download video file response = requests.get(attachment.url) video_bytes = response.content # Extract audio using ffmpeg audio_bytes = extract_audio_from_video(video_bytes) # Create audio attachment for Whisper audio_attachment = poe.Attachment( name="extracted_audio.wav", contents=audio_bytes, content_type="audio/wav" ) # Transcribe audio with Whisper transcription_response = poe.call("Whisper-V3-Large-T", poe.Message(text="", attachments=[audio_attachment])) return transcription_response.text # Check if it's an audio file - if so, use Whisper to transcribe if is_audio_file(filename): transcription_response = poe.call("Whisper-V3-Large-T", poe.Message(text="", attachments=[attachment])) return transcription_response.text # For document files, download and parse response = requests.get(attachment.url) file_content = response.content if filename.endswith('.txt'): return file_content.decode('utf-8', errors='ignore') elif filename.endswith('.pdf'): import pdfplumber pdf_file = io.BytesIO(file_content) text = "" with pdfplumber.open(pdf_file) as pdf: for page in pdf.pages: page_text = page.extract_text() if page_text: text += page_text + "\n" return text elif filename.endswith('.docx') or filename.endswith('.doc'): from docx import Document doc_file = io.BytesIO(file_content) doc = Document(doc_file) text = "" for paragraph in doc.paragraphs: text += paragraph.text + "\n" return text elif filename.endswith('.pptx') or filename.endswith('.ppt'): from pptx import Presentation ppt_file = io.BytesIO(file_content) prs = Presentation(ppt_file) text = "" for slide in prs.slides: for shape in slide.shapes: if hasattr(shape, "text"): text += shape.text + "\n" return text elif filename.endswith('.html') or filename.endswith('.htm'): from bs4 import BeautifulSoup html_text = file_content.decode('utf-8', errors='ignore') soup = BeautifulSoup(html_text, 'html.parser') # Remove script and style elements for script in soup(["script", "style"]): script.decompose() # Get text and clean up whitespace text = soup.get_text() lines = (line.strip() for line in text.splitlines()) chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) text = '\n'.join(chunk for chunk in chunks if chunk) return text else: raise poe.BotError(f"Unsupported file type: {filename}. Please provide a PDF, DOC, DOCX, PPT, PPTX, TXT, HTML, audio file (MP3, WAV, M4A, etc.), or video file (MP4, MOV, AVI, etc.).") # Validate input - need either text or attachment(s) if not poe.query.attachments and not poe.query.text.strip(): raise poe.BotError("Please provide either:\n- Text content to generate study materials from\n- YouTube URL\n- One or more attached files (PDF, DOC, DOCX, PPT, PPTX, TXT, HTML, audio files like MP3/WAV/M4A, or video files like MP4/MOV/AVI)\n- Or any combination of the above") # Determine input source attachments = poe.query.attachments # Can be multiple now text_input = poe.query.text.strip() # Extract the Poe interpreter state state = _state.extract() # Create single progress message progress_msg = poe.default_chat.start_message() async def run_generation(): """Generate study materials with animated progress""" # Phase 1: Extract/collect text from all sources document_text = "" # Handle text input - check for YouTube URL first if text_input: if is_youtube_url(text_input): await ui_call(state, progress_msg.write, f"🎥 **Fetching YouTube transcript...**\n") # Extract video ID video_id = extract_video_id(text_input) if not video_id: raise poe.BotError("Could not extract video ID from YouTube URL") # Fetch transcript in background thread def fetch_with_state(): """Fetch transcript with state context""" _state.set(state) return fetch_youtube_transcript(video_id) try: transcript_text, video_title = await asyncio.to_thread(fetch_with_state) await ui_call(state, progress_msg.write, f" ✓ Retrieved transcript for: {video_title}\n") document_text = f"YouTube Video: {video_title}\n\n{transcript_text}" except Exception as e: # If YouTube fetch fails but we have other content, warn but continue if attachments: await ui_call(state, progress_msg.write, f" ⚠️ Could not fetch YouTube transcript: {str(e)}\n") await ui_call(state, progress_msg.write, f" → Continuing with attached files...\n") else: raise poe.BotError(f"Failed to fetch YouTube transcript: {str(e)}") else: await ui_call(state, progress_msg.write, f"📝 **Processing text input...**\n") document_text = text_input # Handle multiple attachments if present if attachments: await ui_call(state, progress_msg.write, f"📄 **Extracting text from {len(attachments)} file(s)...**\n") # Process each attachment for idx, attachment in enumerate(attachments): # Check if this is a YouTube HTML attachment - if so, fetch transcript instead if is_youtube_url(attachment.name) or is_youtube_url(attachment.url): await ui_call(state, progress_msg.write, f" 🎥 Processing YouTube video ({idx + 1}/{len(attachments)})\n") video_id = extract_video_id(attachment.name) or extract_video_id(attachment.url) if not video_id: await ui_call(state, progress_msg.write, f" ⚠️ Could not extract video ID, skipping\n") continue def fetch_with_state(): """Fetch transcript with state context""" _state.set(state) return fetch_youtube_transcript(video_id) try: transcript_text, video_title = await asyncio.to_thread(fetch_with_state) await ui_call(state, progress_msg.write, f" ✓ Retrieved transcript for: {video_title}\n") attachment_text = f"YouTube Video: {video_title}\n\n{transcript_text}" # Combine with existing text if document_text: document_text = f"{document_text}\n\n---\n\n{attachment_text}" else: document_text = attachment_text except Exception as e: await ui_call(state, progress_msg.write, f" ⚠️ Could not fetch YouTube transcript: {str(e)}\n") continue else: # Regular file attachment await ui_call(state, progress_msg.write, f" → Processing {attachment.name} ({idx + 1}/{len(attachments)})\n") def extract_with_state(): """Extract text with state context""" _state.set(state) return extract_text_from_attachment(attachment) try: attachment_text = await asyncio.to_thread(extract_with_state) # Truncate to first 500000 characters if needed if len(attachment_text) > 500000: attachment_text = attachment_text[:500000] await ui_call(state, progress_msg.write, f" ⚠️ Text truncated to 500,000 characters\n") except Exception as e: raise poe.BotError(f"Failed to extract text from {attachment.name}: {str(e)}") # Combine attachment texts with clear separation if document_text: document_text = f"{document_text}\n\n---\n\n# Content from: {attachment.name}\n\n{attachment_text}" else: document_text = f"# Content from: {attachment.name}\n\n{attachment_text}" if not document_text.strip(): raise poe.BotError("No text content found to generate study materials from") # Create tasks for parallel generation async def generate_title(): _state.set(state) return await asyncio.to_thread( call_with_state, state, "Claude-Sonnet-4.5", f"""Based on the following document content, create a short, engaging title for a study set (max 6 words). Output in ```json format: ```json {{"title": "Your Engaging Title Here"}} ``` Make it specific to the content, not generic. Examples: - "Cell Biology Fundamentals" - "World War II History" - "Python Programming Basics" - "Quantum Physics Concepts" {document_text[:1000]} """ ) async def generate_notes(): _state.set(state) return await asyncio.to_thread( call_with_state, state, "Claude-Sonnet-4.5", f"""You are helping a student study. Based on the following document content, create concise study notes in markdown format. Use: - Clear headers (##, ###) to organize topics - Bullet points (-) for unordered lists - Numbered lists (1., 2., 3.) for ordered items - Bold (**text**) for key terms WITHIN paragraphs or list items - Blockquotes (> text) for important definitions, key quotes, or emphasis - Tables for comparisons, timelines, or structured data (use this exact format): | Header 1 | Header 2 | |----------|----------| | Data 1 | Data 2 | - Short, digestible paragraphs IMPORTANT formatting rules: - Always use a header (##, ###) to introduce a list, NEVER use standalone bold text followed by bullet points - Bold text should only appear within paragraphs or list items, not on its own line before a list - NEVER ADD A LIST AFTER BOLD TEXT Make the notes visually engaging! Use blockquotes to highlight important concepts and tables to organize related information. These formatting elements make notes easier to scan and remember. Focus on essential concepts only. Aim for clarity and brevity. Output ONLY the markdown notes (no JSON, no code blocks, just the raw markdown text). {document_text} """ ) async def generate_flashcards(): _state.set(state) return await asyncio.to_thread( call_with_state, state, "Claude-Sonnet-4.5", f"""You are helping a student study. Based on the following document content, create flashcards to help the student revise key concepts. Output in ```json format: ```json {{ "flashcards": [ {{"question": "Question text here", "answer": "Answer text here"}}, {{"question": "Question text here", "answer": "Answer text here"}} ] }} ``` Create 10-20 flashcards covering all important concepts. {document_text} """ ) async def generate_mcqs(): _state.set(state) return await asyncio.to_thread( call_with_state, state, "Claude-Sonnet-4.5", f"""You are helping a student study. Based on the following document content, create multiple choice questions with 4 options each to test understanding. Output in ```json format: ```json {{ "mcqs": [ {{ "question": "Question text here", "options": ["Option A", "Option B", "Option C", "Option D"], "correct": "A" }} ] }} ``` Create at least 8-10 questions covering all key topics. The "correct" field should be "A", "B", "C", or "D". {document_text} """ ) # Create all tasks title_task = asyncio.create_task(generate_title()) notes_task = asyncio.create_task(generate_notes()) flashcards_task = asyncio.create_task(generate_flashcards()) mcqs_task = asyncio.create_task(generate_mcqs()) tasks = [title_task, notes_task, flashcards_task, mcqs_task] task_names = ["Study Set Title", "Study Notes", "Flashcards", "Quiz Questions"] # Animation loop - update display while tasks are running frame_idx = 0 start_time = time.time() last_update_time = 0 while not all(task.done() for task in tasks): current_time = time.time() # Throttle updates - only update if 150ms has passed since last update if current_time - last_update_time >= 0.15: await asyncio.sleep(0) # Yield before building content # Build progress content elapsed = int(current_time - start_time) loading_frame = LOADING_FRAMES[frame_idx % len(LOADING_FRAMES)] # Count completed tasks completed = sum(1 for task in tasks if task.done()) content = f"{loading_frame} **Generating study materials...** ({completed}/4 complete)\n\n" content += f"⏱️ {elapsed}s elapsed\n\n" # Show status for each task for i, (task, name) in enumerate(zip(tasks, task_names)): if task.done(): content += f"✓ {name}\n" else: content += f"⏳ {name}...\n" # Update the message await ui_call(state, progress_msg.overwrite, content) last_update_time = current_time frame_idx += 1 await asyncio.sleep(0.05) # Check frequently, update less frequently # Get results title_response = await title_task notes_response = await notes_task flashcards_response = await flashcards_task mcqs_response = await mcqs_task # Final update - show completion total_elapsed = int(time.time() - start_time) await ui_call(state, progress_msg.overwrite, f"✓ **All materials generated!** ({total_elapsed}s)\n\n✨ **Building interactive study app...**") return title_response, notes_response, flashcards_response, mcqs_response async def main(): """Main async wrapper for all operations""" # Run generation title_response, notes_response, flashcards_response, mcqs_response = await run_generation() # Extract content from JSON responses title_data = parse_json_from_response(title_response.text, required_fields=["title"]) study_title = title_data["title"] notes = notes_response.text.strip() flashcards_data = parse_json_from_response(flashcards_response.text, required_fields=["flashcards"]) flashcards = flashcards_data["flashcards"] mcqs_data = parse_json_from_response(mcqs_response.text, required_fields=["mcqs"]) mcqs = mcqs_data["mcqs"] if flashcards or mcqs: # Generate HTML study app html_content = generate_study_app_html(flashcards, mcqs, notes, study_title) # Update the same message with final content final_content = "✓ **Your interactive study app is ready!**\n\n" final_content += "## 💡 Interactive Study App\n\n" final_content += "```html\n" final_content += html_content final_content += "\n```\n\n" await ui_call(state, progress_msg.overwrite, final_content) progress_msg.finish() # Run main asyncio.run(main())

Quizly