#!/usr/bin/env python3 """ Multi-Layer Trend Analysis Pipeline Layer 1: EMBED + CLUSTER (local) - sentence-transformers, UMAP, HDBSCAN Layer 2: FAST LLM CLASSIFICATION (Groq Llama 3.3 70B) Layer 3: DEEP ANALYSIS (Claude CLI via Max subscription) Layer 3.5: DEEP RESEARCH (Hybrid Research Service API) Outputs: - output/analysis_clusters.json - output/analysis_classified.json - output/analysis_opportunities.json (includes research data) """ import json import os import sqlite3 import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import List, Dict, Any, Optional import numpy as np from dotenv import load_dotenv # Database operations for opportunities from db_opportunities import ( run_migration, create_analysis_run, update_analysis_run, insert_opportunities_batch, update_opportunity_research_by_cluster, get_latest_analysis_run ) class NumpyEncoder(json.JSONEncoder): """JSON encoder that handles numpy types.""" def default(self, obj): if isinstance(obj, np.integer): return int(obj) if isinstance(obj, np.floating): return float(obj) if isinstance(obj, np.ndarray): return obj.tolist() return super().default(obj) def extract_weighted_score(analysis: Dict[str, Any]) -> float: """ Robustly extract weighted_score from Claude's response. Handles various formats Claude might return: - Direct float: 4.65 - String: "4.65" - String with denominator: "4.65/5.0" - Dict: {"score": 4.65} or {"value": 4.65} - Narrative: "The weighted score is 4.65" Falls back to calculating from dimension scores if all else fails. """ import re raw = analysis.get("weighted_score") # Try direct float if isinstance(raw, (int, float)) and raw > 0: return float(raw) # Try dict extraction if isinstance(raw, dict): for key in ["score", "value", "total", "weighted_score"]: if key in raw and isinstance(raw[key], (int, float)): return float(raw[key]) # Try string parsing if isinstance(raw, str): # Try direct conversion try: val = float(raw) if val > 0: return val except ValueError: pass # Try "4.65/5.0" format if "/" in raw: try: val = float(raw.split("/")[0].strip()) if val > 0: return val except ValueError: pass # Try to extract a decimal number from narrative text match = re.search(r'(\d+\.?\d*)\s*(?:/\s*5|out of 5)?', raw) if match: try: val = float(match.group(1)) if 0 < val <= 5: return val except ValueError: pass # Check alternative field names for alt_field in ["weighted_score_calculation", "weighted_score_math", "final_score", "total_score"]: alt_raw = analysis.get(alt_field) if alt_raw: if isinstance(alt_raw, (int, float)) and alt_raw > 0: return float(alt_raw) if isinstance(alt_raw, str): # Extract final number from calculation string like "= 4.65" match = re.search(r'=\s*(\d+\.?\d*)', alt_raw) if match: try: val = float(match.group(1)) if val > 0: return val except ValueError: pass # FALLBACK: Calculate from dimension scores scores = analysis.get("scores", {}) if scores: # v4 weights (scroll-stop focused) dimension_map_v4 = { "scroll_stop_power": 0.25, "completion_probability": 0.25, "share_save_potential": 0.15, "demand_signal": 0.15, "visual_potential": 0.10, "evergreen_potential": 0.10, } # v2/v3 curiosity-mining weights dimension_map_v3 = { "curiosity_intensity": 0.30, "demand_signal": 0.20, "content_gap": 0.15, "depth_explainability": 0.15, "evergreen_potential": 0.10, "audience_breadth": 0.10, } # v1 old field names for backward compatibility dimension_map_v1 = { "demand_signal": 0.25, "content_gap": 0.25, "explainability": 0.15, "evergreen_potential": 0.15, "audience_breadth": 0.10, "competition_gap": 0.10, } # Detect version based on which fields exist if "scroll_stop_power" in scores: active_map = dimension_map_v4 elif "curiosity_intensity" in scores: active_map = dimension_map_v3 else: active_map = dimension_map_v1 total = 0.0 found_any = False for dim_name, weight in active_map.items(): dim_data = scores.get(dim_name, {}) if isinstance(dim_data, dict): dim_score = dim_data.get("score", 0) elif isinstance(dim_data, (int, float)): dim_score = dim_data else: dim_score = 0 if dim_score > 0: found_any = True total += dim_score * weight if found_any and total > 0: return round(total, 2) # If we get here, we couldn't extract or calculate a score return 0.0 def validate_verdict(analysis: Dict[str, Any]) -> str: """ Validate and potentially correct the verdict based on MAKE_NOW gates. MAKE_NOW requires ALL 6 gates to pass AND weighted_score >= 4.0. If verdict is MAKE_NOW but gates don't all pass, downgrade to WORTH_MAKING. """ verdict = analysis.get("verdict", "SKIP") # Only validate MAKE_NOW verdicts if verdict != "MAKE_NOW": return verdict # Check gates gates = analysis.get("make_now_gates", {}) gates_passed = gates.get("gates_passed", 0) # If gates_passed is not set, count them manually if gates_passed == 0 and gates: gate_names = [ "scroll_stop_test", "universal_access_test", "depth_test", "share_test", "satisfaction_test", "channel_fit_test" ] gates_passed = sum(1 for g in gate_names if gates.get(g) is True) # Check weighted score weighted_score = extract_weighted_score(analysis) # MAKE_NOW requires all 6 gates AND score >= 4.0 if gates_passed < 6 or weighted_score < 4.0: # Downgrade to WORTH_MAKING print(f" [VALIDATE] Downgrading MAKE_NOW to WORTH_MAKING (gates: {gates_passed}/6, score: {weighted_score})") return "WORTH_MAKING" return "MAKE_NOW" def assign_verdicts_by_rank(opportunities: List[Dict]) -> List[Dict]: """ Reassign verdicts based on relative ranking rather than absolute scores. This enforces the target distribution mechanically. Distribution: - Top 15% by weighted_score → MAKE_NOW - Next 35% → WORTH_MAKING - Next 20% → NEEDS_RESEARCH - Bottom 30% → SKIP Exception: Any opportunity the model already marked SKIP stays SKIP regardless of score (the model identified a structural reason it shouldn't be produced). """ # Separate model-assigned SKIPs (keep them) model_skips = [o for o in opportunities if o.get("verdict") == "SKIP"] scoreable = [o for o in opportunities if o.get("verdict") != "SKIP"] # Sort by weighted_score descending scoreable.sort(key=lambda x: extract_weighted_score(x), reverse=True) n = len(scoreable) if n == 0: return opportunities # Calculate cutoff indices make_now_cutoff = max(1, int(n * 0.15)) worth_making_cutoff = make_now_cutoff + max(1, int(n * 0.35)) needs_research_cutoff = worth_making_cutoff + max(1, int(n * 0.20)) adjusted_count = 0 for i, opp in enumerate(scoreable): original = opp.get("verdict", "UNKNOWN") if i < make_now_cutoff: new_verdict = "MAKE_NOW" elif i < worth_making_cutoff: new_verdict = "WORTH_MAKING" elif i < needs_research_cutoff: new_verdict = "NEEDS_RESEARCH" else: new_verdict = "SKIP" # Track if verdict changed if new_verdict != original: opp["verdict_adjusted"] = True opp["original_verdict"] = original opp["verdict"] = new_verdict adjusted_count += 1 print(f" [RANK] Adjusted {adjusted_count} verdicts based on relative ranking") print(f" [RANK] Distribution: MAKE_NOW={make_now_cutoff}, WORTH_MAKING={worth_making_cutoff - make_now_cutoff}, NEEDS_RESEARCH={needs_research_cutoff - worth_making_cutoff}, SKIP={n - needs_research_cutoff + len(model_skips)}") return scoreable + model_skips # Load environment load_dotenv() OUTPUT_DIR = Path(__file__).parent / "output" DB_PATH = OUTPUT_DIR / "trends_history.db" def get_opportunities_path(channel_id: str = None) -> Path: """ Get the path for opportunities JSON file. Args: channel_id: If specified, returns channel-specific path. If None, returns the generic path (backward compatibility). Returns: Path to the opportunities JSON file """ if channel_id: return OUTPUT_DIR / f"analysis_opportunities_{channel_id}.json" return OUTPUT_DIR / "analysis_opportunities.json" def backup_opportunities_file(channel_id: str = None) -> Optional[Path]: """ Create a backup of the existing opportunities file before writing. Args: channel_id: Channel ID for channel-specific file Returns: Path to backup file, or None if no file existed to backup """ import shutil source_path = get_opportunities_path(channel_id) if not source_path.exists(): return None backup_path = source_path.with_suffix('.json.bak') shutil.copy2(source_path, backup_path) return backup_path def load_existing_opportunities(channel_id: str = None) -> Dict[str, Any]: """ Load existing opportunities file if it exists. Args: channel_id: Channel ID for channel-specific file Returns: Existing data dict, or empty structure if file doesn't exist """ path = get_opportunities_path(channel_id) if path.exists(): with open(path, 'r') as f: return json.load(f) return {"opportunities": [], "channel_id": channel_id} def merge_opportunities( existing_data: Dict[str, Any], new_opportunities: List[Dict[str, Any]], replace_mode: bool = False ) -> List[Dict[str, Any]]: """ Merge new opportunities with existing ones. Matching is done by cluster_id. New analysis updates existing entries. Opportunities not in the new batch are preserved. Args: existing_data: Existing opportunities data structure new_opportunities: Newly analyzed opportunities replace_mode: If True, completely replace (old behavior). If False, merge. Returns: Merged list of opportunities """ if replace_mode: return new_opportunities existing_opps = existing_data.get("opportunities", []) # Build index of existing opportunities by cluster_id existing_by_cluster = {} for opp in existing_opps: cluster_id = opp.get("cluster_id") if cluster_id is not None: existing_by_cluster[cluster_id] = opp # Build set of cluster_ids in new batch new_cluster_ids = set() for opp in new_opportunities: cluster_id = opp.get("cluster_id") if cluster_id is not None: new_cluster_ids.add(cluster_id) # Start with new opportunities (they have fresh Layer 1-3 analysis) merged = [] for new_opp in new_opportunities: cluster_id = new_opp.get("cluster_id") if cluster_id is not None and cluster_id in existing_by_cluster: # Merge: preserve Layer 3.5 and Layer 4 fields from existing existing_opp = existing_by_cluster[cluster_id] merged_opp = new_opp.copy() # Preserve research fields (Layer 3.5) research_fields = [ "research_completed", "research_report_path", "research_report_content", "research_word_count", "research_source_count", "research_duration_seconds", "research_generated_at", "research_query", "research_error" ] for field in research_fields: if field in existing_opp and field not in merged_opp: merged_opp[field] = existing_opp[field] # Preserve video generation fields (Layer 4) video_fields = [ "video_concepts", "layer4a_metadata", "layer4a_error", "video_script", "production_spec", "layer4b_metadata", "layer4b_error", "visual_direction", "asset_specs", "layer4c_metadata", "layer4c_error" ] for field in video_fields: if field in existing_opp and field not in merged_opp: merged_opp[field] = existing_opp[field] merged.append(merged_opp) else: merged.append(new_opp) # Add existing opportunities that weren't in the new batch (preserve old data) for cluster_id, existing_opp in existing_by_cluster.items(): if cluster_id not in new_cluster_ids: merged.append(existing_opp) return merged def check_dependencies(): """Verify all required packages are available.""" required = [ "sentence_transformers", "umap", "hdbscan", "groq", ] missing = [] for pkg in required: try: __import__(pkg) except ImportError: missing.append(pkg) if missing: print(f"[ERROR] Missing packages: {missing}") print("Run: pip install sentence-transformers umap-learn hdbscan groq") sys.exit(1) def check_api_keys(): """Verify API keys are configured.""" groq_key = os.getenv("GROQ_API_KEY") if not groq_key: print("[ERROR] GROQ_API_KEY not found in environment or .env file") sys.exit(1) # Test Claude CLI try: result = subprocess.run( ["claude", "--print", "-p", "Say OK"], input="", capture_output=True, text=True, timeout=90 ) if result.returncode != 0: print("[ERROR] Claude CLI not working") print(result.stderr) sys.exit(1) except FileNotFoundError: print("[ERROR] Claude CLI not found. Install with: npm install -g @anthropic-ai/claude-code") sys.exit(1) except subprocess.TimeoutExpired: print("[ERROR] Claude CLI timed out") sys.exit(1) print("[OK] All API keys verified") def get_last_run_id() -> Optional[int]: """Get the most recent collection run ID.""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT MAX(id) FROM collection_runs") result = cursor.fetchone() conn.close() return result[0] if result else None def load_trends_from_db( limit: int = None, since_run_id: int = None, channel_id: str = None ) -> List[Dict[str, Any]]: """Load trend items from SQLite database. Args: limit: Maximum number of items to load since_run_id: If set, only load items from runs >= this ID (for incremental analysis) channel_id: If set, only load items tagged to this channel """ conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cursor = conn.cursor() if channel_id: # Join with item_channels table to filter by channel query = """ SELECT DISTINCT ti.topic_name, ti.source, ti.region, ti.final_score, ti.selftext, ti.num_comments, ti.subreddit, ti.category, ti.url, ti.signal_type, ti.run_id, ic.confidence as channel_confidence FROM trend_items ti JOIN item_channels ic ON ti.id = ic.item_id WHERE ti.topic_name IS NOT NULL AND ti.topic_name != '' AND ic.channel_id = ? """ params = [channel_id] if since_run_id: query += " AND ti.run_id >= ?" params.append(since_run_id) query += " ORDER BY ic.confidence DESC, ti.final_score DESC" if limit: query += f" LIMIT {limit}" cursor.execute(query, params) else: query = """ SELECT DISTINCT topic_name, source, region, final_score, selftext, num_comments, subreddit, category, url, signal_type, run_id FROM trend_items WHERE topic_name IS NOT NULL AND topic_name != '' """ if since_run_id: query += f" AND run_id >= {since_run_id}" query += " ORDER BY final_score DESC" if limit: query += f" LIMIT {limit}" cursor.execute(query) rows = cursor.fetchall() conn.close() trends = [] seen_topics = set() for row in rows: topic = row["topic_name"].strip() if topic.lower() not in seen_topics: seen_topics.add(topic.lower()) trends.append({ "topic_name": topic, "source": row["source"], "region": row["region"], "score": row["final_score"] or 0, "selftext": row["selftext"] or "", "num_comments": row["num_comments"] or 0, "subreddit": row["subreddit"] or "", "category": row["category"] or "", "url": row["url"] or "", "signal_type": row["signal_type"] or "unknown", "run_id": row["run_id"] }) return trends # ============================================================================= # LAYER 1: EMBEDDING + CLUSTERING # ============================================================================= def layer1_embed_and_cluster(trends: List[Dict[str, Any]]) -> Dict[str, Any]: """ Layer 1: Embed topics with sentence-transformers, reduce with UMAP, cluster with HDBSCAN. """ from sentence_transformers import SentenceTransformer import umap import hdbscan print("\n" + "=" * 60) print("LAYER 1: EMBEDDING + CLUSTERING") print("=" * 60) # Extract topic names for embedding topic_texts = [] for t in trends: # Combine topic name with context if available text = t["topic_name"] if t["selftext"]: text += " " + t["selftext"][:200] topic_texts.append(text) print(f"[1.1] Embedding {len(topic_texts)} topics with all-MiniLM-L6-v2...") model = SentenceTransformer("all-MiniLM-L6-v2") embeddings = model.encode(topic_texts, show_progress_bar=True) print(f" Embedding shape: {embeddings.shape}") # Dimensionality reduction with UMAP print("[1.2] Reducing dimensions with UMAP (384 -> 10)...") reducer = umap.UMAP( n_components=10, n_neighbors=15, min_dist=0.1, metric="cosine", random_state=42 ) reduced = reducer.fit_transform(embeddings) print(f" Reduced shape: {reduced.shape}") # Clustering with HDBSCAN print("[1.3] Clustering with HDBSCAN...") clusterer = hdbscan.HDBSCAN( min_cluster_size=5, min_samples=3, metric="euclidean", cluster_selection_method="eom" ) cluster_labels = clusterer.fit_predict(reduced) n_clusters = len(set(cluster_labels)) - (1 if -1 in cluster_labels else 0) n_noise = list(cluster_labels).count(-1) print(f" Found {n_clusters} clusters, {n_noise} noise points") # Organize results by cluster - preserve ALL data for downstream layers clusters = {} for i, (trend, label) in enumerate(zip(trends, cluster_labels)): label_str = str(label) if label_str not in clusters: clusters[label_str] = { "cluster_id": label, "topics": [], "reddit_posts": [], # Separate list for Reddit items with full data "sources": set(), "regions": set(), "signal_types": set() # Track signal types for scoring boost } # Store full topic data including signal_type topic_data = { "topic_name": trend["topic_name"], "source": trend["source"], "region": trend["region"], "score": trend["score"], "url": trend["url"], "signal_type": trend.get("signal_type", "unknown") } clusters[label_str]["topics"].append(topic_data) clusters[label_str]["signal_types"].add(trend.get("signal_type", "unknown")) # If Reddit, also store in reddit_posts with full engagement data if trend["source"] == "reddit": clusters[label_str]["reddit_posts"].append({ "title": trend["topic_name"], "subreddit": trend["subreddit"], "num_comments": trend["num_comments"], "score": trend["score"], "selftext": trend["selftext"][:500] if trend["selftext"] else "", "url": trend["url"] }) clusters[label_str]["sources"].add(trend["source"]) clusters[label_str]["regions"].add(trend["region"]) # Convert sets to lists for JSON serialization and add metrics for c in clusters.values(): c["sources"] = list(c["sources"]) c["regions"] = list(c["regions"]) c["signal_types"] = list(c["signal_types"]) c["topic_count"] = len(c["topics"]) c["cross_source_count"] = len(c["sources"]) c["reddit_post_count"] = len(c["reddit_posts"]) # Count question-type signals (strongest indicator of explainer opportunity) question_count = sum(1 for t in c["topics"] if t.get("signal_type") == "question") c["question_signal_count"] = question_count c["has_question_signal"] = question_count > 0 # Sort clusters by size (excluding noise cluster -1) sorted_clusters = sorted( [c for c in clusters.values() if c["cluster_id"] != -1], key=lambda x: x["topic_count"], reverse=True ) # Add noise cluster at end if exists if "-1" in clusters: noise_cluster = clusters["-1"] noise_cluster["cluster_id"] = -1 noise_cluster["label"] = "Unclustered/Noise" result = { "total_topics": len(trends), "total_clusters": n_clusters, "noise_points": n_noise, "clusters": sorted_clusters, "timestamp": datetime.now(timezone.utc).isoformat() } # Save Layer 1 output output_path = OUTPUT_DIR / "analysis_clusters.json" with open(output_path, "w") as f: json.dump(result, f, indent=2, cls=NumpyEncoder) print(f"\n[1.4] Saved: {output_path}") return result # ============================================================================= # LAYER 2: GROQ FAST CLASSIFICATION + FILTERING # ============================================================================= def layer2_classify_and_filter(clusters_data: Dict[str, Any]) -> Dict[str, Any]: """ Layer 2: Use Groq Llama 3.3 70B to classify and FILTER clusters. 3-category system: EXPLAINER_OPPORTUNITY, CURIOSITY_MAGNET, KILL. """ from groq import Groq import re print("\n" + "=" * 60) print("LAYER 2: GROQ CLASSIFICATION + FILTERING") print("=" * 60) client = Groq(api_key=os.getenv("GROQ_API_KEY")) all_classified = [] passed_clusters = [] killed_clusters = [] clusters = clusters_data["clusters"] for i, cluster in enumerate(clusters[:50]): # Evaluate top 50 clusters # Get top 5 representative topics by score topics = sorted(cluster["topics"], key=lambda x: x["score"], reverse=True)[:5] topic_list = "\n".join([f"- {t['topic_name']} (source: {t['source']})" for t in topics]) # Build Reddit context if available reddit_context = "" if cluster.get("reddit_posts"): reddit_items = sorted(cluster["reddit_posts"], key=lambda x: x.get("num_comments", 0), reverse=True)[:5] reddit_lines = [] for r in reddit_items: reddit_lines.append(f"- r/{r['subreddit']}: \"{r['title']}\" ({r.get('num_comments', 0)} comments)") reddit_context = "Reddit: " + " | ".join(reddit_lines) # Build signal type context signal_context = f"Signal types: {', '.join(cluster.get('signal_types', []))}" if cluster.get("has_question_signal"): signal_context += f" ({cluster['question_signal_count']} question signals)" sources = ', '.join(cluster['sources']) prompt = f"""Classify this topic cluster. Pick ONE category. CLUSTER: Topics: {topic_list} Sources: {sources} {signal_context} {reddit_context} CATEGORIES: EXPLAINER_OPPORTUNITY = People want something EXPLAINED. Look for: - Questions starting with "Why...", "How does...", "What is...", "ELI5..." - Reddit posts asking for understanding, not just news - Topics where the answer requires more than one sentence - Examples that ARE explainer opportunities: "Why does every power plant boil water?", "How do undersea cables handle all internet traffic?", "What are the Epstein files?" - Examples that are NOT: "Who won the Super Bowl?", "What time is the game?", "Breaking: CEO resigns" CURIOSITY_MAGNET = Not a question, but fascinating. Look for: - "What is this thing?" identification posts with high engagement - Counterintuitive facts people love sharing - "I never knew that" moments - Examples: mystery object identification, surprising animal behavior, bizarre historical facts KILL = Everything else. This includes ALL of the following: - Sports scores, game results, player stats, team matchups - Stock prices, market moves, financial tickers - News headlines about events (elections, arrests, speeches, disasters) with no deeper "why" angle - Celebrity gossip, entertainment news, movie/TV releases - YouTube video titles (these are someone else's content) - Music releases, concert announcements - Weather, schedules, product launch dates - Political news that is about WHAT HAPPENED not WHY IT WORKS THAT WAY - Any topic that will be irrelevant in 2 weeks - Topics where a Google search gives you the complete answer in 3 seconds Ask: "Could someone make a 10-minute video explaining this that people would still watch 6 months from now?" If YES → EXPLAINER_OPPORTUNITY or CURIOSITY_MAGNET If NO → KILL Respond in this exact JSON format: {{ "classification": "EXPLAINER_OPPORTUNITY" or "CURIOSITY_MAGNET" or "KILL", "score": number from 1 to 10, "reason": "one sentence why", "the_question": "the specific question people want answered, or null if KILL" }}""" try: response = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=[{"role": "user", "content": prompt}], max_tokens=200, temperature=0.2 ) response_text = response.choices[0].message.content.strip() # Parse JSON response - fail closed on parse errors classification = None try: classification = json.loads(response_text) except json.JSONDecodeError: json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: try: classification = json.loads(json_match.group()) except json.JSONDecodeError: pass # Fail closed: if we can't parse, it's a KILL if classification is None: classification = { "classification": "KILL", "score": 0, "reason": "Could not parse Groq response", "the_question": None } # Normalize classification field cls_type = classification.get("classification", "KILL").upper() if cls_type not in ["EXPLAINER_OPPORTUNITY", "CURIOSITY_MAGNET", "KILL"]: cls_type = "KILL" classification["classification"] = cls_type # Build classified cluster with the_question for Layer 3 classified_cluster = { **cluster, "groq_classification": classification, "the_question": classification.get("the_question") } all_classified.append(classified_cluster) # Filter logic: # - EXPLAINER_OPPORTUNITY with score >= 6 → PASS # - CURIOSITY_MAGNET with score >= 7 → PASS # - KILL → always KILL score = classification.get("score", 0) passes_filter = False if cls_type == "EXPLAINER_OPPORTUNITY" and score >= 6: passes_filter = True elif cls_type == "CURIOSITY_MAGNET" and score >= 7: passes_filter = True the_question = classification.get("the_question", "") q_display = f" → {the_question[:50]}..." if the_question else "" if passes_filter: passed_clusters.append(classified_cluster) print(f" [{i+1}/{min(50, len(clusters))}] ✓ {cls_type} ({score}/10){q_display}") else: killed_clusters.append(classified_cluster) reason = classification.get("reason", "")[:40] print(f" [{i+1}/{min(50, len(clusters))}] ✗ {cls_type} ({score}/10) - {reason}") except Exception as e: print(f" [{i+1}] Error: {str(e)}") classification = { "classification": "KILL", "score": 0, "reason": f"Classification failed: {str(e)}", "the_question": None } killed_clusters.append({**cluster, "groq_classification": classification, "the_question": None}) # Summary stats kill_reasons = {} for c in killed_clusters: reason = c.get("groq_classification", {}).get("classification", "UNKNOWN") kill_reasons[reason] = kill_reasons.get(reason, 0) + 1 print(f"\n[2.1] FILTER RESULTS:") print(f" Total evaluated: {len(all_classified)}") print(f" PASSED: {len(passed_clusters)}") print(f" KILLED: {len(killed_clusters)}") print(f" Kill breakdown: {kill_reasons}") result = { "total_evaluated": len(all_classified), "total_passed": len(passed_clusters), "total_killed": len(killed_clusters), "kill_breakdown": kill_reasons, "passed_clusters": passed_clusters, "killed_clusters": killed_clusters, "timestamp": datetime.now(timezone.utc).isoformat() } # Save Layer 2 output output_path = OUTPUT_DIR / "analysis_classified.json" with open(output_path, "w") as f: json.dump(result, f, indent=2, cls=NumpyEncoder) print(f"[2.2] Saved: {output_path}") return result # ============================================================================= # LAYER 3: CLAUDE DEEP ANALYSIS # ============================================================================= def layer3_deep_analysis(classified_data: Dict[str, Any], channel_id: str = None, max_clusters: int = 0, replace_mode: bool = False) -> Dict[str, Any]: """ Layer 3: Use Claude Opus for deep analysis of each surviving cluster. Args: classified_data: Output from Layer 2 channel_id: Optional channel ID for channel-specific analysis context max_clusters: Limit analysis to N clusters (0 = unlimited, for testing use 3-5) replace_mode: If True, replace entire file. If False (default), merge with existing. """ print("\n" + "=" * 60) print("LAYER 3: CLAUDE OPUS DEEP ANALYSIS") print("=" * 60) # Load channel context if analyzing for a specific channel channel_context = "" channel_name = None hook_criteria = "" if channel_id: from channel_config import get_channel channel = get_channel(channel_id) if channel: channel_name = channel.name competitive_gap = channel.competitive_gap or "" itch_description = channel.itch_description or "" format_description = channel.format_description or "" hook_criteria = channel.hook_criteria or "" channel_context = f""" CHANNEL CONTEXT: You are analyzing opportunities for the YouTube channel "{channel.name}". Channel Identity: {channel.description} """ if itch_description: channel_context += f"The Itch (what viewer need this scratches): {itch_description}\n" if format_description: channel_context += f"Format Pattern: {format_description}\n" if competitive_gap: channel_context += f"Competitive Gap (what makes this channel unique): {competitive_gap}\n" channel_context += """ IMPORTANT: Score topics higher if they directly exploit this channel's competitive gap. Score topics lower if they don't fit the channel identity, even if they're otherwise good opportunities. A great topic for another channel is a bad topic for THIS channel. """ print(f"[3.0] Channel context loaded: {channel.name}") # Get clusters that passed Layer 2 filter passed_clusters = classified_data.get("passed_clusters", []) # Create analysis run record in database analysis_run_id = None if channel_id: try: analysis_run_id = create_analysis_run( channel_id=channel_id, items_analyzed=len(classified_data.get("all_trends", [])), clusters_formed=len(classified_data.get("clusters", [])), layer2_passed=len(passed_clusters) ) print(f"[3.0] Analysis run created: ID {analysis_run_id}") except Exception as e: print(f"[WARN] Failed to create analysis run record: {e}") if not passed_clusters: print("[WARN] No clusters passed Layer 2 filter!") return { "total_analyzed": 0, "opportunities": [], "timestamp": datetime.now(timezone.utc).isoformat() } # Apply max_clusters limit if set (for testing/sanity runs) total_passed = len(passed_clusters) if max_clusters > 0 and len(passed_clusters) > max_clusters: passed_clusters = passed_clusters[:max_clusters] print(f"[3.0] LIMITED: Analyzing {max_clusters}/{total_passed} clusters (--max-clusters flag)") else: print(f"[3.0] Analyzing {len(passed_clusters)} clusters that passed filter...") opportunities = [] for i, cluster in enumerate(passed_clusters): # Analyze ALL passed clusters groq_cls = cluster.get("groq_classification", {}) the_question = cluster.get("the_question") or groq_cls.get("the_question") or groq_cls.get("reason") or "Unknown" print(f"\n[3.{i+1}] Analyzing: {str(the_question)[:60]}...") # Build ALL topic titles all_titles = [t["topic_name"] for t in cluster["topics"]] titles_text = "\n".join([f"- {t}" for t in all_titles]) # Build Reddit context with full data reddit_text = "No Reddit posts in this cluster." if cluster.get("reddit_posts"): reddit_lines = [] for r in cluster["reddit_posts"]: comments = r.get("num_comments", 0) score = r.get("score", 0) subreddit = r.get("subreddit", "unknown") title = r.get("title", "") selftext = r.get("selftext", "")[:200] reddit_lines.append(f"- r/{subreddit}: \"{title}\" (score: {score}, comments: {comments})") if selftext: reddit_lines.append(f" Context: {selftext}...") reddit_text = "\n".join(reddit_lines) sources = ', '.join(cluster['sources']) cross_source_count = cluster['cross_source_count'] question_signal_count = cluster.get('question_signal_count', 0) groq_classification = groq_cls.get('classification', 'Unknown') groq_score = groq_cls.get('score', 0) # Build hook_criteria section for channel-specific guidance hook_criteria_section = "" if hook_criteria: hook_criteria_section = f""" ═══════════════════════════════════════════════════════════ CHANNEL-SPECIFIC HOOK CRITERIA ═══════════════════════════════════════════════════════════ {hook_criteria} """ prompt = f"""You are a short-form video opportunity scout. Your job is to find the ONE specific detail in this cluster that would stop someone mid-scroll — then build a complete video concept around it. {channel_context} {hook_criteria_section} ═══════════════════════════════════════════════════════════ RAW MATERIAL ═══════════════════════════════════════════════════════════ CLUSTER POSTS: {titles_text} REDDIT SIGNALS: {reddit_text} SOURCES: {sources} CROSS-SOURCE COUNT: {cross_source_count} QUESTION SIGNALS: {question_signal_count} ═══════════════════════════════════════════════════════════ PHASE 1: IDENTIFY THE HOOK MECHANISM ═══════════════════════════════════════════════════════════ Every viral hook uses one of these 5 mechanisms. Identify which ONE is strongest in this cluster: 1. CURIOSITY_GAP — "Wait, what?" A specific detail that contradicts what the viewer assumes to be true. They cannot scroll past without resolving it. Example: "CAPTCHAs Don't Actually Care About Your Answers" 2. SELF_RECOGNITION — "That's so me." Viewer sees their own behavior named in the first 1.5 seconds. Creates belonging. Example: "You're Not a Night Owl. You're Afraid of Tomorrow." 3. THREAT_INDIGNATION — "They're doing WHAT to me?" Reveals a specific manipulation the viewer experiences daily. Creates righteous awareness. Example: "Why the 'X' Button on Ads is Designed to Be Impossible to Tap" 4. CONSEQUENCE_CHAIN — "Then what happens?" A simple starting point with cascading non-obvious consequences. Creates domino-effect curiosity. Example: "What Actually Happens If You Never Pay a Parking Ticket" 5. NARRATIVE_SURPRISE — "That actually happened?" A real event that sounds like fiction. Lead with the most unbelievable detail. Example: "The Pope Who Put a Corpse on Trial" Search this cluster for up to 3 candidate hooks. For each one: - Quote the exact source post title - Identify the hook mechanism - Articulate the specific "wait, what?" moment in one sentence - Rate scroll-stop strength 1-5 (5 = physically cannot scroll past this) If NO hook with scroll-stop strength 4+ exists in this cluster, say so honestly. Not every cluster contains a video. ═══════════════════════════════════════════════════════════ PHASE 2: BUILD THE CONCEPT ═══════════════════════════════════════════════════════════ Take your strongest hook and build a video concept. The concept is about that ONE specific detail, not the general theme. A. PREMISE (the video title) The viewer must encounter something they cannot explain and need to resolve. Rules: - Must contain the specific detail that creates the information gap - Must make someone with ZERO prior interest in this topic stop scrolling - Under 15 words. The hook is in specificity, not length. - NO listicle formats ("7 Ways...", "10 Things...") - NO "Why You..." followed by something that makes viewer feel broken - The viewer should feel CURIOUS, not lectured CRITICAL FORMAT RULE: Do NOT default to the "You're Not X. You're Y." template. This format has been overused across short-form content and triggers pattern fatigue in algorithmic feeds. The orienting response fires on NOVELTY — a premise that follows a predictable syntactic pattern fails at Phase 1 regardless of how good the insight is. Each premise must have a distinct sentence structure. If you find yourself starting with "You're Not..." — stop and rewrite. The same insight can always be expressed differently: Instead of: "You're Not Lazy. Your Brain Is Missing a Start Button." Try: "Your Brain Has No Start Button. That's Why You're Stuck." Instead of: "You're Not a Night Owl. You're Afraid of Tomorrow." Try: "The Real Reason You Stay Up Late Has Nothing to Do With Sleep." The insight is the same. The format is fresh. B. FIRST_FRAME What text/image appears in the first frame to maximize scroll-stop? This is the thumbnail + opening visual moment. C. TRIGGER_MAP How this concept triggers the hook mechanism: - Which of the 5 mechanisms does this use? - What specific word/phrase/image creates the trigger? - Why can't the viewer scroll past this? D. OPENING HOOK (first 10 seconds) The spoken words that OPEN the curiosity loop wider. Do NOT explain here — intensify the question. E. CORE REVEAL The satisfying explanation that closes the loop. Must be: - Genuinely surprising (not "it's complicated") - Explainable in 60+ seconds (if shorter, it's a fun fact, not a video) - Leave viewer feeling smarter, not anxious F. DEPTH CHECK Can this sustain a 90-120 second video? Must have: - A reveal that takes at least 60 seconds to properly explain - At least one additional layer of surprise beyond the initial hook - Enough substance that viewer learns something real, not just a fun fact ═══════════════════════════════════════════════════════════ PHASE 3: SCORING & VERDICT ═══════════════════════════════════════════════════════════ SCORING (1-5 each, be brutally honest): SCROLL_STOP_POWER (25%): Can this premise make someone physically stop scrolling? - 5: Impossible to scroll past. "Wait, WHAT?" - 4: Very strong. Most people would stop. - 3: Interesting but skippable. DANGER ZONE — this is where mediocre content lives. - 2: Mild interest. "Huh, cool I guess." - 1: No hook. Educational but not compelling. COMPLETION_PROBABILITY (25%): Will viewers watch to the end? - 5: The reveal is so satisfying they'll watch twice - 4: Strong narrative pull, clear payoff - 3: Might lose people in the middle - 2: Hook is better than the content - 1: No reason to keep watching SHARE_SAVE_POTENTIAL (15%): Will viewers share this or save for later? - 5: "I need to text this to someone RIGHT NOW" - 4: Will send to specific friend who would care - 3: Might share if asked - 2: Unlikely to share - 1: No share impulse DEMAND_SIGNAL (15%): Is there evidence people care about this? - 5: High engagement on the specific detail (not just the theme) - 4: Strong Reddit/cross-source signals - 3: Moderate engagement - 2: Weak signals - 1: No evidence of demand VISUAL_POTENTIAL (10%): Can this be shown, not just told? - 5: Rich visual reveal (mechanism, cross-section, transformation) - 4: Good visual support available - 3: Talking head viable - 2: Hard to visualize - 1: Pure audio content EVERGREEN_POTENTIAL (10%): Will this work in 6 months? - 5: Permanent truth about how the world works - 4: Long-lasting relevance - 3: Moderate shelf life - 2: Tied to current events - 1: Expires quickly WEIGHTED SCORE = (scroll_stop * 0.25) + (completion * 0.25) + (share_save * 0.15) + (demand * 0.15) + (visual * 0.10) + (evergreen * 0.10) ═══════════════════════════════════════════════════════════ MAKE_NOW GATES (ALL must pass for MAKE_NOW verdict) ═══════════════════════════════════════════════════════════ Before giving a MAKE_NOW verdict, answer these 6 yes/no questions: 1. SCROLL_STOP_TEST: Does the premise contain a specific detail that creates an unresolved tension in under 3 seconds? (Not a theme — a detail) 2. UNIVERSAL_ACCESS_TEST: Would someone with zero interest in this topic still click? (Transcends subject matter) 3. DEPTH_TEST: Does the reveal require 60+ seconds to properly explain? (Not a fun fact — a video) 4. SHARE_TEST: Would someone text this premise to a friend unprompted? (Social currency) 5. SATISFACTION_TEST: Does the reveal leave the viewer feeling smarter, empowered, or validated — NOT anxious, guilty, or lectured? (Positive emotional payoff) 6. CHANNEL_FIT_TEST: Does this directly exploit this channel's specific itch and competitive gap? (Not just "good content" — good content FOR THIS CHANNEL) VERDICT RULES: - MAKE_NOW: All 6 gates pass. Weighted score 4.0+. Expected rate: 10-15% of clusters. - WORTH_MAKING: 4-5 gates pass. Weighted score 3.0-3.9. Expected rate: 30-40%. - NEEDS_RESEARCH: Hook is promising but facts need verification. Expected rate: 15-25%. - SKIP: Fewer than 4 gates pass OR no genuine hook exists. Expected rate: 25-35%. AUTOMATIC SKIP — Emotional Framing Guard: Assign SKIP verdict immediately if ANY of these apply: - Premise frames the viewer's existing behavior as pathological or broken - Viewer would feel WORSE about themselves after reading the title (not during — after) - Title uses shame as the hook mechanism (e.g., "You're Not Intense. You're Empty.") - Title suggests the viewer is fundamentally defective rather than curious ═══════════════════════════════════════════════════════════ OUTPUT FORMAT (JSON) ═══════════════════════════════════════════════════════════ Respond in JSON with these exact fields: {{ "hook_candidates": [ {{ "source_post_title": "exact Reddit post title", "hook_mechanism": "CURIOSITY_GAP|SELF_RECOGNITION|THREAT_INDIGNATION|CONSEQUENCE_CHAIN|NARRATIVE_SURPRISE", "the_detail": "the specific surprising detail in one sentence", "scroll_stop_strength": 1-5 }} ], "best_hook_index": 0, "honest_assessment": "Is this cluster genuinely compelling or just analytically viable?", "premise": "the video title built around the specific detail", "first_frame": "what text/image appears in first frame", "trigger_map": {{ "mechanism": "which of the 5 mechanisms", "trigger_phrase": "the specific word/phrase/image", "why_irresistible": "why viewer can't scroll past" }}, "opening_hook": "first 10 seconds spoken", "core_reveal": "the satisfying explanation", "depth_check": "can this sustain 90-120 seconds? why?", "emotional_payoff": "smarter|empowered|validated|relieved|amused", "target_audience": "specific moment when someone would watch this", "scores": {{ "scroll_stop_power": {{"score": 0, "reasoning": "..."}}, "completion_probability": {{"score": 0, "reasoning": "..."}}, "share_save_potential": {{"score": 0, "reasoning": "..."}}, "demand_signal": {{"score": 0, "reasoning": "..."}}, "visual_potential": {{"score": 0, "reasoning": "..."}}, "evergreen_potential": {{"score": 0, "reasoning": "..."}} }}, "weighted_score": 0.0, "make_now_gates": {{ "scroll_stop_test": true|false, "universal_access_test": true|false, "depth_test": true|false, "share_test": true|false, "satisfaction_test": true|false, "channel_fit_test": true|false, "gates_passed": 0 }}, "verdict": "MAKE_NOW|WORTH_MAKING|NEEDS_RESEARCH|SKIP", "verdict_reasoning": "one sentence on why", "source_post_title": "the specific post that inspired this concept", "suggested_title": "same as premise (backward compat)", "structure": "3-4 section outline for the video" }}""" try: # Use Claude CLI with Opus model via stdin result = subprocess.run( ["claude", "--print", "--model", "opus", "-p", "-"], input=prompt, capture_output=True, text=True, timeout=300 ) if result.returncode != 0: print(f" [ERROR] Claude CLI failed: {result.stderr[:100]}") continue response_text = result.stdout.strip() # Parse JSON response try: import re json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: analysis = json.loads(json_match.group()) else: analysis = json.loads(response_text) # Add cluster context to the analysis analysis["cluster_id"] = cluster["cluster_id"] analysis["topic_count"] = cluster["topic_count"] analysis["sources"] = cluster["sources"] analysis["cross_source_count"] = cluster["cross_source_count"] analysis["groq_classification"] = groq_cls.get("classification") analysis["groq_score"] = groq_cls.get("score") analysis["the_question"] = the_question if channel_id: analysis["channel"] = channel_id # Extract and normalize weighted_score using robust parser weighted = extract_weighted_score(analysis) analysis["weighted_score"] = weighted # Backward compat: ensure suggested_title exists (maps from premise) if "premise" in analysis and "suggested_title" not in analysis: analysis["suggested_title"] = analysis["premise"] elif "suggested_title" not in analysis: analysis["suggested_title"] = "" # Validate verdict (downgrade MAKE_NOW if gates don't pass) original_verdict = analysis.get("verdict", "SKIP") validated_verdict = validate_verdict(analysis) if validated_verdict != original_verdict: analysis["verdict"] = validated_verdict analysis["verdict_validated"] = True analysis["original_verdict"] = original_verdict opportunities.append(analysis) verdict = analysis.get("verdict", "Unknown") title = analysis.get("suggested_title", "Unknown")[:50] print(f" → {verdict} (score: {weighted:.1f}) - {title}...") except json.JSONDecodeError as e: print(f" [WARN] Could not parse JSON response") opportunities.append({ "theme": groq_cls.get("theme", "Unknown"), "raw_response": response_text[:2000], "parse_error": str(e), "cluster_id": cluster["cluster_id"] }) except subprocess.TimeoutExpired: print(f" [ERROR] Claude CLI timed out") except Exception as e: print(f" [ERROR] {str(e)}") # Sort by weighted score using the robust extractor scored_opps = [o for o in opportunities if "weighted_score" in o] scored_opps.sort(key=lambda x: extract_weighted_score(x), reverse=True) # Add unscored ones at the end unscored_opps = [o for o in opportunities if "weighted_score" not in o] # Assign verdicts by relative rank (enforces target distribution) scored_opps = assign_verdicts_by_rank(scored_opps) new_opportunities = scored_opps + unscored_opps # Load existing data for merge (unless replace mode) output_path = get_opportunities_path(channel_id) existing_data = load_existing_opportunities(channel_id) if not replace_mode else {"opportunities": []} # Merge new opportunities with existing merged_opportunities = merge_opportunities(existing_data, new_opportunities, replace_mode) # Sort merged results by weighted score merged_opportunities.sort(key=lambda x: extract_weighted_score(x), reverse=True) # Log merge stats existing_count = len(existing_data.get("opportunities", [])) new_count = len(new_opportunities) final_count = len(merged_opportunities) if not replace_mode and existing_count > 0: print(f"\n[3.X] Merge: {existing_count} existing + {new_count} new/updated → {final_count} total") else: print(f"\n[3.X] Generated {final_count} opportunities") result = { "channel_id": channel_id, "total_analyzed": len(passed_clusters), "total_opportunities": final_count, "opportunities": merged_opportunities, "timestamp": datetime.now(timezone.utc).isoformat() } # Backup existing file before writing backup_path = backup_opportunities_file(channel_id) if backup_path: print(f"[3.X] Backup: {backup_path}") # Save Layer 3 output (channel-specific if channel_id provided) with open(output_path, "w") as f: json.dump(result, f, indent=2, cls=NumpyEncoder) print(f"[3.X] Saved: {output_path}") # Insert opportunities into database if analysis_run_id and merged_opportunities: try: opp_ids = insert_opportunities_batch( channel_id=channel_id, analysis_run_id=analysis_run_id, opportunities=merged_opportunities, prompt_version="v3" ) print(f"[3.X] DB: Inserted {len(opp_ids)} opportunities (run_id: {analysis_run_id})") # Update analysis run with completion update_analysis_run( run_id=analysis_run_id, opportunities_found=len(merged_opportunities), status="completed" ) except Exception as e: print(f"[WARN] DB insert failed: {e}") if analysis_run_id: update_analysis_run( run_id=analysis_run_id, status="failed", error_message=str(e) ) # Store analysis_run_id in result for downstream layers result["analysis_run_id"] = analysis_run_id return result # ============================================================================= # LAYER 3.5: DEEP RESEARCH INTEGRATION # ============================================================================= RESEARCH_SERVICE_URL = "http://127.0.0.1:8100" RESEARCH_POLL_INTERVAL = 30 # seconds RESEARCH_MAX_POLL_TIME = 900 # 15 minutes def check_research_service() -> bool: """Check if the research service is running.""" import requests try: response = requests.get(f"{RESEARCH_SERVICE_URL}/health", timeout=5) return response.status_code == 200 except Exception: return False def construct_research_query(opportunity: Dict[str, Any], channel_description: str, max_length: int = 380) -> str: """ Construct a web search query from Layer 3 output. Goes to Tavily (web search), so needs actual search terms, not instructions or creative titles. """ premise = str(opportunity.get("premise", opportunity.get("suggested_title", "Unknown")))[:80] reveal = str(opportunity.get("core_reveal", ""))[:120] itch = str(opportunity.get("the_itch", opportunity.get("opening_hook", "")))[:100] query = premise if reveal and reveal not in ("", "N/A", "None"): query += f" {reveal}" elif itch and itch not in ("", "N/A", "None"): query += f" {itch}" if len(query) > max_length: query = query[:max_length - 3] + "..." return query def submit_research_job(topic: str, thesis: str = None) -> Optional[str]: """Submit an async research job. Returns job_id or None on failure.""" import requests try: payload = {"topic": topic} if thesis: payload["thesis"] = thesis response = requests.post( f"{RESEARCH_SERVICE_URL}/research/async", json=payload, timeout=30 ) if response.status_code == 200: return response.json().get("job_id") else: print(f" [ERROR] Research service returned {response.status_code}: {response.text[:100]}") return None except Exception as e: print(f" [ERROR] Failed to submit research job: {e}") return None def poll_research_job(job_id: str) -> Optional[Dict[str, Any]]: """Poll for research job completion. Returns result or None on failure/timeout.""" import requests import time start_time = time.time() while True: elapsed = time.time() - start_time if elapsed > RESEARCH_MAX_POLL_TIME: print(f" [TIMEOUT] Research job {job_id} exceeded {RESEARCH_MAX_POLL_TIME}s") return None try: response = requests.get(f"{RESEARCH_SERVICE_URL}/research/{job_id}", timeout=30) if response.status_code != 200: print(f" [ERROR] Poll failed: {response.status_code}") return None data = response.json() status = data.get("status") if status == "completed": return data.get("result", {}) elif status == "failed": print(f" [FAILED] Research job failed: {data.get('error', 'Unknown error')}") return None # Still running, wait and poll again time.sleep(RESEARCH_POLL_INTERVAL) except Exception as e: print(f" [ERROR] Poll error: {e}") return None def layer35_deep_research( opportunities_data: Dict[str, Any], channel_id: str = None, top_n: int = 5, skip_research: bool = False ) -> Dict[str, Any]: """ Layer 3.5: Deep Research Integration. Automatically triggers deep research for the top N opportunities using the Hybrid Research Service API. Args: opportunities_data: Output from Layer 3 channel_id: Channel ID for context top_n: Number of top opportunities to research (default 5) skip_research: If True, skip research and just mark all as not researched Returns: Updated opportunities_data with research fields added """ print("\n" + "=" * 60) print("LAYER 3.5: DEEP RESEARCH INTEGRATION") print("=" * 60) # Get analysis_run_id for DB updates analysis_run_id = opportunities_data.get("analysis_run_id") opportunities = opportunities_data.get("opportunities", []) if not opportunities: print("[WARN] No opportunities to research") return opportunities_data # Get channel description for query construction channel_description = "" if channel_id: from channel_config import get_channel channel = get_channel(channel_id) if channel: channel_description = channel.description print(f"[3.5.0] Channel: {channel.name}") if skip_research: print("[3.5.0] Skipping research (--skip-research flag)") for opp in opportunities: opp["research_completed"] = False opp["research_skipped"] = True return opportunities_data # Check if research service is running if not check_research_service(): print("[WARN] Research service not available at " + RESEARCH_SERVICE_URL) print(" Run: cd /home/sietch6/research-sandbox/option1-gpt-researcher && python research_service.py") print(" Marking all opportunities as research_completed: false") for opp in opportunities: opp["research_completed"] = False opp["research_error"] = "Research service not available" return opportunities_data print(f"[3.5.1] Research service available at {RESEARCH_SERVICE_URL}") print(f"[3.5.2] Will research top {top_n} opportunities (of {len(opportunities)} total)") # Sort by weighted_score and take top N scored_opps = sorted( opportunities, key=lambda x: x.get("weighted_score", 0), reverse=True ) to_research = scored_opps[:top_n] not_researched = scored_opps[top_n:] # Mark non-researched opportunities for opp in not_researched: opp["research_completed"] = False opp["research_report_path"] = None opp["research_report_content"] = None opp["research_word_count"] = None opp["research_source_count"] = None opp["research_duration_seconds"] = None opp["research_generated_at"] = None # Process research in batches of 2 (service max concurrent) import time batch_size = 2 researched_count = 0 for batch_start in range(0, len(to_research), batch_size): batch = to_research[batch_start:batch_start + batch_size] batch_num = (batch_start // batch_size) + 1 total_batches = (len(to_research) + batch_size - 1) // batch_size print(f"\n[3.5.3] Processing batch {batch_num}/{total_batches} ({len(batch)} topics)") # Submit jobs for this batch jobs = [] for opp in batch: title = opp.get("suggested_title", "Unknown")[:50] print(f" Submitting: {title}...") query = construct_research_query(opp, channel_description) thesis = str(opp.get("premise", opp.get("suggested_title", ""))) job_id = submit_research_job(query, thesis=thesis) if job_id: jobs.append((opp, job_id, query)) print(f" → Job ID: {job_id}") else: opp["research_completed"] = False opp["research_error"] = "Failed to submit research job" # Poll for completion for opp, job_id, query in jobs: title = opp.get("suggested_title", "Unknown")[:50] print(f" Polling: {title} (job {job_id})...") result = poll_research_job(job_id) if result and result.get("success"): opp["research_completed"] = True opp["research_report_path"] = result.get("report_path", "") opp["research_report_content"] = result.get("report_content", "") opp["research_word_count"] = result.get("word_count", 0) opp["research_source_count"] = result.get("source_count", 0) opp["research_duration_seconds"] = result.get("duration_seconds", 0) opp["research_generated_at"] = result.get("generated_at", "") opp["research_query"] = query researched_count += 1 print(f" ✓ Completed: {result.get('word_count', 0)} words, {result.get('source_count', 0)} sources") # Update database if we have a run ID if analysis_run_id and opp.get("cluster_id") is not None: try: update_opportunity_research_by_cluster( analysis_run_id=analysis_run_id, cluster_id=opp["cluster_id"], research_status="completed", research_data={ "report_content": opp.get("research_report_content"), "word_count": opp.get("research_word_count"), "source_count": opp.get("research_source_count"), "duration_seconds": opp.get("research_duration_seconds"), "query": opp.get("research_query"), "generated_at": opp.get("research_generated_at") } ) except Exception as e: print(f" [WARN] DB update for research failed: {e}") else: opp["research_completed"] = False opp["research_error"] = "Research job failed or timed out" print(f" ✗ Failed") # Update database with failure if analysis_run_id and opp.get("cluster_id") is not None: try: update_opportunity_research_by_cluster( analysis_run_id=analysis_run_id, cluster_id=opp["cluster_id"], research_status="failed", research_data={"error": "Research job failed or timed out"} ) except Exception: pass # Don't fail the pipeline for DB errors # Update opportunities list with researched items first opportunities_data["opportunities"] = to_research + not_researched opportunities_data["research_stats"] = { "total_opportunities": len(opportunities), "researched_count": researched_count, "top_n_requested": top_n, "research_timestamp": datetime.now(timezone.utc).isoformat() } print(f"\n[3.5.4] Research complete: {researched_count}/{top_n} successful") # Save updated output (channel-specific if channel_id provided) output_path = get_opportunities_path(channel_id) with open(output_path, "w") as f: json.dump(opportunities_data, f, indent=2, cls=NumpyEncoder) print(f"[3.5.5] Saved: {output_path}") return opportunities_data def run_research_for_opportunity(opportunity_id: str, channel_id: str = None) -> bool: """ Manually trigger research for a specific opportunity. Args: opportunity_id: The ID of the opportunity to research channel_id: Optional channel ID for context Returns: True if research was successful, False otherwise """ # Load current opportunities (channel-specific if channel_id provided) output_path = get_opportunities_path(channel_id) if not output_path.exists(): print(f"[ERROR] No opportunities file found at {output_path}") return False with open(output_path, "r") as f: data = json.load(f) opportunities = data.get("opportunities", []) # Find the opportunity target_opp = None target_idx = None for idx, opp in enumerate(opportunities): if str(opp.get("id", idx)) == str(opportunity_id) or str(idx) == str(opportunity_id): target_opp = opp target_idx = idx break if target_opp is None: print(f"[ERROR] Opportunity {opportunity_id} not found") return False if target_opp.get("research_completed"): print(f"[INFO] Opportunity already has research completed") return True # Get channel description channel_description = "" if channel_id: from channel_config import get_channel channel = get_channel(channel_id) if channel: channel_description = channel.description # Check service if not check_research_service(): print(f"[ERROR] Research service not available") return False # Submit and poll title = target_opp.get("suggested_title", "Unknown") print(f"Researching: {title}") query = construct_research_query(target_opp, channel_description) thesis = str(target_opp.get("premise", target_opp.get("suggested_title", ""))) job_id = submit_research_job(query, thesis=thesis) if not job_id: return False print(f"Job submitted: {job_id}") result = poll_research_job(job_id) if result and result.get("success"): target_opp["research_completed"] = True target_opp["research_report_path"] = result.get("report_path", "") target_opp["research_report_content"] = result.get("report_content", "") target_opp["research_word_count"] = result.get("word_count", 0) target_opp["research_source_count"] = result.get("source_count", 0) target_opp["research_duration_seconds"] = result.get("duration_seconds", 0) target_opp["research_generated_at"] = result.get("generated_at", "") target_opp["research_query"] = query # Save back opportunities[target_idx] = target_opp data["opportunities"] = opportunities with open(output_path, "w") as f: json.dump(data, f, indent=2) print(f"✓ Research completed: {result.get('word_count', 0)} words") return True else: print(f"✗ Research failed") return False # ============================================================================= # MAIN EXECUTION # ============================================================================= def safe_str(val, max_len: int = 80) -> str: """Safely convert a value to a truncated string, handling dicts and None.""" if val is None: return "N/A" if isinstance(val, dict): # Try common keys that might contain the actual value for key in ["value", "text", "content", "summary", "description", "format", "type"]: if key in val: return safe_str(val[key], max_len) # Otherwise just stringify the dict return str(val)[:max_len] if isinstance(val, list): # For lists (like structure), join first few items if val and isinstance(val[0], dict): # Extract section titles from structure list sections = [] for v in val[:4]: title = v.get("section", v.get("title", v.get("name", ""))) if title: sections.append(str(title)[:25]) if sections: return " → ".join(sections)[:max_len] return ", ".join(str(v)[:20] for v in val[:4])[:max_len] return str(val)[:max_len] def print_summary(opportunities_data: Dict[str, Any]): """Print a human-readable summary of deep analysis results.""" print("\n" + "=" * 60) print("DEEP ANALYSIS SUMMARY") print("=" * 60) opportunities = opportunities_data.get("opportunities", []) if not opportunities: print("No opportunities analyzed.") return # Separate by 4 verdict tiers make_now = [o for o in opportunities if "MAKE_NOW" in str(o.get("verdict", "")).upper()] worth_making = [o for o in opportunities if "WORTH_MAKING" in str(o.get("verdict", "")).upper()] needs_research = [o for o in opportunities if "NEEDS_RESEARCH" in str(o.get("verdict", "")).upper()] skip = [o for o in opportunities if "SKIP" in str(o.get("verdict", "")).upper()] total = len(opportunities) print(f"\nVERDICT BREAKDOWN (4 tiers):") print(f" MAKE_NOW: {len(make_now)} ({len(make_now)/total*100:.0f}%)") print(f" WORTH_MAKING: {len(worth_making)} ({len(worth_making)/total*100:.0f}%)") print(f" NEEDS_RESEARCH: {len(needs_research)} ({len(needs_research)/total*100:.0f}%)") print(f" SKIP: {len(skip)} ({len(skip)/total*100:.0f}%)") if make_now: print("\n" + "-" * 60) print("MAKE NOW (immediate opportunity)") print("-" * 60) for i, opp in enumerate(make_now[:5]): if "raw_response" in opp: print(f"\n[{i+1}] Parse error - raw response saved") continue theme = safe_str(opp.get("theme"), 50) title = safe_str(opp.get("suggested_title"), 70) score = opp.get("weighted_score", 0) if isinstance(score, dict): score = score.get("score", score.get("value", 0)) try: score = float(score) except (ValueError, TypeError): score = 0.0 fmt = safe_str(opp.get("suggested_format"), 60) hook = safe_str(opp.get("opening_hook"), 100) structure = safe_str(opp.get("structure"), 80) print(f"\n[{i+1}] {theme} (Score: {score:.1f}/5.0)") print(f" Title: {title}") print(f" Format: {fmt}") if hook and hook != "N/A": print(f" Hook: {hook}...") if structure and structure != "N/A": print(f" Structure: {structure}") if worth_making: print("\n" + "-" * 60) print("WORTH MAKING (no rush)") print("-" * 60) for i, opp in enumerate(worth_making[:5]): if "raw_response" in opp: continue title = safe_str(opp.get("suggested_title"), 70) score = opp.get("weighted_score", 0) if isinstance(score, dict): score = score.get("score", score.get("value", 0)) try: score = float(score) except (ValueError, TypeError): score = 0.0 print(f" [{i+1}] ({score:.1f}) {title}") if needs_research: print("\n" + "-" * 60) print("NEEDS RESEARCH") print("-" * 60) for i, opp in enumerate(needs_research[:3]): title = safe_str(opp.get("suggested_title"), 50) research_needed = safe_str(opp.get("research_needed"), 80) if research_needed == "N/A": research_needed = safe_str(opp.get("verdict_reasoning"), 80) score = opp.get("weighted_score", 0) if isinstance(score, dict): score = score.get("score", score.get("value", 0)) try: score = float(score) except (ValueError, TypeError): score = 0.0 print(f" [{i+1}] ({score:.1f}) {title}") print(f" NEEDS: {research_needed}") if skip: print("\n" + "-" * 60) print("SKIP") print("-" * 60) for i, opp in enumerate(skip[:3]): title = safe_str(opp.get("suggested_title"), 50) reason = safe_str(opp.get("verdict_reasoning"), 80) print(f" [{i+1}] {title}") print(f" Why: {reason}") def main(): """Run the 3-layer analysis pipeline.""" import argparse parser = argparse.ArgumentParser(description="3-Layer Trend Analysis Pipeline") parser.add_argument( "--since-last-run", action="store_true", help="Only analyze items from the most recent collection run" ) parser.add_argument( "--since-run", type=int, help="Only analyze items from runs >= this run ID" ) parser.add_argument( "--limit", type=int, help="Maximum number of topics to load" ) parser.add_argument( "--channel", type=str, help="Only analyze items tagged to this channel" ) parser.add_argument( "--list-channels", action="store_true", help="List available channels and exit" ) parser.add_argument( "--top-n", type=int, default=1, # Was 5, then 2 - using 1 per channel for now help="Number of top opportunities to auto-research in Layer 3.5 (default: 1)" ) parser.add_argument( "--skip-research", action="store_true", help="Skip Layer 3.5 deep research step" ) parser.add_argument( "--replace", action="store_true", help="Replace entire opportunities file instead of merging (destructive, use with caution)" ) parser.add_argument( "--research-only", type=str, metavar="OPPORTUNITY_ID", help="Run research for a specific opportunity ID only (skip Layers 1-3)" ) parser.add_argument( "--max-clusters", type=int, default=0, help="Limit Layer 3 analysis to N clusters (0 = unlimited, for testing use 3-5)" ) parser.add_argument( "--generate-videos", action="store_true", help="Run Layers 4a→4b→4c video generation for all researched opportunities" ) parser.add_argument( "--generate-video", type=int, metavar="ID", help="Run Layers 4a→4b→4c for a specific opportunity by index (0-based)" ) parser.add_argument( "--prompt-version", type=str, default="v1", help="Prompt version for video generation (default: v1)" ) args = parser.parse_args() # Handle --list-channels if args.list_channels: from channel_config import get_enabled_channels, get_channel_stats print("Available channels:") for ch_id, ch in get_enabled_channels().items(): stats = get_channel_stats(ch_id) print(f" {ch_id}: {ch.name}") print(f" Items tagged: {stats['item_count']}") if stats['last_analysis']: print(f" Last analysis: {stats['last_analysis']}") sys.exit(0) # Handle --research-only mode if args.research_only: print("=" * 60) print("MANUAL RESEARCH TRIGGER") print("=" * 60) success = run_research_for_opportunity(args.research_only, channel_id=args.channel) sys.exit(0 if success else 1) # Handle --generate-videos or --generate-video (standalone video generation) if args.generate_videos or args.generate_video is not None: from video_generation import ( run_video_generation_for_opportunities, get_latest_opportunities_file, save_opportunities_timestamped ) # Find the channel-specific opportunities file input_path = get_latest_opportunities_file(args.channel) if not input_path: print(f"[ERROR] No opportunities file found for channel '{args.channel}'. Run analysis first.") sys.exit(1) print(f"[INFO] Loading from: {input_path}") with open(input_path, "r") as f: opportunities_data = json.load(f) # Run video generation opportunities_data = run_video_generation_for_opportunities( opportunities_data=opportunities_data, channel_id=args.channel, specific_id=args.generate_video, prompt_version=args.prompt_version, skip_research=args.skip_research ) # Save to timestamped file (also updates latest for dashboard) saved_path = save_opportunities_timestamped(opportunities_data, args.channel) print(f"\n[OK] Video generation complete.") print(f" Saved: {saved_path}") print(f" Latest: {get_opportunities_path(args.channel)}") sys.exit(0) print("=" * 60) print("TREND ANALYSIS PIPELINE") if args.channel: from channel_config import get_channel ch = get_channel(args.channel) if ch: print(f"Channel: {ch.name}") else: print(f"[ERROR] Unknown channel: {args.channel}") sys.exit(1) print("=" * 60) print(f"Started: {datetime.now(timezone.utc).isoformat()}") # Pre-flight checks check_dependencies() check_api_keys() # Ensure DB schema is up to date try: run_migration() except Exception as e: print(f"[WARN] Migration check failed: {e}") # Determine run filter since_run_id = None if args.since_last_run: since_run_id = get_last_run_id() print(f"\n[MODE] Analyzing only from latest run (run_id: {since_run_id})") elif args.since_run: since_run_id = args.since_run print(f"\n[MODE] Analyzing from run_id >= {since_run_id}") # Load data print("\n[0] Loading trends from database...") trends = load_trends_from_db( limit=args.limit, since_run_id=since_run_id, channel_id=args.channel ) print(f" Loaded {len(trends)} unique topics") if args.channel: print(f" (filtered to channel: {args.channel})") # Show signal type breakdown signal_counts = {} for t in trends: st = t.get("signal_type", "unknown") signal_counts[st] = signal_counts.get(st, 0) + 1 print(f" Signal types: {signal_counts}") if len(trends) < 10: print("[ERROR] Not enough data to analyze. Run the collection pipeline first.") sys.exit(1) # Layer 1: Embed + Cluster clusters_data = layer1_embed_and_cluster(trends) # Layer 2: Groq Classification + Filtering classified_data = layer2_classify_and_filter(clusters_data) # Layer 3: Claude Deep Analysis (with channel context if specified) opportunities_data = layer3_deep_analysis( classified_data, channel_id=args.channel, max_clusters=args.max_clusters, replace_mode=args.replace ) # Layer 3.5: Deep Research Integration opportunities_data = layer35_deep_research( opportunities_data, channel_id=args.channel, top_n=args.top_n, skip_research=args.skip_research ) # Summary print_summary(opportunities_data) # Generate dashboard data print("\n[4] Generating dashboard data...") try: cmd = ["python3", "generate_dashboard_data.py"] if args.channel: cmd.extend(["--channel", args.channel]) result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: print(" Dashboard data updated successfully") else: print(f" Warning: Dashboard generation failed: {result.stderr}") except Exception as e: print(f" Warning: Could not generate dashboard data: {e}") print("\n" + "=" * 60) print("ANALYSIS COMPLETE") print("=" * 60) print("Output files:") print(f" - {OUTPUT_DIR}/analysis_clusters.json") print(f" - {OUTPUT_DIR}/analysis_classified.json") print(f" - {get_opportunities_path(args.channel)}") print(f" - {OUTPUT_DIR}/dashboard_data.json") print(f"\nDashboard: file://{OUTPUT_DIR}/dashboard.html") if __name__ == "__main__": main()