# 数据提取模块：多源数据采集与增量提取

# 定义数据源配置
$DataSources = @{
    CsvPath    = './data/sales_records.csv'
    JsonPath   = './data/product_catalog.json'
    ApiEndpoint = 'https://api.example.com/v1/orders'
    LastSyncFile = './data/.last_sync_timestamp'
}

# 读取上次同步时间戳（增量提取的关键）
$LastSyncTime = if (Test-Path $DataSources.LastSyncFile) {
    Get-Content $DataSources.LastSyncFile -Raw | ForEach-Object {
        [DateTime]::Parse($_.Trim())
    }
} else {
    [DateTime]::MinValue
}

Write-Host "上次同步时间: $LastSyncTime" -ForegroundColor Cyan

# 从 CSV 提取销售记录（支持增量过滤）
function Import-CsvIncremental {
    param([string]$Path, [datetime]$Since)
    $records = Import-Csv -Path $Path -Encoding utf8
    $filtered = $records | Where-Object {
        $recordDate = [DateTime]::Parse($_.OrderDate)
        $recordDate -gt $Since
    }
    Write-Host "CSV 提取完成: 共 $($records.Count) 条, 增量 $($filtered.Count) 条"
    return $filtered
}

# 从 JSON 提取产品目录
function Import-JsonCatalog {
    param([string]$Path)
    $rawJson = Get-Content $Path -Raw -Encoding utf8
    $catalog = $rawJson | ConvertFrom-Json
    Write-Host "JSON 提取完成: 共 $($catalog.Count) 个产品"
    return $catalog
}

# 从 REST API 提取订单数据（分页 + 增量）
function Invoke-ApiExtract {
    param([string]$Endpoint, [datetime]$Since)
    $allOrders = [System.Collections.Generic.List[object]]::new()
    $page = 1
    $hasMore = $true
    $headers = @{ 'Accept' = 'application/json' }

    while ($hasMore) {
        $params = @{
            Uri     = "$Endpoint`?page=$page&since=$($Since.ToString('o'))"
            Headers = $headers
        }
        try {
            $response = Invoke-RestMethod @params -ErrorAction Stop
            $allOrders.AddRange($response.data)
            $hasMore = $response.has_more
            $page++
        } catch {
            Write-Warning "API 请求失败 (页 $page): $($_.Exception.Message)"
            $hasMore = $false
        }
    }
    Write-Host "API 提取完成: 共 $($allOrders.Count) 条订单"
    return $allOrders
}

# 执行数据采集
$csvData    = Import-CsvIncremental -Path $DataSources.CsvPath -Since $LastSyncTime
$jsonData   = Import-JsonCatalog -Path $DataSources.JsonPath
$apiData    = Invoke-ApiExtract -Endpoint $DataSources.ApiEndpoint -Since $LastSyncTime

# 更新同步时间戳
[DateTime]::UtcNow.ToString('o') | Set-Content $DataSources.LastSyncFile -NoNewline

Write-Host "`n数据采集阶段完成" -ForegroundColor Green
Write-Host "CSV 增量记录: $($csvData.Count) 条"
Write-Host "JSON 产品数:   $($jsonData.Count) 个"
Write-Host "API 订单数:    $($apiData.Count) 条"

# 数据转换模块：清洗、映射、去重与富化

# 模拟采集到的原始数据
$RawRecords = @(
    @{ OrderId = 'ORD-001'; Customer = '张三'; Amount = '1299.50'; Date = '2026-04-18'; Region = '' }
    @{ OrderId = 'ORD-002'; Customer = '李四'; Amount = '899.00';  Date = '2026-04-18'; Region = '华东' }
    @{ OrderId = 'ORD-003'; Customer = '王五'; Amount = '-50.00';  Date = '2026-04-19'; Region = '华南' }
    @{ OrderId = 'ORD-002'; Customer = '李四'; Amount = '899.00';  Date = '2026-04-18'; Region = '华东' }
    @{ OrderId = 'ORD-004'; Customer = '';      Amount = '2450.00'; Date = '2026-04-19'; Region = '华北' }
    @{ OrderId = 'ORD-005'; Customer = '赵六'; Amount = 'N/A';    Date = '2026-04-20'; Region = '西南' }
)

Write-Host "原始记录数: $($RawRecords.Count)" -ForegroundColor Yellow

# 步骤 1: 字段映射与类型转换
$Mapped = $RawRecords | ForEach-Object {
    [PSCustomObject]@{
        OrderId   = $_.OrderId
        Customer  = $_.Customer
        Amount    = $_.Amount
        OrderDate = $_.Date
        Region    = $_.Region
        IsValid   = $true
        Issues    = [System.Collections.Generic.List[string]]::new()
    }
}

# 步骤 2: 数据验证与标记
$Validated = $Mapped | ForEach-Object {
    $record = $_

    # 检查客户名是否为空
    if ([string]::IsNullOrWhiteSpace($record.Customer)) {
        $record.IsValid = $false
        $record.Issues.Add('客户名为空')
    }

    # 尝试转换金额
    $amountValue = 0.0
    if ([double]::TryParse($record.Amount, [ref]$amountValue)) {
        $record.Amount = $amountValue
    } else {
        $record.IsValid = $false
        $record.Issues.Add("金额无法解析: $($record.Amount)")
    }

    # 检查金额是否为负数（异常值）
    if ($amountValue -lt 0) {
        $record.Issues.Add('金额为负数（可能是退款）')
    }

    # 补充缺失的区域信息
    if ([string]::IsNullOrWhiteSpace($record.Region)) {
        $record.Region = '未知'
        $record.Issues.Add('区域信息缺失，已填充默认值')
    }

    # 转换日期格式
    $record.OrderDate = [DateTime]::Parse($record.OrderDate).ToString('yyyy-MM-dd')

    return $record
}

# 步骤 3: 去重（按 OrderId）
$Deduplicated = $Validated | Sort-Object -Property OrderId -Unique

# 步骤 4: 数据富化（添加计算字段）
$Enriched = $Deduplicated | ForEach-Object {
    $tier = switch ($_.Amount) {
        { $_ -ge 2000 } { 'VIP' }
        { $_ -ge 1000 } { '高级' }
        { $_ -ge 500 }  { '标准' }
        default         { '普通' }
    }

    $_ | Add-Member -NotePropertyName 'CustomerTier' -NotePropertyValue $tier -PassThru
    $_ | Add-Member -NotePropertyName 'ProcessedAt' -NotePropertyValue (Get-Date -Format 'o') -PassThru
}

# 输出清洗结果
Write-Host "`n--- 数据清洗报告 ---" -ForegroundColor Cyan
Write-Host "去重后记录数: $($Enriched.Count)"
$validCount = ($Enriched | Where-Object { $_.IsValid }).Count
$invalidCount = ($Enriched | Where-Object { -not $_.IsValid }).Count
Write-Host "有效记录:     $validCount"
Write-Host "无效记录:     $invalidCount"

Write-Host "`n--- 有效数据 ---" -ForegroundColor Green
$Enriched | Where-Object { $_.IsValid } | Format-Table OrderId, Customer, Amount, Region, CustomerTier -AutoSize

Write-Host "--- 异常数据 ---" -ForegroundColor Red
$Enriched | Where-Object { -not $_.IsValid } | ForEach-Object {
    Write-Host "  $($_.OrderId): $($($_.Issues) -join ', ')"
}

# 数据加载与调度模块

# 日志工具函数
$LogPath = './data/pipeline_run.log'

function Write-PipelineLog {
    param(
        [string]$Message,
        [ValidateSet('INFO', 'WARN', 'ERROR')]
        [string]$Level = 'INFO'
    )
    $timestamp = Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff'
    $entry = "[$timestamp] [$Level] $Message"
    Add-Content -Path $LogPath -Value $entry -Encoding utf8
    $color = switch ($Level) {
        'INFO'  { 'White' }
        'WARN'  { 'Yellow' }
        'ERROR' { 'Red' }
    }
    Write-Host $entry -ForegroundColor $color
}

# 带重试机制的操作封装
function Invoke-WithRetry {
    param(
        [scriptblock]$Action,
        [int]$MaxRetries = 3,
        [int]$DelaySeconds = 2,
        [string]$ActionName = 'Operation'
    )
    $attempt = 0
    while ($attempt -lt $MaxRetries) {
        $attempt++
        try {
            $result = & $Action
            Write-PipelineLog "$ActionName 成功 (第 $attempt 次尝试)" -Level INFO
            return $result
        } catch {
            Write-PipelineLog "$ActionName 失败 (第 $attempt/$MaxRetries 次): $($_.Exception.Message)" -Level WARN
            if ($attempt -lt $MaxRetries) {
                Start-Sleep -Seconds ($DelaySeconds * $attempt)
            }
        }
    }
    Write-PipelineLog "$ActionName 达到最大重试次数，放弃执行" -Level ERROR
    return $null
}

# 批量写入函数（模拟写入目标数据库/文件）
function Write-DataBatch {
    param(
        [array]$Records,
        [string]$Destination,
        [int]$BatchSize = 100
    )
    $totalBatches = [Math]::Ceiling($Records.Count / $BatchSize)
    $successCount = 0
    $failCount = 0

    for ($i = 0; $i -lt $Records.Count; $i += $BatchSize) {
        $batchNum = [Math]::Floor($i / $BatchSize) + 1
        $batch = $Records[$i..([Math]::Min($i + $BatchSize - 1, $Records.Count - 1))]

        $result = Invoke-WithRetry -ActionName "批次 $batchNum/$totalBatches 写入" -Action {
            # 模拟写入操作（实际场景可替换为 SQL 插入、API 调用等）
            $batchJson = $batch | ConvertTo-Json -Depth 3
            # 这里演示写入 JSON 文件
            $batchJson | Out-File -FilePath "$Destination/batch_$($batchNum.ToString('D4')).json" -Encoding utf8
            return $batch.Count
        }

        if ($null -ne $result) {
            $successCount += $result
        } else {
            $failCount += $batch.Count
        }
    }

    return @{ Success = $successCount; Failed = $failCount }
}

# 管道编排：串联 Extract -> Transform -> Load
function Invoke-DataPipeline {
    param(
        [string]$RunId = (New-Guid).ToString('N').Substring(0, 8)
    )

    $runStart = Get-Date
    Write-PipelineLog "========== 管道运行 $RunId 开始 ==========" -Level INFO

    try {
        # 阶段 1: 加载已清洗的数据（模拟）
        Write-PipelineLog '阶段 1/3: 加载已处理数据...' -Level INFO
        $processedData = @(
            @{ OrderId = 'ORD-001'; Customer = '张三'; Amount = 1299.50 }
            @{ OrderId = 'ORD-002'; Customer = '李四'; Amount = 899.00 }
            @{ OrderId = 'ORD-003'; Customer = '王五'; Amount = 2450.00 }
            @{ OrderId = 'ORD-006'; Customer = '孙七'; Amount = 3200.00 }
            @{ OrderId = 'ORD-007'; Customer = '周八'; Amount = 150.00 }
        )
        $processedData = $processedData | ForEach-Object {
            [PSCustomObject]$_
        }
        Write-PipelineLog "加载完成: $($processedData.Count) 条记录" -Level INFO

        # 阶段 2: 数据质量检查
        Write-PipelineLog '阶段 2/3: 数据质量检查...' -Level INFO
        $qualityCheck = $processedData | Where-Object {
            $_.Amount -gt 0 -and -not [string]::IsNullOrWhiteSpace($_.Customer)
        }
        $rejected = $processedData.Count - $qualityCheck.Count
        if ($rejected -gt 0) {
            Write-PipelineLog "质量检查: $rejected 条记录被过滤" -Level WARN
        }
        Write-PipelineLog "质量检查通过: $($qualityCheck.Count) 条记录" -Level INFO

        # 阶段 3: 批量写入
        Write-PipelineLog '阶段 3/3: 批量写入目标...' -Level INFO
        $outputDir = './data/output'
        if (-not (Test-Path $outputDir)) {
            New-Item -ItemType Directory -Path $outputDir -Force | Out-Null
        }
        $writeResult = Write-DataBatch -Records $qualityCheck -Destination $outputDir -BatchSize 2

        # 汇总报告
        $runEnd = Get-Date
        $duration = ($runEnd - $runStart).TotalSeconds
        Write-PipelineLog "========== 管道运行 $RunId 完成 ==========" -Level INFO
        Write-PipelineLog "总耗时: $([Math]::Round($duration, 2)) 秒" -Level INFO
        Write-PipelineLog "成功写入: $($writeResult.Success) 条, 失败: $($writeResult.Failed) 条" -Level INFO

    } catch {
        Write-PipelineLog "管道运行异常中断: $($_.Exception.Message)" -Level ERROR
        Write-PipelineLog $_.ScriptStackTrace -Level ERROR
    }
}

# 执行完整管道
Invoke-DataPipeline -RunId 'RUN20260420'

# 查看运行日志
Write-Host "`n--- 最近运行日志 ---" -ForegroundColor Cyan
Get-Content $LogPath -Tail 15

function Convert-DataFormat {
    [CmdletBinding()]
    param(
        [Parameter(Mandatory, ValueFromPipeline)]
        $InputObject,

        [Parameter(Mandatory)]
        [ValidateSet('Json', 'Csv', 'Xml', 'Yaml', 'HashTable')]
        [string]$ToFormat,

        [int]$Depth = 10
    )

    process {
        switch ($ToFormat) {
            'Json' {
                $InputObject | ConvertTo-Json -Depth $Depth -EnumsAsStrings
            }
            'Csv' {
                if ($InputObject -is [string]) {
                    $InputObject | ConvertFrom-Csv | ConvertTo-Csv -NoTypeInformation
                }
                else {
                    $InputObject | ConvertTo-Csv -NoTypeInformation
                }
            }
            'Xml' {
                $InputObject | ConvertTo-Xml -NoTypeInformation -As Stream
            }
            'HashTable' {
                if ($InputObject -is [string]) {
                    $json = $InputObject
                }
                else {
                    $json = $InputObject | ConvertTo-Json -Depth $Depth -Compress
                }
                [System.Text.Json.JsonSerializer]::Deserialize(
                    $json,
                    [System.Collections.Generic.Dictionary[string, object]],
                    (New-Object System.Text.Json.JsonSerializerOptions -Property @{
                        PropertyNameCaseInsensitive = $true
                    })
                )
            }
            'Yaml' {
                # YAML 没有内置 cmdlet，通过 JSON 中转后手动格式化
                $json = if ($InputObject -is [string]) { $InputObject } else {
                    $InputObject | ConvertTo-Json -Depth $Depth
                }
                $obj = $json | ConvertFrom-Json
                ConvertTo-YamlString -InputObject $obj -Indent 0
            }
        }
    }
}

function ConvertTo-YamlString {
    param($InputObject, [int]$Indent = 0)
    $space = '  ' * $Indent
    $sb = [System.Text.StringBuilder]::new()

    if ($InputObject -is [System.Collections.IEnumerable] -and
        $InputObject -isnot [string]) {
        foreach ($item in $InputObject) {
            $null = $sb.Append("${space}- ")
            if ($item -is [System.Collections.IDictionary] -or
                $item -is [PSCustomObject]) {
                $null = $sb.AppendLine()
                $null = $sb.Append((ConvertTo-YamlString $item ($Indent + 1)))
            }
            else {
                $null = $sb.AppendLine("'$item'")
            }
        }
    }
    elseif ($InputObject -is [System.Collections.IDictionary] -or
        $InputObject -is [PSCustomObject]) {
        $props = if ($InputObject -is [System.Collections.IDictionary]) {
            $InputObject.GetEnumerator()
        }
        else {
            $InputObject.PSObject.Properties
        }
        foreach ($prop in $props) {
            $name = $prop.Key ?? $prop.Name
            $val = $prop.Value ?? $prop
            $null = $sb.Append("${space}${name}: ")
            if ($val -is [System.Collections.IEnumerable] -and
                $val -isnot [string]) {
                $null = $sb.AppendLine()
                $null = $sb.Append((ConvertTo-YamlString $val ($Indent + 1)))
            }
            elseif ($val -is [System.Collections.IDictionary] -or
                $val -is [PSCustomObject]) {
                $null = $sb.AppendLine()
                $null = $sb.Append((ConvertTo-YamlString $val ($Indent + 1)))
            }
            else {
                $null = $sb.AppendLine("'$val'")
            }
        }
    }
    else {
        $null = $sb.AppendLine("${space}'$InputObject'")
    }

    $sb.ToString()
}

# 示例：CSV 转 JSON
$csvData = @'
Name,Department,Salary,StartDate
张三,工程部,25000,2024-03-15
李四,市场部,18000,2023-08-20
王五,工程部,30000,2022-11-01
赵六,人事部,22000,2025-01-10
'@

Write-Host "=== CSV 转 JSON ===" -ForegroundColor Cyan
$jsonOutput = $csvData | ConvertFrom-Csv | Convert-DataFormat -ToFormat Json
$jsonOutput

Write-Host "`n=== JSON 转 HashTable ===" -ForegroundColor Cyan
$jsonOutput | Convert-DataFormat -ToFormat HashTable |
    ForEach-Object { $_.GetEnumerator() } |
    Format-Table Key, Value -AutoSize

Write-Host "`n=== CSV 转 XML ===" -ForegroundColor Cyan
$xmlOutput = $csvData | ConvertFrom-Csv | Convert-DataFormat -ToFormat Xml
$xmlOutput | Select-Object -First 10

function Invoke-DataCleanse {
    [CmdletBinding()]
    param(
        [Parameter(Mandatory, ValueFromPipeline)]
        [PSCustomObject]$InputObject,

        # 字段映射表：旧名 → 新名
        [Parameter(Mandatory)]
        [hashtable]$FieldMap,

        # 需要转换为 DateTime 的字段
        [string[]]$DateFields,

        # 需要转换为数值的字段
        [string[]]$NumericFields,

        # 默认空值填充
        $DefaultNullValue = 'N/A'
    )

    begin {
        $seen = [System.Collections.Generic.HashSet[string]]::new()
        $duplicateCount = 0
        $totalProcessed = 0
    }

    process {
        $totalProcessed++

        # 构建去重键（使用所有字段值的拼接）
        $keyParts = $InputObject.PSObject.Properties.Value |
            Where-Object { $_ } |
            ForEach-Object { $_.ToString() }
        $dedupKey = $keyParts -join '|'

        if (-not $seen.Add($dedupKey)) {
            $duplicateCount++
            return
        }

        $result = [ordered]@{}

        foreach ($prop in $InputObject.PSObject.Properties) {
            $fieldName = $prop.Name
            $value = $prop.Value

            # 字段名映射
            if ($FieldMap.ContainsKey($fieldName)) {
                $fieldName = $FieldMap[$fieldName]
            }

            # 空值处理
            if ([string]::IsNullOrWhiteSpace($value)) {
                $value = $DefaultNullValue
            }
            else {
                $value = $value.Trim()

                # 日期字段转换
                if ($DateFields -and $fieldName -in $DateFields) {
                    if ($value -ne $DefaultNullValue) {
                        if ([datetime]::TryParse($value, [ref]$parsed)) {
                            $value = $parsed
                        }
                    }
                }

                # 数值字段转换
                if ($NumericFields -and $fieldName -in $NumericFields) {
                    if ($value -ne $DefaultNullValue) {
                        $cleaned = $value -replace '[,，]', ''
                        if ([double]::TryParse($cleaned, [ref]$num)) {
                            $value = $num
                        }
                    }
                }
            }

            $result[$fieldName] = $value
        }

        [PSCustomObject]$result
    }

    end {
        Write-Verbose "处理完成: 总计 $totalProcessed 条, 去除重复 $duplicateCount 条, 输出 $($totalProcessed - $duplicateCount) 条"
    }
}

# 模拟原始脏数据
$rawData = @'
emp_name,dept,salary,start_date,email
张三,工程部,25000,2024-03-15,zhangsan@corp.com
李四,市场部,18000,2023-08-20,lisi@corp.com
张三,工程部,25000,2024-03-15,zhangsan@corp.com
王五,,30000,,
赵六,人事部,22,000,2025-01-10,zhaoliu@corp.com
孙七,工程部,28000,invalid-date,sunqi@corp.com
'@

$fieldMapping = @{
    'emp_name'   = 'EmployeeName'
    'dept'       = 'Department'
    'salary'     = 'AnnualSalary'
    'start_date' = 'StartDate'
    'email'      = 'ContactEmail'
}

$cleaned = $rawData | ConvertFrom-Csv |
    Invoke-DataCleanse `
        -FieldMap $fieldMapping `
        -DateFields 'StartDate' `
        -NumericFields 'AnnualSalary' `
        -DefaultNullValue '未知' `
        -Verbose

Write-Host "=== 清洗结果 ===" -ForegroundColor Cyan
$cleaned | Format-Table -AutoSize

Write-Host "`n=== 数据类型验证 ===" -ForegroundColor Cyan
$cleaned | Get-Member -MemberType NoteProperty |
    Select-Object Name, @{
        N = 'SampleType'
        E = {
            $val = $cleaned[0].($_.Name)
            if ($null -ne $val) { $val.GetType().Name } else { 'null' }
        }
    } | Format-Table -AutoSize

function Invoke-DataEtlPipeline {
    [CmdletBinding()]
    param(
        [string]$OutputPath = './etl-output',
        [switch]$IncludeSummary
    )

    # ---- Extract 阶段：从多源提取数据 ----

    Write-Host "[Extract] 从 CSV 提取服务器资产数据..." -ForegroundColor Yellow
    $serverCsv = @'
Hostname,IP,OS,CPU_CORES,RAM_GB,STATUS
WEB-01,192.168.1.10,Ubuntu 22.04,4,16,Running
WEB-02,192.168.1.11,Ubuntu 22.04,4,16,Running
DB-01,192.168.1.20,CentOS 7,8,64,Running
DB-02,192.168.1.21,CentOS 7,8,64,Stopped
CACHE-01,192.168.1.30,Ubuntu 22.04,2,8,Running
'@

    $servers = $serverCsv | ConvertFrom-Csv
    Write-Host "  提取 $($servers.Count) 条服务器记录"

    Write-Host "[Extract] 从 JSON 提取监控指标数据..." -ForegroundColor Yellow
    $metricsJson = @'
[
    {"host": "WEB-01", "cpu_pct": 45.2, "mem_pct": 62.1, "disk_pct": 33.7, "timestamp": "2026-02-23T08:00:00Z"},
    {"host": "WEB-02", "cpu_pct": 12.8, "mem_pct": 45.3, "disk_pct": 28.9, "timestamp": "2026-02-23T08:00:00Z"},
    {"host": "DB-01", "cpu_pct": 78.5, "mem_pct": 85.2, "disk_pct": 71.4, "timestamp": "2026-02-23T08:00:00Z"},
    {"host": "CACHE-01", "cpu_pct": 23.1, "mem_pct": 91.5, "disk_pct": 15.2, "timestamp": "2026-02-23T08:00:00Z"}
]
'@

    $metrics = $metricsJson | ConvertFrom-Json
    Write-Host "  提取 $($metrics.Count) 条监控指标"

    Write-Host "[Extract] 从 XML 提取告警配置..." -ForegroundColor Yellow
    $alertXml = @'
<alerts>
    <rule name="cpu_high" threshold="70" severity="critical" target_pattern="DB-*"/>
    <rule name="mem_warning" threshold="80" severity="warning" target_pattern="*"/>
    <rule name="disk_high" threshold="60" severity="warning" target_pattern="DB-*"/>
</alerts>
'@

    [xml]$xmlDoc = $alertXml
    $alertRules = $xmlDoc.alerts.rule | ForEach-Object {
        [PSCustomObject]@{
            RuleName      = $_.name
            Threshold     = [int]$_.threshold
            Severity      = $_.severity
            TargetPattern = $_.target_pattern
        }
    }
    Write-Host "  提取 $($alertRules.Count) 条告警规则"

    # ---- Transform 阶段：关联、转换、计算 ----

    Write-Host "`n[Transform] 关联服务器资产与监控指标..." -ForegroundColor Green

    $enriched = foreach ($srv in $servers) {
        $metric = $metrics | Where-Object { $_.host -eq $srv.Hostname }
        $triggeredAlerts = @()

        foreach ($rule in $alertRules) {
            $pattern = $rule.TargetPattern -replace '\*', '.*'
            if ($srv.Hostname -match "^$pattern$") {
                $violated = $false
                $fieldMap = @{
                    'cpu_high'    = 'cpu_pct'
                    'mem_warning' = 'mem_pct'
                    'disk_high'   = 'disk_pct'
                }
                $fieldName = $fieldMap[$rule.RuleName]
                if ($fieldName -and $metric -and
                    $metric.$fieldName -gt $rule.Threshold) {
                    $violated = $true
                }
                if ($violated) {
                    $triggeredAlerts += "[{0}] {1} ({2}% > {3}%)" -f
                        $rule.Severity, $rule.RuleName,
                        $metric.$fieldName, $rule.Threshold
                }
            }
        }

        [PSCustomObject][ordered]@{
            Hostname   = $srv.Hostname
            IPAddress  = $srv.IP
            OS         = $srv.OS
            CPU_Cores  = [int]$srv.CPU_CORES
            RAM_GB     = [int]$srv.RAM_GB
            Status     = $srv.STATUS
            CPU_Usage  = if ($metric) { $metric.cpu_pct } else { $null }
            MEM_Usage  = if ($metric) { $metric.mem_pct } else { $null }
            DISK_Usage = if ($metric) { $metric.disk_pct } else { $null }
            AlertCount = $triggeredAlerts.Count
            Alerts     = if ($triggeredAlerts) {
                $triggeredAlerts -join '; '
            } else { '无' }
        }
    }

    # ---- Load 阶段：输出到多种目标格式 ----

    $null = New-Item -ItemType Directory -Path $OutputPath -Force -ErrorAction SilentlyContinue

    Write-Host "`n[Load] 导出为 CSV 报表..." -ForegroundColor Magenta
    $csvFile = Join-Path $OutputPath 'server-report.csv'
    $enriched | ConvertTo-Csv -NoTypeInformation | Out-File $csvFile -Encoding utf8
    Write-Host "  已写入: $csvFile"

    Write-Host "[Load] 导出为 JSON..." -ForegroundColor Magenta
    $jsonFile = Join-Path $OutputPath 'server-report.json'
    $enriched | ConvertTo-Json -Depth 5 | Out-File $jsonFile -Encoding utf8
    Write-Host "  已写入: $jsonFile"

    Write-Host "[Load] 导出为 HTML 报表..." -ForegroundColor Magenta
    $htmlFile = Join-Path $OutputPath 'server-report.html'
    $htmlParams = @{
        Title  = '服务器状态报表'
        Body   = '<h1>服务器状态报表</h1>' +
                 '<p>生成时间: {0}</p>' -f (Get-Date -Format 'yyyy-MM-dd HH:mm:ss')
        Head   = '<style>table { border-collapse: collapse; } ' +
                 'td, th { border: 1px solid #ccc; padding: 6px; } ' +
                 '.critical { color: red; font-weight: bold; }</style>'
    }
    $enriched | ConvertTo-Html @htmlParams | Out-File $htmlFile -Encoding utf8
    Write-Host "  已写入: $htmlFile"

    if ($IncludeSummary) {
        Write-Host "`n=== ETL 汇总 ===" -ForegroundColor Cyan
        Write-Host ("  服务器总数: {0}" -f $enriched.Count)
        Write-Host ("  运行中: {0}" -f ($enriched | Where-Object Status -eq 'Running').Count)
        Write-Host ("  已停机: {0}" -f ($enriched | Where-Object Status -eq 'Stopped').Count)
        Write-Host ("  触发告警的服务器: {0}" -f ($enriched | Where-Object AlertCount -gt 0).Count)
        Write-Host ("  平均 CPU 使用率: {0:N1}%" -f (
            ($enriched | Where-Object CPU_Usage | Measure-Object -Property CPU_Usage -Average).Average
        ))
        Write-Host ("  平均内存使用率: {0:N1}%" -f (
            ($enriched | Where-Object MEM_Usage | Measure-Object -Property MEM_Usage -Average).Average
        ))
    }

    return $enriched
}

# 执行完整 ETL 管道
$result = Invoke-DataEtlPipeline -IncludeSummary

Write-Host "`n=== 最终报表 ===" -ForegroundColor Cyan
$result | Format-Table Hostname, IPAddress, OS, Status,
    @{N='CPU%';E={'{0:N1}' -f $_.CPU_Usage}},
    @{N='MEM%';E={'{0:N1}' -f $_.MEM_Usage}},
    @{N='DISK%';E={'{0:N1}' -f $_.DISK_Usage}},
    AlertCount -AutoSize

# 场景：从遗留系统导出的 CSV 数据质量差，需要清洗后再导入新系统

$rawCsv = @"
EmployeeID,Name,Department,Salary,HireDate,Status
E001,张三,技术部,15000.50,2020-03-15,Active
E002,李四,市场部,,2021-07-20,Active
E003,王五,技术部,18000,2022-01-10,Inactive
E004,,财务部,12000,2023/06/01,Active
E005,赵六,技术部,20000.99,invalid,Active
E006,钱七,,16500,2020-11-30,Active
"@

function Import-AndCleanCsv {
    param(
        [Parameter(Mandatory)]
        [string[]]$CsvLines,

        [string]$DefaultDepartment = "未分配",

        [decimal]$DefaultSalary = 0
    )

    # 导入 CSV
    $data = $CsvLines | ConvertFrom-Csv

    $cleaned = foreach ($row in $data) {
        # 清洗姓名：去除空值
        $name = if ([string]::IsNullOrWhiteSpace($row.Name)) {
            "未知员工($($row.EmployeeID))"
        } else {
            $row.Name.Trim()
        }

        # 清洗部门：空值使用默认值
        $dept = if ([string]::IsNullOrWhiteSpace($row.Department)) {
            $DefaultDepartment
        } else {
            $row.Department.Trim()
        }

        # 清洗薪资：非数字使用默认值
        $salary = if ($row.Salary -match '^\d+\.?\d*$') {
            [decimal]$row.Salary
        } else {
            $DefaultSalary
        }

        # 清洗日期：多种格式统一转换
        $hireDate = $null
        if ($row.HireDate -match '^\d{4}-\d{2}-\d{2}$') {
            $hireDate = [datetime]::ParseExact($row.HireDate, "yyyy-MM-dd", $null)
        } elseif ($row.HireDate -match '^\d{4}/\d{2}/\d{2}$') {
            $hireDate = [datetime]::ParseExact($row.HireDate, "yyyy/MM/dd", $null)
        }

        # 输出清洗后的对象
        [PSCustomObject]@{
            EmployeeID = $row.EmployeeID
            Name       = $name
            Department = $dept
            Salary     = $salary
            HireDate   = if ($hireDate) { $hireDate.ToString("yyyy-MM-dd") } else { "N/A" }
            Status     = $row.Status
            Cleaned    = $true
        }
    }

    return $cleaned
}

$cleanData = Import-AndCleanCsv -CsvLines ($rawCsv -split "`n")
$cleanData | Format-Table -AutoSize

# 场景：将扁平的 CSV 数据转换为嵌套的 JSON 结构供 API 消费

$flatData = @"
ID,Name,Email,DeptID,DeptName,RoleID,RoleName
U001,张三,zhangsan@contoso.com,D001,技术部,R001,开发工程师
U002,李四,lisi@contoso.com,D001,技术部,R002,测试工程师
U003,王五,wangwu@contoso.com,D002,市场部,R003,市场经理
U004,赵六,zhaoliu@contoso.com,D002,市场部,R003,市场经理
"@

function ConvertFrom-FlatToNestedJson {
    param(
        [Parameter(Mandatory)]
        [string[]]$CsvLines
    )

    $data = $CsvLines | ConvertFrom-Csv

    # 按部门分组，构建嵌套结构
    $nested = $data | Group-Object -Property DeptID | ForEach-Object {
        $deptGroup = $_

        # 按角色分组
        $roles = $deptGroup.Group | Group-Object -Property RoleID | ForEach-Object {
            $roleGroup = $_
            $members = $roleGroup.Group | ForEach-Object {
                @{
                    id    = $_.ID
                    name  = $_.Name
                    email = $_.Email
                }
            }

            @{
                roleId   = $roleGroup.Name
                roleName = ($roleGroup.Group | Select-Object -First 1).RoleName
                members  = @($members)
            }
        }

        @{
            departmentId   = $deptGroup.Name
            departmentName = ($deptGroup.Group | Select-Object -First 1).DeptName
            roles          = @($roles)
        }
    }

    $result = @{
        exportDate = (Get-Date).ToString("yyyy-MM-ddTHH:mm:ss")
        totalCount = $data.Count
        departments = @($nested)
    }

    return $result | ConvertTo-Json -Depth 5
}

$jsonOutput = ConvertFrom-FlatToNestedJson -CsvLines ($flatData -split "`n")
Write-Host $jsonOutput

# 场景：从 API 获取的嵌套 JSON 需要展开为扁平的 CSV 供报表使用

$apiJson = @"
{
    "users": [
        {
            "id": "U001",
            "profile": {
                "name": "张三",
                "age": 28,
                "address": {
                    "city": "北京",
                    "district": "海淀区"
                }
            },
            "skills": ["PowerShell", "Python", "Docker"],
            "active": true
        },
        {
            "id": "U002",
            "profile": {
                "name": "李四",
                "age": 32,
                "address": {
                    "city": "上海",
                    "district": "浦东新区"
                }
            },
            "skills": ["C#", "SQL Server"],
            "active": false
        }
    ]
}
"@

function Expand-NestedJsonToCsv {
    param(
        [Parameter(Mandatory)]
        [string]$JsonString
    )

    $data = $JsonString | ConvertFrom-Json

    $flatRows = foreach ($user in $data.users) {
        [PSCustomObject]@{
            ID       = $user.id
            Name     = $user.profile.name
            Age      = $user.profile.age
            City     = $user.profile.address.city
            District = $user.profile.address.district
            Skills   = ($user.skills -join "; ")
            SkillCount = $user.skills.Count
            Active   = $user.active
        }
    }

    return $flatRows
}

$flatRows = Expand-NestedJsonToCsv -JsonString $apiJson
$flatRows | Export-Csv -Path "$env:TEMP\users_flat.csv" -NoTypeInformation -Encoding UTF8
Write-Host "已导出到 $env:TEMP\users_flat.csv"
Get-Content "$env:TEMP\users_flat.csv"

# 场景：定期从源系统迁移数据到目标系统，只迁移新增或变更的记录

function Start-IncrementalMigration {
    param(
        [Parameter(Mandatory)]
        [string]$SourcePath,

        [Parameter(Mandatory)]
        [string]$TargetPath,

        [Parameter(Mandatory)]
        [string]$KeyField,

        [string]$CompareField = "UpdatedAt"
    )

    # 加载源数据
    $source = Import-Csv -Path $SourcePath -Encoding UTF8
    $target = if (Test-Path $TargetPath) {
        Import-Csv -Path $TargetPath -Encoding UTF8
    } else {
        @()
    }

    # 构建目标数据的索引
    $targetIndex = @{}
    foreach ($row in $target) {
        $targetIndex[$row.$KeyField] = $row
    }

    $stats = @{
        Total    = $source.Count
        Inserted = 0
        Updated  = 0
        Skipped  = 0
    }

    $result = foreach ($srcRow in $source) {
        $key = $srcRow.$KeyField

        if ($targetIndex.ContainsKey($key)) {
            # 记录已存在，检查是否需要更新
            $tgtRow = $targetIndex[$key]
            $srcVal = $srcRow.$CompareField
            $tgtVal = $tgtRow.$CompareField

            if ($srcVal -gt $tgtVal) {
                # 源数据更新，覆盖目标
                $srcRow | Add-Member -MemberType NoteProperty -Name "_action" -Value "UPDATE" -Force
                $stats.Updated++
                $srcRow
            } else {
                # 无变化，跳过
                $stats.Skipped++
            }
        } else {
            # 新记录，插入
            $srcRow | Add-Member -MemberType NoteProperty -Name "_action" -Value "INSERT" -Force
            $stats.Inserted++
            $srcRow
        }
    }

    # 合并数据：保留未变更的目标记录 + 更新/新增的记录
    $unchanged = $target | Where-Object {
        $key = $_.$KeyField
        $srcKeys = $source | ForEach-Object { $_.$KeyField }
        $key -in $srcKeys -and -not ($result | Where-Object { $_.$KeyField -eq $key })
    }

    $finalData = @($unchanged) + @($result | Where-Object { $_._action -eq "UPDATE" }) +
                  @($result | Where-Object { $_._action -eq "INSERT" })

    # 移除内部字段后导出
    $finalData | ForEach-Object { $_.PSObject.Properties.Remove("_action") }
    $finalData | Export-Csv -Path $TargetPath -NoTypeInformation -Encoding UTF8

    Write-Host "迁移完成：" -ForegroundColor Green
    Write-Host "  总记录数：$($stats.Total)"
    Write-Host "  新增：$($stats.Inserted)" -ForegroundColor Cyan
    Write-Host "  更新：$($stats.Updated)" -ForegroundColor Yellow
    Write-Host "  跳过（无变化）：$($stats.Skipped)" -ForegroundColor Gray

    return $stats
}

# 使用示例（模拟源数据）
$sourceData = @"
ID,Name,Department,UpdatedAt
E001,张三,技术部,2025-08-01
E002,李四,市场部,2025-08-05
E003,王五,财务部,2025-08-06
"@

$targetData = @"
ID,Name,Department,UpdatedAt
E001,张三,技术部,2025-07-15
E002,李四,市场部,2025-08-05
"@

$sourceData | Set-Content "$env:TEMP\source.csv" -Encoding UTF8
$targetData | Set-Content "$env:TEMP\target.csv" -Encoding UTF8

Start-IncrementalMigration -SourcePath "$env:TEMP\source.csv" `
    -TargetPath "$env:TEMP\target.csv" `
    -KeyField "ID" `
    -CompareField "UpdatedAt"

# 场景：批量合并多个 CSV 文件，统一格式后输出为 JSON

function Merge-MultipleCsv {
    param(
        [Parameter(Mandatory)]
        [string]$InputFolder,

        [string]$Filter = "*.csv",

        [string]$OutputPath,

        [string[]]$FieldMapping
    )

    $allData = @()
    $files = Get-ChildItem -Path $InputFolder -Filter $Filter -File

    Write-Host "找到 $($files.Count) 个文件待合并" -ForegroundColor Cyan

    foreach ($file in $files) {
        Write-Host "  处理：$($file.Name)" -ForegroundColor Gray

        $csv = Import-Csv -Path $file.FullName -Encoding UTF8

        foreach ($row in $csv) {
            # 如果指定了字段映射，只提取需要的字段
            if ($FieldMapping) {
                $mappedRow = @{}
                foreach ($field in $FieldMapping) {
                    if ($row.PSObject.Properties[$field]) {
                        $mappedRow[$field] = $row.$field
                    }
                }
                $mappedRow["_sourceFile"] = $file.Name
                $allData += [PSCustomObject]$mappedRow
            } else {
                $row | Add-Member -MemberType NoteProperty -Name "_sourceFile" `
                    -Value $file.Name -Force
                $allData += $row
            }
        }
    }

    Write-Host "合并完成，共 $($allData.Count) 条记录" -ForegroundColor Green

    # 根据输出路径的扩展名决定格式
    if ($OutputPath) {
        $ext = [System.IO.Path]::GetExtension($OutputPath).ToLower()
        switch ($ext) {
            ".json" {
                $allData | ConvertTo-Json -Depth 3 | Set-Content $OutputPath -Encoding UTF8
            }
            ".csv" {
                $allData | Export-Csv -Path $OutputPath -NoTypeInformation -Encoding UTF8
            }
            ".xml" {
                $allData | ConvertTo-Xml -NoTypeInformation | Select-Object -ExpandProperty OuterXml |
                    Set-Content $OutputPath -Encoding UTF8
            }
            default {
                $allData | Export-Csv -Path $OutputPath -NoTypeInformation -Encoding UTF8
            }
        }
        Write-Host "已输出到：$OutputPath" -ForegroundColor Green
    }

    return $allData
}

# 批量格式转换工具
function Convert-DataFormat {
    param(
        [Parameter(Mandatory)]
        [string]$InputPath,

        [Parameter(Mandatory)]
        [string]$OutputPath
    )

    $ext = [System.IO.Path]::GetExtension($InputPath).ToLower()

    # 读取数据
    $data = switch ($ext) {
        ".csv"  { Import-Csv -Path $InputPath -Encoding UTF8 }
        ".json" { Get-Content $InputPath -Raw | ConvertFrom-Json }
        default { throw "不支持的输入格式：$ext" }
    }

    # 如果 JSON 顶层是数组，ConvertFrom-Json 已经返回数组
    # 如果是单个对象，包装为数组
    if ($data -isnot [System.Array]) {
        $data = @($data)
    }

    # 写入目标格式
    $outExt = [System.IO.Path]::GetExtension($OutputPath).ToLower()
    switch ($outExt) {
        ".json" {
            $data | ConvertTo-Json -Depth 5 | Set-Content $OutputPath -Encoding UTF8
        }
        ".csv" {
            $data | Export-Csv -Path $OutputPath -NoTypeInformation -Encoding UTF8
        }
        default {
            throw "不支持的输出格式：$outExt"
        }
    }

    Write-Host "转换完成：$InputPath -> $OutputPath" -ForegroundColor Green
}

# 示例
Convert-DataFormat -InputPath "$env:TEMP\source.csv" -OutputPath "$env:TEMP\source.json"