erp-data-extractor

name	erp-data-extractor
description	Extract and analyze data from construction ERP systems. Pull project data for analytics, reporting, and integration.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🔄","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

ERP Data Extractor

Business Case

Problem Statement

ERP data extraction challenges:

• Complex database structures
• Multiple interconnected modules
• Data transformation needs
• Integration with analytics

Solution

Structured extraction and transformation of construction ERP data for analytics, reporting, and cross-system integration.

Technical Implementation

import pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import date, datetime
from enum  Enum
 json


 ():
    PROJECT = 
    COST = 
    PROCUREMENT = 
    INVENTORY = 
    HR = 
    EQUIPMENT = 
    SUBCONTRACT = 
    BILLING = 



 :
    name: 
    module: ERPModule
    table_name: 
    columns: []
    filters: [, ] = field(default_factory=)



 :
    source: 
    module: ERPModule
    data: pd.DataFrame
    extracted_at: datetime
    record_count: 


 :
    

     ():
        .erp_name = erp_name
        .data_sources: [DataSource] = []
        .extracted_data: [, ExtractedData] = {}
        ._connection = 

     ():
        
        .data_sources.append(source)

     ():
        

        .add_data_source(DataSource(
            name=,
            module=ERPModule.PROJECT,
            table_name=,
            columns=[, , , , , , , ]
        ))

        .add_data_source(DataSource(
            name=,
            module=ERPModule.PROJECT,
            table_name=,
            columns=[, , , , , ]
        ))

     ():
        

        .add_data_source(DataSource(
            name=,
            module=ERPModule.COST,
            table_name=,
            columns=[, , , , , , ]
        ))

        .add_data_source(DataSource(
            name=,
            module=ERPModule.COST,
            table_name=,
            columns=[, , , , , ]
        ))

     ():
        

        .add_data_source(DataSource(
            name=,
            module=ERPModule.PROCUREMENT,
            table_name=,
            columns=[, , , , , , ]
        ))

        .add_data_source(DataSource(
            name=,
            module=ERPModule.PROCUREMENT,
            table_name=,
            columns=[, , , , ]
        ))

     ():
        

        source = ((s  s  .data_sources  s.name == source_name), )
          source:
             

        
        available_cols = [c  c  source.columns  c  df.columns]
        extracted = df[available_cols].copy()

        
         col, value  source.filters.items():
             col  extracted.columns:
                extracted = extracted[extracted[col] == value]

        .extracted_data[source_name] = ExtractedData(
            source=source_name,
            module=source.module,
            data=extracted,
            extracted_at=datetime.now(),
            record_count=(extracted)
        )

         .extracted_data[source_name]

     () -> pd.DataFrame:
        

         source_name   .extracted_data:
             pd.DataFrame()

        df = .extracted_data[source_name].data.copy()

         transform  transformations:
            action = transform.get()

             action == :
                df = df.rename(columns=transform.get(, {}))

             action == :
                col = transform.get()
                op = transform.get(, )
                val = transform.get()
                 op == :
                    df = df[df[col] == val]
                 op == :
                    df = df[df[col] > val]
                 op == :
                    df = df[df[col] < val]

             action == :
                new_col = transform.get()
                formula = transform.get()
                 formula == :
                    df[new_col] = df[transform[]] - df[transform[]]

             action == :
                col = transform.get()
                df[col] = pd.to_datetime(df[col])

         df

     () -> pd.DataFrame:
        

         left_source   .extracted_data  right_source   .extracted_data:
             pd.DataFrame()

        left_df = .extracted_data[left_source].data
        right_df = .extracted_data[right_source].data

         pd.merge(left_df, right_df, left_on=left_key, right_on=right_key, how=join_type)

     () -> pd.DataFrame:
        

         source_name   .extracted_data:
             pd.DataFrame()

        df = .extracted_data[source_name].data
         df.groupby(group_by).agg(aggregations).reset_index()

     () -> [, ]:
        

        summary = {
            : .erp_name,
            : (.data_sources),
            : (.extracted_data),
            : (e.record_count  e  .extracted_data.values()),
            : {}
        }

         ext  .extracted_data.values():
            module = ext.module.value
             module   summary[]:
                summary[][module] = {: , : }
            summary[][module][] += 
            summary[][module][] += ext.record_count

         summary

     () -> :
        

         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary = .get_extraction_summary()
            summary_df = pd.DataFrame([{
                : summary[],
                : summary[],
                : summary[],
                : summary[]
            }])
            summary_df.to_excel(writer, sheet_name=, index=)

            
             name, extracted  .extracted_data.items():
                sheet_name = name[:]  
                extracted.data.to_excel(writer, sheet_name=sheet_name, index=)

         output_path

     () -> :
        

        output = {
            : .get_extraction_summary(),
            : {}
        }

         name, extracted  .extracted_data.items():
            output[][name] = {
                : extracted.module.value,
                : extracted.extracted_at.isoformat(),
                : extracted.record_count,
                : extracted.data.to_dict(orient=)
            }

         (output_path, )  f:
            json.dump(output, f, indent=, default=)

         output_path

     () -> :
        

        columns = .join(source.columns)
        query = 

         source.filters:
            conditions = []
             col, value  source.filters.items():
                 (value, ):
                    conditions.append()
                :
                    conditions.append()
            query +=  + .join(conditions)

         query +