Initial project build from sources developed in the standalone F&O folder

from parse_res import ResEstimateData

import psycopg2

from psycopg2 import sql

class LoadRes(ResEstimateData):

    def __init__(self, project_title, resource_pool_name, filename, pg_conn):

        super().__init__(filename)

        self.pg_con = pg_conn

        self.resource_poolid = self._get_id("fo_itsd_estimate",

                                            "resource_pools",

                                            "resource_pool_id",

                                            "description", resource_pool_name)

        try:

            self.workorder_id = self._insert_record_and_get_pk("fo_itsd_estimate", "workorders",

                                                               {'title':project_title,

                                                                'resource_pool_id':self.resource_poolid, 'estimate_file_name':filename},

                                                               'workorder_id')

        except Exception as e:

            if e.pgcode == '23505':

                print ('Estimate file already loaded')

                exit(999)

    def _get_id(self, schema_name, table_name, column, filter_column, value):

        # Construct the SQL query safely to prevent SQL injection

        # This uses psycopg2.sql objects to safely handle identifiers (table/column names)

        query = sql.SQL("SELECT {col} FROM {tbl} WHERE {key} = {val}").format(

            col=sql.Identifier(column),

            tbl=sql.SQL('.').join([

                sql.Identifier(schema_name),

                sql.Identifier(table_name)

            ]),

            key=sql.Identifier(filter_column),

            val=sql.Placeholder()  # For the data value

        )

        try:

            # Use a 'with' statement for the cursor to ensure it's closed automatically

            with self.pg_con.cursor() as cursor:

                # Execute the query with the data values

                cursor.execute(query, (value,))

                # Fetch the returned primary key from the cursor

                return cursor.fetchone()[0]

        except Exception as e:

            print(query, (value,))

            raise e

    def _insert_record_and_get_pk(self, schema_name, table_name, data, pk_column='id'):

        """

        Inserts a record into a table and returns the new primary key.

        Args:

            table_name (str): The name of the target table.

            data (dict): A dictionary where keys are column names and values are the data to insert.

            pk_column (str): The name of the serial primary key column (defaults to 'id').

        Returns:

            The value of the primary key for the newly inserted row, or None if insertion fails.

        """

        new_id = None

        columns = data.keys()

        # Construct the SQL query safely to prevent SQL injection

        # This uses psycopg2.sql objects to safely handle identifiers (table/column names)

        query = sql.SQL("INSERT INTO {tbl} ({cols}) VALUES ({vals}) RETURNING {pk}").format(

            tbl=sql.SQL('.').join([

                sql.Identifier(schema_name),

                sql.Identifier(table_name)

            ]),

            cols=sql.SQL(', ').join(map(sql.Identifier, columns)),

            vals=sql.SQL(', ').join(sql.Placeholder() * len(columns)),

            pk=sql.Identifier(pk_column)

        )

        try:

            # Use a 'with' statement for the cursor to ensure it's closed automatically

            with self.pg_con.cursor() as cursor:

                # Execute the query with the data values

                cursor.execute(query, list(data.values()))

                # Fetch the returned primary key from the cursor

                new_id = cursor.fetchone()[0]

                # Commit the transaction to make the insert permanent

                self.pg_con.commit()

                print(f"✅ Record inserted successfully into '{table_name}'. New ID: {new_id}")

        except (Exception, psycopg2.Error) as error:

            print(f"❌ Error inserting record: {error}")

            # Roll back the transaction in case of an error

            if self.pg_con:

                self.pg_con.rollback()

            # return None

            raise error

        # TO DO - On insert fail, return the Record ID of the existing record

        return new_id

    def load(self):

        """Function to coordinate loading of tables"""

        # Load Resources (people)

        # Need to build a dictionary of names and resource_ids for use later

        dct_names = {}

        for name in self.get_resource_names():

            try:

                dct_names[name] = self._insert_record_and_get_pk('fo_itsd_estimate',

                                                                 'resources',

                                                                 {'full_name':name, 'resource_pool_id':self.resource_poolid},

                                                                 'resource_id')

            except Exception as e:

                if e.pgcode == '23505':

                    dct_names[name] = self._get_id('fo_itsd_estimate','resources','resource_id','full_name',name)

                    pass

        # Load Activities

        dct_est_act_id ={}

        for activity in self.get_wbs_data()['Activities']:

            dct_est_act_id[activity[0]] = self._insert_record_and_get_pk('fo_itsd_estimate',

                                                                         'activities',

                                                                         {'activity_description':activity[1], 'est_act_id':activity[0], 'workorder_id':self.workorder_id},

                                                                         'activity_id')

        # Load Tasks

        for task in self.get_person_hours_task():

            taskrec = {}

            wbs = [wbs for wbs in self.get_wbs_data()['Tasks'] if wbs[0] == task[1]['Item']]

            task_dates = [{'start_date':taskdate[1], 'end_date':taskdate[2]} for taskdate in self.get_task_dates()

                          if taskdate[0]==task[1]['Item']]

            taskrec['resource_id'] = dct_names[task[0]]

            taskrec['hours'] = task[1]['Hours']

            taskrec['contingency_percent'] = task[1]['Cont %']

            taskrec['start_date'] = task_dates[0]['start_date']

            taskrec['end_date']= task_dates[0]['end_date']

            taskrec['activity_id'] = dct_est_act_id[wbs[0][1]]

            taskrec['task_description']= wbs[0][3]

            taskrec['est_task_id']= wbs[0][2]

            self._insert_record_and_get_pk('fo_itsd_estimate','task_details', taskrec,

                                           'task_id')

        return

if __name__ == '__main__':

    # Use project title to allow us to tie this to the information in the Feature Weighting Table

    project_title = 'COREII'

    resource_pool_name = 'ITSD'

    # The name of the Resolution Export file

    res_filename = r'/mnt/c/Users/uvp/Downloads/estimate (8).xlsx'

    # Connect to the database holding the estimate information

    pg_connect_string = (F"host='hgis-prj-mgmt', port='5438', dbname='proj_status', "

                         F"user='postgres',password='postgres'")

    # NOTE: Replace with your actual database connection details

    DB_PARAMS = {

        "dbname": "proj_status",

        "user": "postgres",

        "password": "postgres",

        "host": "hgis-prj-mgmt",

        "port": "5438"

    }

    #Load the information

    pg_conn = psycopg2.connect(**DB_PARAMS)

    objLoadRes = LoadRes(project_title, resource_pool_name, res_filename, pg_conn)

    objLoadRes.load()

        excel_file = openpyxl.load_workbook(str_filename)

        self.sheet = excel_file.active

        self.rows = list(self.sheet.iter_rows())

        the_header = self._get_header()

        the_header = self._gen_header()

        if 'Item' not in the_header[0].keys():

            raise Exception('Item Column missing')

        self.dct_header = the_header[0]

        self.header_bottom = the_header[1]

        self.name_column_id = self.dct_header['Resources']

        self.unit_column_id = self.dct_header['Unit']

        self.dct_docmodel =  self._get_sections()

        self.dct_docmodel =  self._gen_sections()

    def _get_sections(self):

    def _gen_sections(self):

        # Rules: 1: The beginning of a section defines the end of a previous section

        #        2: FY define the beginning of a FY section can contain a Labor and Materials section

        #        3: Labor and Materials values define the Beginning of the sections.

        #  Sort the Dictionary by values, now we have our parse order

        sorted_keys = sorted(dct_sections)

        dct_docmodel = {key: dct_sections[key] for key in sorted_keys}

        print(dct_docmodel)

        # print(dct_docmodel)

        return dct_docmodel

    def _get_section_keys(self, label):

    def _gen_section_keys(self, label):

        # Define the labor sections in the document

        section_bounds = []

        start_keys = [key for key, value in self.dct_docmodel.items() if value[1] == label]

            # Find the next key to define the end of the section

            try:

                first_key_found = next(key for key in self.dct_docmodel if key > start_key)

                print(f"The first key larger than {start_key} is: {first_key_found}")

                # print(f"The first key larger than {start_key} is: {first_key_found}")

                section_bounds.append([start_key, first_key_found])

            except StopIteration:

                print(f"No key found larger than {start_key}.")

                # print(f"No key found larger than {start_key}.")

                section_bounds.append([start_key, None])

        return section_bounds

    def _get_header(self):

    def _gen_header(self):

        """extract Row 3 to get a feeling of the column names"""

        row_number = 3

        for x in range (1,6):

        header_row = self.sheet[row_number]

        return {cell.value: cell.column - 1 for cell in header_row if cell.value}, row_number

    def _get_labor_details(self):

    def _gen_labor_details(self):

        # Build the Field array to extract

        dct_task_detail = {}

        lst_tasks = []

        name = ''

        lst_potential_fields =['Resources', 'Start Date', 'End Date', 'Hours', 'Cont %', 'Activity Description', 'Description', 'Item']

        lst_fields = [[field, self.dct_header[field]] for field in lst_potential_fields if field in self.dct_header]

        labor_sections = self._get_section_keys('Labor')

        labor_sections = self._gen_section_keys('Labor')

        for labor_section in labor_sections:

            # Read until null no more records OR the rowid of a Materials section is reached

            for row in self.rows[labor_section[0]:labor_section[1]]:

                        dct_task_detail = {}

        return lst_tasks

    def _get_materials_details(self):

    def _gen_materials_details(self):

        lst_potential_fields = ['Resources', 'Unit', 'Unit Cost', 'Qty', 'Activity Description', 'Description', 'Item']

        lst_fields = [[field, self.dct_header[field]] for field in lst_potential_fields if field in self.dct_header]

        dct_material_detail = {}

        lst_material = []

        name = ''

        material_sections = self._get_section_keys('Materials')

        material_sections = self._gen_section_keys('Materials')

        for material_section in material_sections:

            for row in self.rows[material_section[0]:material_section[1]]:

                # Read until null no more records OR the rowid of a Labor section is reached

                return True

        return False

    def generate_activity_dict(self):

    def _generate_activity_dict(self):

        lst_activites = []

        dct_activity = {}

        lst_details = self._get_labor_details()

        lst_details = self._gen_labor_details()

        lst_potential_fields = ['Resources', 'Start Date', 'End Date', 'Activity Description']

        lst_fields = [[field, self.dct_header[field]] for field in lst_potential_fields if field in self.dct_header]

        for row in lst_details:

            dct_activity ={}

        return lst_activites

    def generate_activity_task_mapping(self):

    def _generate_activity_task_mapping(self):

        lst_activites = []

        dct_activity = {}

        lst_details = self._get_labor_details()

        lst_potential_fields = ['Resources', 'Start Date', 'End Date', 'Description', 'Item']

        lst_details = self._gen_labor_details()

        lst_potential_fields = ['Resources', 'Start Date', 'End Date', 'Description', 'Item', 'Activity Description']

        lst_fields = [[field, self.dct_header[field]] for field in lst_potential_fields if field in self.dct_header]

        for row in lst_details:

            for field in lst_fields:

        name = ''

        lst_activites = []

        dct_activity = {}

        lst_details = self._get_labor_details()

        lst_potential_fields = ['Resources', 'Hours', 'Cont %', 'Item']

        lst_details = self._gen_labor_details()

        lst_potential_fields = ['Hours', 'Cont %', 'Item']

        lst_fields = [[field, self.dct_header[field]] for field in lst_potential_fields if field in self.dct_header]

        for row in lst_details:

            for field in lst_fields:

            name = ''

        return lst_activites

    def get_task_dates(self):

        lst_activites = self._generate_activity_task_mapping()

        return [[dct_activity['Item'], dct_activity['Start Date'], dct_activity['End Date']]

                for dct_activity in lst_activites]

    def get_resource_names(self):

        """Return a list of names

            get the rows greater than 6 where there is a value in column b, but not c or d"""

        """ TO DO: Add logic to tag name as Labor or Materials"""

        lst_names = [row[self.name_column_id].value for row in self.rows if

                     self._is_in_merged_range(row[self.name_column_id + 1].coordinate)]

        # Trim header from list

        set_names = set(lst_names[self.header_bottom:])

        return list(set_names)

        lst_activites = self.get_person_hours_task()

        lst_names = [row[0] for row in lst_activites]

        st_names = set(lst_names)

        return list(st_names)

    def gen_wbs_data(self):

        ...

    def get_wbs_data(self):

        lst_tasks = self._generate_activity_task_mapping()

        lst_raw_data = [[row['Item'], row['Activity Description'], row['Description']] for row in lst_tasks]

        activity_numbers = [[str(row[0]).split('.')[0], row[1]] for row in lst_raw_data]

        lst_activities = [list(x) for x in set(tuple(x) for x in activity_numbers)]

        task_numbers = [[row[0], str(row[0]).split('.')[0], str(row[0]).split('.')[1], row[2]] for row in lst_raw_data]

        lst_tasks = [list(x) for x in set(tuple(x) for x in task_numbers)]

        dct_wbs = {'Activities':lst_activities, 'Tasks':lst_tasks}

        return dct_wbs

if __name__=="__main__":

    try:

        obj_est = ResEstimateData(r'/mnt/c/Users/uvp/Downloads/estimate (8).xlsx')

    # print(obj_est._get_header())

    # print(obj_est.get_resource_names())

    # print(obj_est.generate_activity_dict())

    # print(obj_est.get_person_hours_task())

    # print(obj_est.generate_activity_task_mapping())

    print(obj_est._get_materials_details())

 No newline at end of file

        [print(name) for name in obj_est.get_resource_names()]

        print(obj_est.get_wbs_data())

        [print(task_dates) for task_dates in obj_est.get_task_dates()]

        [print(hours) for hours in obj_est.get_person_hours_task()]

        #TO DO Get meterials info

    except Exception as e:

        raise e

 No newline at end of file

psycopg2-binary

 No newline at end of file

import psycopg2

from psycopg2 import SQL

from ../libraries/db_Loader import LoadRes

# This is a sample Python script.

# Press Shift+F10 to execute it or replace it with your code.

# Press Double Shift to search everywhere for classes, files, tool windows, actions, and settings.

def print_hi(name):

    # Use a breakpoint in the code line below to debug your script.

    print(f'Hi, {name}')  # Press Ctrl+F8 to toggle the breakpoint.

# Press the green button in the gutter to run the script.

if __name__ == '__main__':

    # Use project title to allow us to tie this to the information in the Feature Weighting Table

    project_title = 'COREII'

    resource_pool_name = 'ITSD'

    # The name of the Resolution Export file

    res_filename = r'/mnt/c/Users/uvp/Downloads/estimate (8).xlsx'

    # Connect to the database holding the estimate information

    pg_connect_string = (F"host='hgis-prj-mgmt', port='5438', dbname='proj_status', "

                         F"user='postgres',password='postgres'")

    # NOTE: Replace with your actual database connection details

    DB_PARAMS = {

        "dbname": "proj_status",

        "user": "postgres",

        "password": "postgres",

        "host": "hgis-prj-mgmt",

        "port": "5438"

    }

    #Load the information

    pg_conn = psycopg2.connect(**DB_PARAMS)

    objLoadRes = LoadRes(project_title, resource_pool_name, res_filename, pg_conn)

    objLoadRes.load()