--- name: "erp-integration-analysis" description: "Analyze ERP system integration for construction data flows. Map and optimize data flows between ERP modules" homepage: "https://datadrivenconstruction.io" metadata: {"openclaw": {"emoji": "🔗", "os": ["win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}} ---

ERP Integration Analysis

Overview

Based on DDC methodology (Chapter 1.2), this skill analyzes ERP system integration patterns in construction organizations, mapping data flows between modules and identifying optimization opportunities.

Book Reference: "Технологии и системы управления в современном строительстве" / "Technologies and Management Systems in Modern Construction"

Quick Start

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Set, Tuple
from datetime import datetime
import json

class ERPModule(Enum):
    """Common ERP modules in construction"""
    FINANCE = "finance"
    PROJECT_MANAGEMENT = "project_management"
    PROCUREMENT = "procurement"
    INVENTORY = "inventory"
    HR = "human_resources"
    PAYROLL = "payroll"
    EQUIPMENT = "equipment"
    SUBCONTRACTS = "subcontracts"
    BILLING = "billing"
    COST_CONTROL = "cost_control"
    DOCUMENT_MANAGEMENT = "document_management"
    REPORTING = "reporting"

class IntegrationMethod(Enum):
    """Types of integration methods"""
    API = "api"
    DATABASE = "database"
    FILE_EXPORT = "file_export"
    MANUAL = "manual"
    WEBHOOK = "webhook"
    MESSAGE_QUEUE = "message_queue"
    ETL = "etl"

class DataFlowDirection(Enum):
    """Direction of data flow"""
    INBOUND = "inbound"
    OUTBOUND = "outbound"
    BIDIRECTIONAL = "bidirectional"

@dataclass
class DataFlow:
    """Represents a data flow between systems/modules"""
    source_module: str
    target_module: str
    data_type: str
    frequency: str  # real-time, hourly, daily, weekly, manual
    method: IntegrationMethod
    direction: DataFlowDirection
    volume: str  # low, medium, high
    critical: bool = False
    issues: List[str] = field(default_factory=list)

@dataclass
class ERPSystem:
    """ERP system definition"""
    name: str
    vendor: str
    version: str
    modules: List[ERPModule]
    database: str
    has_api: bool
    api_type: Optional[str] = None  # REST, SOAP, GraphQL
    custom_modules: List[str] = field(default_factory=list)

@dataclass
class IntegrationPoint:
    """Integration point between systems"""
    id: str
    source_system: str
    target_system: str
    method: IntegrationMethod
    endpoint: Optional[str] = None
    authentication: Optional[str] = None
    data_format: str = "json"
    status: str = "active"
    reliability_score: float = 1.0
    last_sync: Optional[datetime] = None

@dataclass
class IntegrationAnalysis:
    """Complete integration analysis results"""
    erp_system: ERPSystem
    external_systems: List[str]
    data_flows: List[DataFlow]
    integration_points: List[IntegrationPoint]
    integration_score: float
    bottlenecks: List[str]
    recommendations: List[str]
    data_flow_diagram: Dict


class ERPIntegrationAnalyzer:
    """
    Analyze ERP system integration for construction data flows.
    Based on DDC methodology Chapter 1.2.
    """

    def __init__(self):
        self.module_dependencies = self._define_module_dependencies()
        self.critical_flows = self._define_critical_flows()

    def _define_module_dependencies(self) -> Dict[ERPModule, List[ERPModule]]:
        """Define typical module dependencies"""
        return {
            ERPModule.PROJECT_MANAGEMENT: [
                ERPModule.COST_CONTROL,
                ERPModule.PROCUREMENT,
                ERPModule.HR,
                ERPModule.DOCUMENT_MANAGEMENT
            ],
            ERPModule.COST_CONTROL: [
                ERPModule.FINANCE,
                ERPModule.PROJECT_MANAGEMENT,
                ERPModule.BILLING
            ],
            ERPModule.PROCUREMENT: [
                ERPModule.INVENTORY,
                ERPModule.FINANCE,
                ERPModule.SUBCONTRACTS
            ],
            ERPModule.BILLING: [
                ERPModule.FINANCE,
                ERPModule.PROJECT_MANAGEMENT,
                ERPModule.COST_CONTROL
            ],
            ERPModule.PAYROLL: [
                ERPModule.HR,
                ERPModule.FINANCE,
                ERPModule.PROJECT_MANAGEMENT
            ],
            ERPModule.INVENTORY: [
                ERPModule.PROCUREMENT,
                ERPModule.PROJECT_MANAGEMENT,
                ERPModule.FINANCE
            ],
            ERPModule.EQUIPMENT: [
                ERPModule.PROJECT_MANAGEMENT,
                ERPModule.FINANCE,
                ERPModule.INVENTORY
            ],
            ERPModule.SUBCONTRACTS: [
                ERPModule.PROCUREMENT,
                ERPModule.FINANCE,
                ERPModule.PROJECT_MANAGEMENT
            ]
        }

    def _define_critical_flows(self) -> List[Tuple[str, str]]:
        """Define business-critical data flows"""
        return [
            ("project_management", "cost_control"),
            ("cost_control", "finance"),
            ("procurement", "inventory"),
            ("billing", "finance"),
            ("hr", "payroll"),
            ("project_management", "billing")
        ]

    def analyze_erp_integration(
        self,
        erp_system: ERPSystem,
        external_systems: List[Dict],
        integration_points: List[IntegrationPoint],
        transaction_logs: Optional[List[Dict]] = None
    ) -> IntegrationAnalysis:
        """
        Perform comprehensive ERP integration analysis.

        Args:
            erp_system: The ERP system to analyze
            external_systems: List of external systems
            integration_points: Defined integration points
            transaction_logs: Optional transaction logs for analysis

        Returns:
            Complete integration analysis
        """
        # Map all data flows
        data_flows = self._map_data_flows(
            erp_system, integration_points, transaction_logs
        )

        # Calculate integration score
        integration_score = self._calculate_integration_score(
            erp_system, data_flows, integration_points
        )

        # Identify bottlenecks
        bottlenecks = self._identify_bottlenecks(
            data_flows, integration_points
        )

        # Generate recommendations
        recommendations = self._generate_recommendations(
            erp_system, data_flows, bottlenecks
        )

        # Create data flow diagram
        diagram = self._create_flow_diagram(
            erp_system, external_systems, data_flows
        )

        return IntegrationAnalysis(
            erp_system=erp_system,
            external_systems=[s["name"] for s in external_systems],
            data_flows=data_flows,
            integration_points=integration_points,
            integration_score=integration_score,
            bottlenecks=bottlenecks,
            recommendations=recommendations,
            data_flow_diagram=diagram
        )

    def _map_data_flows(
        self,
        erp: ERPSystem,
        integration_points: List[IntegrationPoint],
        logs: Optional[List[Dict]]
    ) -> List[DataFlow]:
        """Map all data flows in the system"""
        flows = []

        # Internal module flows
        for module in erp.modules:
            dependencies = self.module_dependencies.get(module, [])
            for dep in dependencies:
                if dep in erp.modules:
                    is_critical = (module.value, dep.value) in self.critical_flows

...