import logging import os import socket import sys from datetime import datetime from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field sys.path.append(".") # from utils.logger import ServiceLogger import logging from configs.config import OPENAI_API_KEY, OPENAI_MODEL_4O, OPENAI_MODEL_MINI from services.ppt_generator.data_classes.project import Project from services.ppt_generator.presentation_types import presentation_types from services.ppt_training.slide_processor import SlideProcessor from services.ppt_training.training_data_models import * type_of_ppt = presentation_types.keys() XCM_logger = logging.getLogger() class AnalyzePPT: def __init__(self, Project: Project): XCM_logger.info("Initialized AnalyzePPT for project: %s", Project.project_id) self.Project = Project self.slides_info = None self.PPTData: PPTData = None def process_ppt(self): XCM_logger.info("Processing PPT for project: %s", self.Project.project_id) try: if self.Project.final_ppt_path is None: XCM_logger.warning("Project doesn't have a final ppt path") return None slide_processor = SlideProcessor( self.Project.final_ppt_path, self.Project.project_id ) if self.slides_info is None: raise ValueError( "Error: self.slides_info is None, cannot proceed with analysis." ) self.slides_info = slide_processor.process_slides() slide_processor.ppt_object.slides_data = self.slides_info self.PPTData = slide_processor.ppt_object except Exception as e: # pylint: disable=broad-except XCM_logger.error( "Error processing PPT for project: %s", self.Project.project_id, exc_info=True, ) raise e # slide_processor.upload_to_database() def analyze_ppt(self): "Analyze the ppt" XCM_logger.info("Analyzing PPT for project: %s", self.Project.project_id) slides_analysis = [] # get the for slide in self.slides_info: try: if slide.is_section_divider is not True: if len(self.convert_slide_to_markdown(slide)) < 30: continue else: slides_analysis.append(self._analyze_slide(slide)) except Exception as e: # pylint: disable=broad-except XCM_logger.error( "Error analyzing slide: %s", slide.title, exc_info=True ) # get the overview of the ppt ppt_overview = self._get_ppt_overview(slides_analysis) return { "slides_analysis": slides_analysis, "ppt_overview": ppt_overview, } def _get_ppt_overview(self, slides_analysis): "Get the overview of the ppt" llm = ChatOpenAI( api_key=OPENAI_API_KEY, model=OPENAI_MODEL_4O, temperature=0.5, ) chat_prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are an investment banking expert with 20 years of experience in the {industry}. Specifically, you are an expert in the {sector} sector.", ), ( "system", "You are going to help understand what the story line in the presentation is.", ), ("human", "There are {number_of_slides} slides in the presentation."), ("human", "The slides are: {slides}"), ( "system", "You will return back the story of the presentation and how the presentation story could be improved.", ), ("system", "You will return a json object as {json_output}"), ] ) parser = JsonOutputParser(pydantic_object=PresentationAnalysis) chain = chat_prompt | llm | parser chain_output = chain.invoke( { "industry": self.Project.industry, "sector": self.Project.sector, "number_of_slides": len(slides_analysis), "slides": "\n\n".join( [ f"## Slide{i}: {slide_details['slide_overview']}" for i, slide_details in enumerate(slides_analysis) ] ), "json_output": parser.get_format_instructions(), } ) return chain_output def _analyze_slide(self, slide: SlideData): "Analyze a slide" llm = ChatOpenAI( api_key=OPENAI_API_KEY, model=OPENAI_MODEL_MINI, temperature=0.1, ) slide_as_markdown = self.convert_slide_to_markdown(slide) chat_prompt = ChatPromptTemplate.from_messages( [ ( "system", """You are an investment banking expert with 20 years of experience in the {industry}. \ Specifically, you are an expert in the {sector} sector.""", ), ( "system", """You are going to be training a new junior analyst joinning the team. \ You are going to teach them how to make a {type_of_ppt} presentation. \ You will have access to a slide, and you will provide directions on how to make it again. \ You will not use a Company's name and try to be as generic as possible so that the direction can be used for multiple companies.""", ), ("human", """Slide information: \n {slide_as_markdown}"""), ("system", "You will return a json object as {json_output}"), ] ) parser = JsonOutputParser(pydantic_object=SlideAnalysis) chain = chat_prompt | llm | parser chain_output = chain.invoke( { "industry": self.Project.industry, "sector": self.Project.sector, "type_of_ppt": self.Project.pitch_type, "slide_as_markdown": slide_as_markdown, "json_output": parser.get_format_instructions(), } ) return chain_output def convert_slide_to_markdown(self, slide: SlideData): "Convert the slide to markdown" slide_content = "" slide_content += f"## Slide title: {slide.title} \n\n" slide_content += f"## Slide content: \n" slide_content += "\n".join(slide.text) slide_content += "\n\n" if slide.tables is not None: slide_content += f"## Slide tables: \n" for table in slide.tables: table_str = "" for row in table: table_str += "| " + " | ".join(row) + " |\n" slide_content += table_str slide_content += "\n\n" return slide_content if __name__ == "__main__": project = Project.check_project_in_db("altlaw") analyze_ppt = AnalyzePPT(project) analyze_ppt.process_ppt() ppt_analysis = analyze_ppt.analyze_ppt() analyze_ppt.PPTData.PPT_analysis = PresentationAnalysis( **ppt_analysis["ppt_overview"] ) analyze_ppt.PPTData.slides_analysis = [ SlideAnalysis(**slide) for slide in ppt_analysis["slides_analysis"] ] analyze_ppt.PPTData.save_to_db()