"Convert structure to pptx" import asyncio import importlib # from utils.logger import ServiceLogger import logging import os from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone import pptx from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI from configs.config import OPENAI_MODEL_MINI from services.company_profile.data_classes.company_info import CompanyInfo from services.ppt_generator.company_profile_slide import CompanyProfileSlide from services.ppt_generator.data_classes.project import Section, Slide from services.ppt_generator.data_classes.slide_layout_models import * from services.ppt_generator.layouts.xcm.layouts import (CompsTable, DefaultSettings, FullPage, SideBySide, SlideWorkArea, TopBottom) from services.recommender.acquisitions_recommender import \ AcquisitionsRecommender from services.stock_info.public_comparable import PublicComparables from utils.chroma_db import ChromaDB from utils.client_check import ClientConfig from utils.dynamo_db import DynamoDB from utils.researcher.chroma_search_v2 import InternalSearch from utils.researcher.researcher_v2 import Researcher as ResearcherV2 from utils.url_parser import parsed_url XCM_logger = logging.getLogger() class CreatePPT: "Create a pptx from the structure" PPT_TEMPLATE = r"ppt_templates/ppt_template.pptx" filename = None filepath = None layout_choices = "" def __init__(self, project, main_company, client_config=None) -> None: "Initialize the pptx object." self.main_company = main_company self.client_config = client_config self.slide_work_area = SlideWorkArea() self.default_settings = DefaultSettings() if client_config: self.slide_work_area = self.client_config.slide_work_area self.default_settings = self.client_config.default_settings # self.PPT_TEMPLATE = self.client_config.ppt_template # read in the file containing the layout choices with open(self.client_config.layout_choices, "r") as f: self.layout_choices = f.read() # self.layout_choices = self.client_config.layout_choices self.prs = pptx.Presentation(self.PPT_TEMPLATE) self.project: Project = project if project.sections: self.slides_structure = project.sections else: self.slides_structure = [] self.targets = project.targets self.company = project.company_name self.company_url = project.company_url def find_appropriate_layout(self, slide): "Based on the layouts that the client has, find the appropriate layout for the slide" from services.ppt_generator.data_classes.presentation_outline import \ SlideContent chat_prompt = ChatPromptTemplate.from_messages( [ ("system", "You are an powerpoint slide desginer."), ("human", f"The slide we are creating is titled: {slide.title}"), ("human", f"The content of the slide is: {slide.content}"), ( "human", f"The information you have gathered is \n {slide.research or ''} \n", ), ("human", "Choose from the following layout options: {layout_options}"), ("system", "You will respond in a json object as {json_output}"), ] ) llm = ChatOpenAI( api_key=os.getenv("OPENAI_API_KEY"), model=OPENAI_MODEL_MINI, temperature=0.5, ) parser = JsonOutputParser(pydantic_object=SlideContent) chain = chat_prompt | llm | parser chain_output = chain.invoke( { "layout_options": self.layout_choices, "json_output": parser.get_format_instructions(), } ) response = SlideContent(**chain_output) return response def add_research_to_slides(self, prs_slide, slide): "Add research to the slides" notes_slide = prs_slide.notes_slide notes_slide.notes_text_frame.text += slide.research def conduct_research(self, slide_info: Slide, section: Section): """Conduct research for the slide.""" import time start = time.time() primary_researcher = InternalSearch( project=self.project, client=self.client_config, ) internal_docs_exist = primary_researcher.project_contain_docs() if internal_docs_exist: XCM_logger.info("Conducting research for slide: %s", slide_info.title) research_combined = primary_researcher.break_down_question_and_answer( question=slide_info.content + "\n".join(slide_info.questions) ) XCM_logger.info( "Internal research for slide: %s completed in %d seconds", slide_info.title, time.time() - start, ) new_research_questions, needs_new_research = ( self.external_research_required(slide_info, research_combined) ) print("Do we need external research?: %s", needs_new_research) if needs_new_research is True or internal_docs_exist is False: get_secondary_start = time.time() research_combined += self.get_secondary_research( new_research_questions, research_combined ) XCM_logger.info( "Secondary research for slide: %s completed in %d seconds", slide_info.title, time.time() - get_secondary_start, ) ## store research in a separate dynamodb table item_id = self._store_research_in_dynamodb( research_combined, slide_info, section ) if not item_id: return None slide_info.research = item_id ## update the project for _section in self.project.sections: if _section.uuid == section.uuid: for slide in section.slides: if slide.uuid == slide_info.uuid: slide.research = item_id self.project.update_project_in_db() end = time.time() print(f"Conducted research for {slide_info.title} took {end - start} seconds") return research_combined def _store_research_in_dynamodb( self, research: str, slide: Slide, section: Section ): "Store research in dynamodb" db = DynamoDB() item_id = Slide.build_slide_research_id( self.project.project_id, section.uuid, slide.uuid ) item = { "id": item_id, "project_id": self.project.project_id, "slide_id": slide.uuid, "section": section.uuid, "research": research, "created_at": datetime.now(timezone.utc).isoformat(), "updated_at": datetime.now(timezone.utc).isoformat(), } try: db.upload_to_dynamodb(db.get_table("slides_research"), item) return item_id except Exception as e: print(f"Error storing research in DynamoDB: {e}") return None def get_secondary_research(self, new_research_questions, primary_research): """Conduct secondary research if additional information is needed.""" secondary_researcher = ResearcherV2( parent_question="questions to answer: " + ", \n".join(new_research_questions), primary_research=primary_research, persona=f"{self.project.industry} Investment Banker making a {self.project.pitch_type} for {self.project.company_name}", ) secondary_research = secondary_researcher.conduct_research() return f"\n\n\nFrom Internet:\n{secondary_research}" def external_research_required(self, slide, primary_research=None): "Check if external research is required" class research_required(BaseModel): new_research_questions: list[str] = Field( default=[], title="The new research questions" ) needs_new_research: bool = Field( default=False, title="Whether new research is needed" ) chat_prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are an expert AI researcher skilled in understanding if it would be helpful to source information from Google.", ), ("human", "The slide we are creating is titled: {slide_title}"), ("human", "The content of the slide is: {slide_content}"), ( "human", "The information you have gathered is \n {primary_research} \n", ), ( "human", "Do you need to conduct an internet search to get more information? True or False. Only return the answer", ), ( "human", "You will want to conduct research for competitors, industry, or other generally publicly available information informationt that could help.", ), ( "system", "You are going to respond in a json object as {json_output}", ), ] ) parser = JsonOutputParser(pydantic_object=research_required) llm = ChatOpenAI( api_key=os.getenv("OPENAI_API_KEY"), model=OPENAI_MODEL_MINI, temperature=0.5, ) chain = chat_prompt | llm | parser chain_output = chain.invoke( { "slide_title": slide.title, "slide_content": slide.content, "primary_research": primary_research, "json_output": parser.get_format_instructions(), } ) chain_output = research_required(**chain_output) return chain_output.new_research_questions, chain_output.needs_new_research def convert_structure_to_ppt( self, ): "Convert the structure to ppt" prs = self.prs # add an overview of the company as the first few slides company_profile_slide = CompanyProfileSlide( company_url=self.company_url, ppt_object=prs, new_deck=False, ) self.beginning_slide(prs) # save the ppt self.create_company_profile(prs, company_profile_slide) ## conduct research for the presentation for section in self.slides_structure: for slide in section.slides: try: prs_slide = prs.slides.add_slide( self.get_slide_layout(prs, "title_only") ) # 3 is the index of the title only layout self.add_research_to_slides(prs_slide, slide) self.create_slide(prs_slide, slide) # save the ppt except Exception as e: print(e) logging.error( "Error in creating slide: %s \n Project: %s", slide.title, self.project.project_id, exc_info=True, ) continue # self.add_research_to_slides(prs_slide, slide) self.save_ppt(prs) self.project.update_project_in_db() self.save_ppt(prs) self.create_target_profiles(prs) self.create_comps_pages(prs) self.save_ppt(prs) return self.filepath def create_slide(self, prs_slide, slide: Slide): "Create the slide" for placeholder in prs_slide.placeholders: if placeholder.placeholder_format.idx == 0: placeholder.text = slide.title prs_slide.name = slide.title slide.content_layout = self.find_appropriate_layout(slide) if slide.content_layout.layout in [ "Layout A", "Layout B", ]: slide_creator = SideBySide( prs_slide, self.slide_work_area, slide.content_layout.information_to_provide, self.project, slide, client_config=self.client_config, research=slide.research, ) slide_creator.left_side() slide_creator.right_side() elif slide.content_layout.layout == "Layout C": slide_creator = TopBottom( prs_slide, self.slide_work_area, slide.content_layout.information_to_provide, self.project, slide, client_config=self.client_config, research=slide.research, ) slide_creator.top_half() slide_creator.bottom_half() elif slide.content_layout.layout in [ "Layout D", "Layout E", "Layout F", "Layout G", "Layout H", "Layout I", ]: if len(slide.content_layout.information_to_provide.keys()) > 1: print("More than one content layout provided") # set the slide content layout to the first layout slide_info = list( slide.content_layout.information_to_provide.keys() )[0] slide.content_layout.information_to_provide = { slide_info: slide.content_layout.information_to_provide[ slide_info ] } slide_creator = FullPage( prs_slide, self.slide_work_area, slide.content_layout.information_to_provide, self.project, slide, client_config=self.client_config, research=slide.research, ) slide_creator.add_content() else: print("Layout not found") raise ValueError("Layout not found") def research_presentation(self): """Conduct research for the presentation using multi-threading for API calls.""" # Use half of the CPU cores for threading (at least 1) max_workers = min(4, os.cpu_count() // 2) # max_workers = 1 def conduct_research_task(slide, section): """Function to fetch research data for a slide.""" research_item_id = Slide.build_slide_research_id( self.project.project_id, section.uuid, slide.uuid ) def get_research_from_dynamodb(research_item_id): db = DynamoDB() item = db.get_item(db.get_table("slides_research"), research_item_id) if item is not None: return item.get("research", "") return "" if getattr(slide, "research", None): if research_item_id == getattr(slide, "research"): return slide, get_research_from_dynamodb(research_item_id) return slide, getattr(slide, "research") if getattr(slide, "extra", {}) is not None: if slide.extra.get("research", None): if research_item_id == slide.extra.get("research"): return slide, get_research_from_dynamodb(research_item_id) return slide, slide.extra.get("research") ## get the slide research from dynamodb research_from_dynambo = "" research_from_dynambo = get_research_from_dynamodb(research_item_id) if research_from_dynambo: return slide, research_from_dynambo return slide, self.conduct_research(slide, section) # API Call # return slide, self.conduct_research(slide) # API Call with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit(conduct_research_task, slide, section): slide for section in self.project.sections for slide in section.slides } for future in as_completed(futures): slide, research = future.result() # slide.research = research def create_company_profile(self, prs, company_profile_slide): company_profile_slide.create_slides("private_company_profile") if company_profile_slide.company_info.stock_ticker is not None: company_profile_slide.create_slides("stock_chart_output") # save the ppt self.save_ppt(prs) def beginning_slide(self, prs): "Create the beginning slide" if self.client_config and getattr(self.client_config, "layout_file", None): client_module = importlib.import_module( f"services.ppt_generator.layouts.{self.client_config.layout_file}.layouts" ) # TODO: Adjust the import module with the actual module in the config file # download the company logo # TODO: Change Bob Cool to the actual person if hasattr(client_module, "BeginningSlides"): beg_slides = client_module.BeginningSlides( prs, "Bob Cool", self.project ) beg_slides.main() self.save_ppt(prs) pass def get_slide_layout(self, prs, layout_name): "Get the slide layout from the layout name" for layout in prs.slide_layouts: if layout.name == layout_name: return layout def create_target_profiles(self, prs): "Create the target profiles" # create a page that is the overview for all of the companies if not self.targets: return company_groups = self.group_companies(self.targets) grid_details = self.create_grid(company_groups) # create a landscape slide for the overview self.create_landscape( prs, grid_details, title="Potential acquisition target landscape" ) for group in company_groups["groups"]: for company in group["companies"]: company_profile_slide = CompanyProfileSlide( company_url=company, ppt_object=prs, new_deck=False, ) company_profile_slide.create_slides("private_company_profile") self.save_ppt(prs) def create_grid(self, company_groups): "Create the grid for the overview of the companies" grid_details = [] for group in company_groups["groups"]: grid_detail = { "title": group["group_name"], "text": group["rationale"], "images": [], } for company in group["companies"]: company_url = company company_info = CompanyInfo.get_company_info(company_url) grid_detail["images"].append(company_info.logo.logo_url) grid_details.append(grid_detail) return grid_details def create_landscape(self, prs, company_groups, title): "Create a landscape slide for the overview of the companies" prs_slide = prs.slides.add_slide(prs.slide_layouts[4]) for placeholder in prs_slide.placeholders: if placeholder.placeholder_format.idx == 0: placeholder.text = title slide_creator = FullPage( prs_slide, SlideWorkArea(), {"full_page": "grid"}, self.project, None, conduct_research_flag=False, ) slide_creator.make_slide.grid( prs_slide, top=SlideWorkArea().top, left=SlideWorkArea().left, width=SlideWorkArea().width, height=SlideWorkArea().height, grid_details=company_groups, ) def group_companies(self, companies_to_group): "Group the companies based on overview and products" # get companies from ChromaDB chroma_db = ChromaDB() acq_company = chroma_db.product_collection.get( ids=parsed_url(self.company_url).url )["documents"][0] target_companies = [] for company in companies_to_group: company_url = parsed_url(company).url company_details = chroma_db.product_collection.get(ids=company_url) # get logo and company name target_companies.append( { "company_url": company_url, "product_description": company_details["documents"][0], } ) # group companies acq_recommender = AcquisitionsRecommender() company_groups = acq_recommender.regroup_companies( acq_company_description=acq_company, companies=target_companies, open_ai_model=acq_recommender.openai_llm_3, ) return company_groups def create_comps_pages(self, prs): "Create the comp pages" if self.project.public_comps is None: return if len(self.project.public_comps) == 0: return # make the comps set if self.main_company["stock_ticker"]: self.project.public_comps.append(self.main_company["stock_ticker"]) public_comps = PublicComparables(self.project.public_comps) # create the first page with tables slide_layout = self.get_slide_layout(prs, "title_only") prs_slide = prs.slides.add_slide(slide_layout) for placeholder in prs_slide.placeholders: if placeholder.placeholder_format.idx == 0: placeholder.text = "Public Comparisions" slide_creator = CompsTable( prs_slide, self.slide_work_area, public_comps, ) self.save_ppt(prs) # create the next pages with the charts def save_ppt(self, prs): "Save the ppt" if self.filename is None: self.filename = f"{self.company}_{self.project.pitch_type}_{datetime.now().strftime('%Y-%m-%d')}" if self.filepath is None: self.filepath = r"ppt_templates/finalized/%s.pptx" % (self.filename) prs.save(self.filepath) if __name__ == "__main__": from services.ppt_generator.data_classes.project import Project from utils.dynamo_db import DynamoDB db = DynamoDB() project = db.get_item(db.projects, "xcm_demo_1") project = Project(**project) # project.targets = [ # "unity.com", # "snap.com", # "https://www.homedepot.com/", # "https://www.picomes.com/", # "https://www.airbnb.com/", # "https://www.jfrog.com", # "https://www.doroni.io", # ] project.sections.pop(1) print(project) ppt = CreatePPT(project, project.company_url) ppt.convert_structure_to_ppt() # ppt.create_target_profiles(ppt.prs) filename = f"{project.company_name}_{datetime.now().strftime('%Y-%m-%d')}" filepath = r"ppt_templates/finalized/%s.pptx" % (filename) ppt.prs.save(filepath)