371 lines
14 KiB
Python
371 lines
14 KiB
Python
"""Executive Summary Generator for YouTube video summaries.
|
|
|
|
This service creates professional executive summaries suitable for business leaders
|
|
and decision-makers, focusing on key insights, business value, and actionable items.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
from dataclasses import dataclass
|
|
|
|
from ..services.deepseek_service import DeepSeekService
|
|
from ..core.exceptions import ServiceError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ExecutiveMetrics:
|
|
"""Key metrics extracted for executive summary."""
|
|
duration_minutes: int
|
|
word_count: int
|
|
main_topics: List[str]
|
|
sentiment_score: float
|
|
complexity_level: str
|
|
confidence_score: float
|
|
|
|
|
|
@dataclass
|
|
class ExecutiveSummary:
|
|
"""Executive summary data structure."""
|
|
overview: str
|
|
key_metrics: ExecutiveMetrics
|
|
business_value: Optional[str]
|
|
action_items: List[str]
|
|
strategic_implications: List[str]
|
|
sentiment_analysis: Dict[str, Any]
|
|
processing_time_seconds: float
|
|
created_at: datetime
|
|
|
|
|
|
class ExecutiveSummaryGenerator:
|
|
"""Service for generating professional executive summaries."""
|
|
|
|
def __init__(self, ai_service: Optional[DeepSeekService] = None):
|
|
"""Initialize executive summary generator.
|
|
|
|
Args:
|
|
ai_service: AI service for summary generation
|
|
"""
|
|
self.ai_service = ai_service or DeepSeekService()
|
|
|
|
# Executive summary configuration
|
|
self.max_overview_paragraphs = 3
|
|
self.target_reading_time = "2-3 minutes"
|
|
self.executive_temperature = 0.3 # Lower temperature for consistency
|
|
self.max_tokens = 1200
|
|
|
|
logger.info("ExecutiveSummaryGenerator initialized")
|
|
|
|
async def generate_executive_summary(
|
|
self,
|
|
content: str,
|
|
video_title: str = "",
|
|
video_duration_seconds: int = 0,
|
|
summary_type: str = "business"
|
|
) -> ExecutiveSummary:
|
|
"""Generate executive summary from video content.
|
|
|
|
Args:
|
|
content: Video transcript or summary content
|
|
video_title: Title of the video
|
|
video_duration_seconds: Video duration in seconds
|
|
summary_type: Type of executive summary (business, strategic, technical)
|
|
|
|
Returns:
|
|
Executive summary object
|
|
"""
|
|
start_time = datetime.now()
|
|
|
|
if not content or len(content.strip()) < 100:
|
|
raise ServiceError("Content too short for executive summary generation")
|
|
|
|
try:
|
|
# Extract metrics first
|
|
metrics = await self._extract_content_metrics(
|
|
content, video_duration_seconds
|
|
)
|
|
|
|
# Generate executive overview
|
|
overview = await self._generate_executive_overview(
|
|
content, video_title, summary_type, metrics
|
|
)
|
|
|
|
# Extract business value and strategic implications
|
|
business_analysis = await self._analyze_business_value(content, summary_type)
|
|
|
|
# Generate action items
|
|
action_items = await self._generate_action_items(content, summary_type)
|
|
|
|
processing_time = (datetime.now() - start_time).total_seconds()
|
|
|
|
return ExecutiveSummary(
|
|
overview=overview,
|
|
key_metrics=metrics,
|
|
business_value=business_analysis.get("business_value"),
|
|
action_items=action_items,
|
|
strategic_implications=business_analysis.get("strategic_implications", []),
|
|
sentiment_analysis=business_analysis.get("sentiment_analysis", {}),
|
|
processing_time_seconds=processing_time,
|
|
created_at=datetime.now()
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error generating executive summary: {e}")
|
|
raise ServiceError(f"Executive summary generation failed: {str(e)}")
|
|
|
|
async def _extract_content_metrics(
|
|
self,
|
|
content: str,
|
|
duration_seconds: int
|
|
) -> ExecutiveMetrics:
|
|
"""Extract key metrics from content."""
|
|
try:
|
|
system_prompt = """You are an analytical assistant extracting key metrics from video content.
|
|
|
|
Analyze the content and extract:
|
|
- Main topics (3-5 key themes)
|
|
- Sentiment score (0.0 to 1.0, where 0.5 is neutral)
|
|
- Complexity level (beginner/intermediate/advanced)
|
|
- Confidence score (how clear/actionable the content is, 0.0-1.0)
|
|
|
|
Return ONLY valid JSON with this exact structure:
|
|
{
|
|
"main_topics": ["topic1", "topic2", "topic3"],
|
|
"sentiment_score": 0.7,
|
|
"complexity_level": "intermediate",
|
|
"confidence_score": 0.8
|
|
}"""
|
|
|
|
response = await self.ai_service.generate_response(
|
|
prompt=f"Analyze this content:\n\n{content[:3000]}",
|
|
system_prompt=system_prompt,
|
|
temperature=0.1,
|
|
max_tokens=300
|
|
)
|
|
|
|
import json
|
|
try:
|
|
metrics_data = json.loads(response)
|
|
except json.JSONDecodeError:
|
|
# Fallback to manual parsing if JSON fails
|
|
metrics_data = {
|
|
"main_topics": ["Content Analysis", "Key Insights"],
|
|
"sentiment_score": 0.6,
|
|
"complexity_level": "intermediate",
|
|
"confidence_score": 0.7
|
|
}
|
|
|
|
return ExecutiveMetrics(
|
|
duration_minutes=max(1, duration_seconds // 60),
|
|
word_count=len(content.split()),
|
|
main_topics=metrics_data.get("main_topics", [])[:5],
|
|
sentiment_score=metrics_data.get("sentiment_score", 0.6),
|
|
complexity_level=metrics_data.get("complexity_level", "intermediate"),
|
|
confidence_score=metrics_data.get("confidence_score", 0.7)
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error extracting metrics, using defaults: {e}")
|
|
# Return default metrics if extraction fails
|
|
return ExecutiveMetrics(
|
|
duration_minutes=max(1, duration_seconds // 60),
|
|
word_count=len(content.split()),
|
|
main_topics=["Content Analysis"],
|
|
sentiment_score=0.6,
|
|
complexity_level="intermediate",
|
|
confidence_score=0.7
|
|
)
|
|
|
|
async def _generate_executive_overview(
|
|
self,
|
|
content: str,
|
|
video_title: str,
|
|
summary_type: str,
|
|
metrics: ExecutiveMetrics
|
|
) -> str:
|
|
"""Generate the main executive overview paragraphs."""
|
|
|
|
executive_prompts = {
|
|
"business": """You are a senior business analyst creating an executive summary for leadership.
|
|
|
|
Focus on business impact, strategic implications, and decision-making insights.
|
|
Write EXACTLY 2-3 paragraphs that a CEO or executive would find valuable.
|
|
Use professional business language and focus on outcomes and opportunities.""",
|
|
|
|
"strategic": """You are a strategic consultant creating a high-level summary for senior leadership.
|
|
|
|
Focus on strategic positioning, competitive advantages, and long-term implications.
|
|
Write EXACTLY 2-3 paragraphs with strategic depth and market context.""",
|
|
|
|
"technical": """You are a CTO creating a technical executive summary for senior leadership.
|
|
|
|
Focus on technical implications, innovation opportunities, and implementation considerations.
|
|
Write EXACTLY 2-3 paragraphs that translate technical concepts for executive audiences."""
|
|
}
|
|
|
|
system_prompt = executive_prompts.get(summary_type, executive_prompts["business"])
|
|
|
|
context_info = f"""
|
|
Video: {video_title}
|
|
Duration: {metrics.duration_minutes} minutes
|
|
Complexity: {metrics.complexity_level}
|
|
Main Topics: {', '.join(metrics.main_topics[:3])}
|
|
"""
|
|
|
|
prompt = f"""{context_info}
|
|
|
|
Content to analyze:
|
|
{content[:4000]}
|
|
|
|
Create a professional executive summary that provides clear value to senior leadership.
|
|
Focus on insights they can act upon and decisions they need to make."""
|
|
|
|
response = await self.ai_service.generate_response(
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
temperature=self.executive_temperature,
|
|
max_tokens=600
|
|
)
|
|
|
|
return response.strip()
|
|
|
|
async def _analyze_business_value(
|
|
self,
|
|
content: str,
|
|
summary_type: str
|
|
) -> Dict[str, Any]:
|
|
"""Analyze business value and strategic implications."""
|
|
|
|
system_prompt = """You are a business strategy consultant analyzing content for business value.
|
|
|
|
Extract:
|
|
- Business value proposition (what value this provides)
|
|
- Strategic implications (3-4 key strategic insights)
|
|
- Sentiment analysis (positive/neutral/negative with confidence)
|
|
|
|
Return ONLY valid JSON:
|
|
{
|
|
"business_value": "Clear value statement",
|
|
"strategic_implications": ["implication1", "implication2", "implication3"],
|
|
"sentiment_analysis": {
|
|
"sentiment": "positive",
|
|
"confidence": 0.8,
|
|
"key_indicators": ["indicator1", "indicator2"]
|
|
}
|
|
}"""
|
|
|
|
response = await self.ai_service.generate_response(
|
|
prompt=f"Analyze business value:\n\n{content[:3000]}",
|
|
system_prompt=system_prompt,
|
|
temperature=0.2,
|
|
max_tokens=500
|
|
)
|
|
|
|
try:
|
|
import json
|
|
return json.loads(response)
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"business_value": "Analysis provides strategic insights for decision-making",
|
|
"strategic_implications": ["Requires further analysis", "Consider implementation impact"],
|
|
"sentiment_analysis": {
|
|
"sentiment": "neutral",
|
|
"confidence": 0.6,
|
|
"key_indicators": ["Mixed signals"]
|
|
}
|
|
}
|
|
|
|
async def _generate_action_items(
|
|
self,
|
|
content: str,
|
|
summary_type: str
|
|
) -> List[str]:
|
|
"""Generate specific action items for executives."""
|
|
|
|
system_prompt = """You are an executive consultant creating actionable next steps.
|
|
|
|
Generate 3-5 specific, actionable items that executives can delegate or act upon.
|
|
Each action should be:
|
|
- Specific and measurable
|
|
- Assignable to a person or team
|
|
- Have clear business value
|
|
|
|
Return ONLY a JSON array of strings:
|
|
["Action item 1", "Action item 2", "Action item 3"]"""
|
|
|
|
response = await self.ai_service.generate_response(
|
|
prompt=f"Create action items from:\n\n{content[:3000]}",
|
|
system_prompt=system_prompt,
|
|
temperature=0.3,
|
|
max_tokens=400
|
|
)
|
|
|
|
try:
|
|
import json
|
|
action_items = json.loads(response)
|
|
return action_items if isinstance(action_items, list) else []
|
|
except json.JSONDecodeError:
|
|
# Fallback action items
|
|
return [
|
|
"Review key findings with relevant stakeholders",
|
|
"Assess implementation feasibility and resource requirements",
|
|
"Develop action plan with timeline and ownership"
|
|
]
|
|
|
|
async def generate_metadata_header(
|
|
self,
|
|
summary: ExecutiveSummary,
|
|
video_title: str,
|
|
video_url: str = ""
|
|
) -> str:
|
|
"""Generate professional metadata header for exports."""
|
|
|
|
header = f"""# Executive Summary: {video_title}
|
|
|
|
**Analysis Date**: {summary.created_at.strftime("%B %d, %Y")}
|
|
**Processing Time**: {summary.processing_time_seconds:.1f} seconds
|
|
**Content Duration**: {summary.key_metrics.duration_minutes} minutes
|
|
**Word Count**: {summary.key_metrics.word_count:,} words
|
|
**Complexity Level**: {summary.key_metrics.complexity_level.title()}
|
|
**Confidence Score**: {summary.key_metrics.confidence_score:.1%}
|
|
|
|
"""
|
|
if video_url:
|
|
header += f"**Source**: {video_url} \n\n"
|
|
|
|
return header
|
|
|
|
async def generate_executive_footer(self, summary: ExecutiveSummary) -> str:
|
|
"""Generate professional footer with analysis metadata."""
|
|
|
|
sentiment = summary.sentiment_analysis.get("sentiment", "neutral")
|
|
sentiment_confidence = summary.sentiment_analysis.get("confidence", 0.0)
|
|
|
|
footer = f"""
|
|
|
|
---
|
|
|
|
## Analysis Metadata
|
|
|
|
**Content Sentiment**: {sentiment.title()} (confidence: {sentiment_confidence:.1%})
|
|
**Main Topics**: {', '.join(summary.key_metrics.main_topics)}
|
|
**Generated**: {summary.created_at.strftime("%Y-%m-%d %H:%M:%S")}
|
|
**Quality Score**: {summary.key_metrics.confidence_score:.1%}
|
|
|
|
*This executive summary was generated using AI analysis and is intended for strategic decision-making support.*
|
|
"""
|
|
|
|
return footer
|
|
|
|
def get_executive_summary_stats(self) -> Dict[str, Any]:
|
|
"""Get service statistics."""
|
|
return {
|
|
"service_name": "ExecutiveSummaryGenerator",
|
|
"max_overview_paragraphs": self.max_overview_paragraphs,
|
|
"target_reading_time": self.target_reading_time,
|
|
"executive_temperature": self.executive_temperature,
|
|
"max_tokens": self.max_tokens
|
|
} |