| import os |
| from pathlib import Path |
| import logging |
| from tqdm import tqdm |
| import requests |
| from src.analysis.analysis_cleaner import AnalysisCleaner |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class CreativeAnalyzer: |
| def __init__(self): |
| |
| self.api_key = os.getenv("ANTHROPIC_API_KEY") |
| if not self.api_key: |
| raise ValueError("ANTHROPIC_API_KEY not found") |
|
|
| self.api_url = "https://api.anthropic.com/v1/messages" |
| self.model = "claude-3-sonnet-20240229" |
| self.headers = { |
| "x-api-key": self.api_key, |
| "anthropic-version": "2023-06-01", |
| "content-type": "application/json" |
| } |
|
|
| |
| self.chunk_size = 6000 |
|
|
| def query_claude(self, prompt: str) -> str: |
| """Send request to Claude API with proper response handling""" |
| try: |
| payload = { |
| "model": self.model, |
| "max_tokens": 4096, |
| "messages": [{ |
| "role": "user", |
| "content": prompt |
| }] |
| } |
|
|
| response = requests.post(self.api_url, headers=self.headers, json=payload) |
|
|
| if response.status_code == 200: |
| response_json = response.json() |
| |
| if ('content' in response_json and |
| isinstance(response_json['content'], list) and |
| len(response_json['content']) > 0 and |
| 'text' in response_json['content'][0]): |
| return response_json['content'][0]['text'] |
| else: |
| logger.error("Invalid response structure") |
| logger.error(f"Response: {response_json}") |
| return None |
| else: |
| logger.error(f"API Error: {response.status_code}") |
| logger.error(f"Response: {response.text}") |
| return None |
|
|
| except Exception as e: |
| logger.error(f"Error making API request: {str(e)}") |
| logger.error("Full error details:", exc_info=True) |
| return None |
|
|
| def count_tokens(self, text: str) -> int: |
| """Estimate token count using simple word-based estimation""" |
| words = text.split() |
| return int(len(words) * 1.3) |
|
|
| def chunk_screenplay(self, text: str) -> list: |
| """Split screenplay into chunks with overlap for context""" |
| logger.info("Chunking screenplay...") |
|
|
| scenes = text.split("\n\n") |
| chunks = [] |
| current_chunk = [] |
| current_size = 0 |
| overlap_scenes = 2 |
|
|
| for scene in scenes: |
| scene_size = self.count_tokens(scene) |
|
|
| if current_size + scene_size > self.chunk_size and current_chunk: |
| overlap = current_chunk[-overlap_scenes:] if len(current_chunk) > overlap_scenes else current_chunk |
| chunks.append("\n\n".join(current_chunk)) |
| current_chunk = overlap + [scene] |
| current_size = sum(self.count_tokens(s) for s in current_chunk) |
| else: |
| current_chunk.append(scene) |
| current_size += scene_size |
|
|
| if current_chunk: |
| chunks.append("\n\n".join(current_chunk)) |
|
|
| logger.info(f"Split screenplay into {len(chunks)} chunks with {overlap_scenes} scene overlap") |
| return chunks |
|
|
| def analyze_plot_development(self, chunk: str, previous_plot_points: str = "") -> str: |
| prompt = f"""You are a professional screenplay analyst. Building on this previous analysis: |
| {previous_plot_points} |
| |
| Continue analyzing the story's progression. Tell me what happens next, focusing on new developments and changes. Reference specific moments from this section but don't repeat what we've covered. |
| |
| Consider: |
| - How events build on what came before |
| - Their impact on story direction |
| - Changes to the narrative |
| |
| Use flowing paragraphs and support with specific examples. |
| |
| Screenplay section to analyze: |
| {chunk}""" |
|
|
| return self.query_claude(prompt) |
|
|
| def analyze_character_arcs(self, chunk: str, plot_context: str, previous_character_dev: str = "") -> str: |
| prompt = f"""You are a professional screenplay analyst. Based on these plot developments: |
| {plot_context} |
| |
| And previous character analysis: |
| {previous_character_dev} |
| |
| Continue analyzing how the characters evolve. Focus on their growth, changes, and key moments from this section. Build on, don't repeat, previous analysis. |
| |
| Consider: |
| - Character choices and consequences |
| - Relationship dynamics |
| - Internal conflicts and growth |
| |
| Use flowing paragraphs with specific examples. |
| |
| Screenplay section to analyze: |
| {chunk}""" |
|
|
| return self.query_claude(prompt) |
|
|
| def analyze_dialogue_progression(self, chunk: str, character_context: str, previous_dialogue: str = "") -> str: |
| prompt = f"""You are a professional screenplay analyst. Understanding the character context: |
| {character_context} |
| |
| And previous dialogue analysis: |
| {previous_dialogue} |
| |
| Analyze the dialogue in this section from a screenwriting perspective. What makes it effective or distinctive? |
| |
| Consider: |
| - How dialogue reveals character |
| - Subtext and meaning |
| - Character voices and patterns |
| - Impact on relationships |
| |
| Use specific dialogue examples in flowing paragraphs. |
| |
| Screenplay section to analyze: |
| {chunk}""" |
|
|
| return self.query_claude(prompt) |
|
|
| def analyze_themes(self, chunk: str, plot_context: str, character_context: str) -> str: |
| prompt = f"""You are a professional screenplay analyst. Based on these plot developments: |
| {plot_context} |
| |
| And character journeys: |
| {character_context} |
| |
| Analyze how themes develop in this section. What deeper meanings emerge? How do they connect to previous themes? |
| |
| Consider: |
| - Core ideas and messages |
| - Symbolic elements |
| - How themes connect to character arcs |
| - Social or philosophical implications |
| |
| Support with specific examples in flowing paragraphs. |
| |
| Screenplay section to analyze: |
| {chunk}""" |
|
|
| return self.query_claude(prompt) |
|
|
| def analyze_screenplay(self, screenplay_path: Path) -> bool: |
| """Main method to generate creative analysis""" |
| logger.info("Starting creative analysis") |
|
|
| try: |
| |
| with open(screenplay_path, 'r', encoding='utf-8') as file: |
| screenplay_text = file.read() |
|
|
| |
| chunks = self.chunk_screenplay(screenplay_text) |
|
|
| |
| plot_analysis = [] |
| character_analysis = [] |
| dialogue_analysis = [] |
| theme_analysis = [] |
|
|
| |
| logger.info("First Pass: Analyzing plot development") |
| with tqdm(total=len(chunks), desc="Analyzing plot") as pbar: |
| for chunk in chunks: |
| result = self.analyze_plot_development( |
| chunk, |
| "\n\n".join(plot_analysis) |
| ) |
| if result: |
| plot_analysis.append(result) |
| else: |
| logger.error("Failed to get plot analysis") |
| return False |
| pbar.update(1) |
|
|
| |
| logger.info("Second Pass: Analyzing character arcs") |
| with tqdm(total=len(chunks), desc="Analyzing characters") as pbar: |
| for chunk in chunks: |
| result = self.analyze_character_arcs( |
| chunk, |
| "\n\n".join(plot_analysis), |
| "\n\n".join(character_analysis) |
| ) |
| if result: |
| character_analysis.append(result) |
| else: |
| logger.error("Failed to get character analysis") |
| return False |
| pbar.update(1) |
|
|
| |
| logger.info("Third Pass: Analyzing dialogue") |
| with tqdm(total=len(chunks), desc="Analyzing dialogue") as pbar: |
| for chunk in chunks: |
| result = self.analyze_dialogue_progression( |
| chunk, |
| "\n\n".join(character_analysis), |
| "\n\n".join(dialogue_analysis) |
| ) |
| if result: |
| dialogue_analysis.append(result) |
| else: |
| logger.error("Failed to get dialogue analysis") |
| return False |
| pbar.update(1) |
|
|
| |
| logger.info("Fourth Pass: Analyzing themes") |
| with tqdm(total=len(chunks), desc="Analyzing themes") as pbar: |
| for chunk in chunks: |
| result = self.analyze_themes( |
| chunk, |
| "\n\n".join(plot_analysis), |
| "\n\n".join(character_analysis) |
| ) |
| if result: |
| theme_analysis.append(result) |
| else: |
| logger.error("Failed to get theme analysis") |
| return False |
| pbar.update(1) |
|
|
| |
| cleaner = AnalysisCleaner() |
| cleaned_analyses = { |
| 'plot': cleaner.clean_analysis("\n\n".join(plot_analysis)), |
| 'character': cleaner.clean_analysis("\n\n".join(character_analysis)), |
| 'dialogue': cleaner.clean_analysis("\n\n".join(dialogue_analysis)), |
| 'theme': cleaner.clean_analysis("\n\n".join(theme_analysis)) |
| } |
|
|
| |
| output_path = screenplay_path.parent / "creative_analysis.txt" |
| with open(output_path, 'w', encoding='utf-8') as f: |
| f.write("SCREENPLAY CREATIVE ANALYSIS\n\n") |
|
|
| sections = [ |
| ("PLOT PROGRESSION", cleaned_analyses['plot']), |
| ("CHARACTER ARCS", cleaned_analyses['character']), |
| ("DIALOGUE PROGRESSION", cleaned_analyses['dialogue']), |
| ("THEMATIC DEVELOPMENT", cleaned_analyses['theme']) |
| ] |
|
|
| for title, content in sections: |
| f.write(f"### {title} ###\n\n") |
| f.write(content) |
| f.write("\n\n") |
|
|
| logger.info(f"Analysis saved to: {output_path}") |
| return True |
|
|
| except Exception as e: |
| logger.error(f"Error in creative analysis: {str(e)}") |
| logger.error("Full error details:", exc_info=True) |
| return False |
