erp-ai-latest/src/services/sop_generator.py

import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
from src.prompts.sops import *
from src.models.response_schemas import *

class SopGeneratorDocument:
    def __init__(self):
        self.api_key = os.getenv("OPENAI_API_KEY")
        self.client = OpenAI(api_key=self.api_key)
        self.model = "gpt-4o-2024-08-06"

    def _extract_text_from_docs(self, docs):
        """Extract text content from document objects."""
        return [doc.page_content for doc in docs]
    # Existing methods...

    def extract_sops_from_doc(self, docs) -> VisionMissionResponse:
        """
        Extracts Vision, Mission, and SOPs categorized into 'must,' 'shall,' and 'will' from the document.
        
        :param docs: The document(s) from which to extract information.
        :return: VisionMissionResponse containing the vision, mission, and role-specific SOPs.
        """

        try:
                docs_text = self._extract_text_from_docs(docs)
                prompt = get_sop_extraction_from_doc()
                response = self.client.beta.chat.completions.parse(
                    model=self.model,
                    messages=[
                        {
                            "role": "system",
                            "content": f'''{prompt}'''
                        },
                        {
                            "role": "user",
                            "content": [{"type": "text", "text": text} for text in docs_text],
                        }
                    ],
                    response_format=SOPsResponse,
                    max_tokens=4096,
                    temperature=0.1
                )

                # Parse the response from the LLM
                extracted_text = json.loads(response.choices[0].message.content)

                return extracted_text

        except:
            return False

    def extract_vision_mission(self, docs) -> VisionMissionResponse:
        """
        Extracts Vision, Mission, and SOPs categorized into 'must,' 'shall,' and 'will' from the document.
        
        :param docs: The document(s) from which to extract information.
        :return: VisionMissionResponse containing the vision, mission, and role-specific SOPs.
        """

        try:
                docs_text = self._extract_text_from_docs(docs)
                prompt = get_vision_mission_extraction_from_doc()
                response = self.client.beta.chat.completions.parse(
                    model=self.model,
                    messages=[
                        {
                            "role": "system",
                            "content": f'''{prompt}'''
                        },
                        {
                            "role": "user",
                            "content": [{"type": "text", "text": text} for text in docs_text],
                        }
                    ],
                    response_format=VisionMissionResponse,
                    max_tokens=4096,
                    temperature=0.1
                )

                # Parse the response from the LLM
                extracted_text = json.loads(response.choices[0].message.content)

                return extracted_text

        except:
            return False

    def extract_departments_and_managers(self, docs):
        """
        Extract departments and managerial roles from the document.
        
        :param docs: List of document chunks
        :return: Dictionary containing departments and their managerial roles
        """
        try:
            docs_text = self._extract_text_from_docs(docs)
            prompt = get_departments_and_roles_extraction_prompt()
                
            response = self.client.beta.chat.completions.parse(
                    model=self.model,
                    messages=[
                        {"role": "system", "content": prompt},
                        {"role": "user", "content": [{"type": "text", "text": text} for text in docs_text]}
                    ],
                    response_format=DepartmentsAndRolesResponse,
                    max_tokens=4096,
                    temperature=0.1
                )
                
                
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return False




class SopPersonalAssessment:
    def __init__(self):
        self.api_key = os.getenv("OPENAI_API_KEY")
        self.client = OpenAI(api_key=self.api_key)
        self.model = "gpt-4o-mini"

    # Existing methods...

    def extract_sops_from_questionnaire(self, questionnaire_data: dict) -> VisionMissionResponse:
        """
        Extracts SOPs categorized into 'must,' 'shall,' and 'will' based on free-text questionnaire information.

        :param questionnaire_data: A dictionary containing plain text responses to the questionnaire.
        :return: SOPsResponse containing the vision, mission, and role-specific SOPs
        """

        try:
            # Extract the plain text answers from the questionnaire
            vision = questionnaire_data.get("vision", "No vision provided")
            roles = questionnaire_data.get("roles", "No roles provided")  # This will be a plain text string
            responsibilities = questionnaire_data.get("responsibilities", "No responsibilities provided")  # Plain text string
            project_details = questionnaire_data.get("project_details", "No additional project details provided")

            # Get the appropriate prompt for the questionnaire scenario
            prompt = get_sop_personalassessment_from_questionnaire()

            # Combine the plain text data into a string to pass to the LLM
            user_content = f'''
            Vision: {vision}
            Roles: {roles}
            Responsibilities: {responsibilities}
            Project Details: {project_details}
            '''

            # Send the combined plain text content to the LLM
            response = self.client.beta.chat.completions.parse(
                model=self.model,
                messages=[
                    {
                        "role": "system",
                        "content": f'''{prompt}'''
                    },
                    {
                        "role": "user",
                        "content": user_content,
                    }
                ],
                response_format=SOPsResponse,
                max_tokens=2048,
                temperature=0.1
            )

            # Parse the response from the LLM
            extracted_text = json.loads(response.choices[0].message.content)

            return extracted_text

        except Exception as e:
            print(f"Error occurred: {str(e)}")
            return False


    def generate_sops_by_role_and_area(self, roles: List[dict]) -> RoleSops:

        try:
            sops_by_role = []
            for role_info in roles:
                role = role_info['role']
                sop_types = role_info['sop_types']  # List of SOP types: ["will", "shall", "must"]
                areas = role_info['areas']  # List of areas: ["communication", "development", etc.]


                prompt = get_sop_personalassessment_from_area_role(role,areas,sop_types)
                response = self.client.beta.chat.completions.parse(
                model=self.model,
                messages=[
                    {
                        "role": "system",
                        "content": f'''{prompt}
                                    ''',
                    }
                ],
                response_format=RoleSops,
                max_tokens=1024,
                temperature=0.1
            )   
                extracted_text = json.loads(response.choices[0].message.content)
                # You can customize this to generate SOPs based on the role, SOP types, and areas
                sops_by_role.append(extracted_text)

            

            return sops_by_role
        
        except:
            return False
        


class SopGeneratorExecutive:
    def __init__(self):
        self.api_key = os.getenv("OPENAI_API_KEY")
        self.client = OpenAI(api_key=self.api_key)
        self.model = "gpt-4o-mini"

    def extract_sops_from_executive_vision_goals_doc(self, data: dict,executives:List) -> SOPsResponse:
        """
        Extracts SOPs categorized into 'must,' 'shall,' and 'will' based on executive vision and goals.

        :param data: A dictionary containing vision and goals.
        :return: SOPsResponse containing the SOPs for executives
        """
        try:
             
            vision = data.get("vision", "No vision provided")
            goals = data.get("goals", "No goals provided")
            
            prompt = get_sop_executive_from_vision_goals(executives)

            user_content = f'''
            Vision: {vision}
            Goals: {goals}
            '''

            response = self.client.beta.chat.completions.parse(
                model=self.model,
                messages=[
                    {
                        "role": "system",
                        "content": f'''{prompt}'''
                    },
                    {
                        "role": "user",
                        "content": user_content,
                    }
                ],
                response_format=Categories,
                max_tokens=2048,
                temperature=0.1
            )

            extracted_text = json.loads(response.choices[0].message.content)
            return extracted_text

        except Exception as e:
            print(f"Error occurred: {str(e)}")
            return False


    def generate_sops_for_department_managers(self, docs):
        try:
            # First, extract departments and managers
            sop_doc = SopGeneratorDocument()
            departments_and_roles = sop_doc.extract_departments_and_managers(docs)

            if not departments_and_roles or not departments_and_roles.get('departments'):
                return False

            departments_with_sops = []

            for department in departments_and_roles['departments']:
                managers_with_sops = []
                for role in department['managerial_roles']:
                    prompt = get_sop_for_department_managers()
                    response = self.client.beta.chat.completions.parse(
                        model=self.model,
                        messages=[
                            {"role": "system", "content": prompt},
                            {"role": "user", "content": f"Generate SOPs for {role['title']} in {department['name']} department."}
                        ],
                        response_format=ManagerSOPs,
                        max_tokens=1024,
                        temperature=0.1
                    )
                    manager_sops = json.loads(response.choices[0].message.content)
                    managers_with_sops.append(ManagerWithSOPs(title=role['title'], sops=manager_sops))

                departments_with_sops.append(DepartmentManagerSOPs(
                    name=department['name'],
                    managers=managers_with_sops
                ))

            return ExecutiveManagerSOPsResponse(departments=departments_with_sops)

        except Exception as e:
            print(f"Error in generate_sops_for_department_managers: {str(e)}")
            return False


    def generate_sops_from_questionnaire(self, questionnaire_data: dict, executives: List[str], managers: List[str], departments: List[str]):
        try:
            prompt = get_sop_executive_from_questionnaire()

            # Prepare the questionnaire data for the prompt
            user_content = json.dumps(questionnaire_data, indent=2)

            response = self.client.beta.chat.completions.parse(
                model=self.model,
                messages=[
                    {"role": "system", "content": prompt},
                    {"role": "user", "content": f"Generate SOPs based on this questionnaire:\n{user_content}\n\nExecutives to consider: {', '.join(executives)}\nManagers to consider: {', '.join(managers)}\nDepartments to consider: {', '.join(departments)}"}
                ],
                response_format={"type": "json_object"},
                max_tokens=4096,
                temperature=0.1
            )

            sops_data = json.loads(response.choices[0].message.content)

            # Process executive SOPs
            executive_sops = {}
            for executive in executives:
                if executive in sops_data['executives']:
                    executive_sops[executive] = Categories(**sops_data['executives'][executive])
                else:
                    executive_sops[executive] = Categories()

            # Process department manager SOPs
            departments_with_sops = []
            for dept_name in departments:
                dept_data = next((d for d in sops_data['departments'] if d['name'].lower() == dept_name.lower()), None)
                if dept_data:
                    managers_with_sops = [
                        ManagerWithSOPs(
                            title=manager,
                            sops=ManagerSOPs(**dept_data['managers'])
                        )
                        for manager in managers
                    
                    ]
                    if managers_with_sops:
                        departments_with_sops.append(DepartmentManagerSOPs(
                            name=dept_name,
                            managers=managers_with_sops
                        ))

            return {
                "executive_sops": executive_sops,
                "department_sops": ExecutiveManagerSOPsResponse(departments=departments_with_sops)
            }

        except Exception as e:
            print(f"Error in generate_sops_from_questionnaire: {str(e)}")
            return False