File size: 10,910 Bytes
edb392c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 |
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):
# Initialize Claude
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"
}
# Set chunk size
self.chunk_size = 6000 # Claude handles larger chunks well
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()
# Get the message content from Claude's response
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:
# Read screenplay
with open(screenplay_path, 'r', encoding='utf-8') as file:
screenplay_text = file.read()
# Split into chunks
chunks = self.chunk_screenplay(screenplay_text)
# Initialize analyses
plot_analysis = []
character_analysis = []
dialogue_analysis = []
theme_analysis = []
# First Pass: Plot Development
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)
# Second Pass: Character Arcs
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)
# Third Pass: Dialogue Progression
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)
# Fourth Pass: Thematic Development
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)
# Clean up analyses
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))
}
# Save 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 |