#Requires -Version 5.1
|
<#
|
22 所检测柜数据的只读取数程序 —— 运行在 22 所那台机器上
|
==========================================================================
|
它做的全部事情（甲方五分钟即可审完）：
|
|
1. 用 Windows 认证连**本机** SQL Server（Integrated Security）。这个文件里没有任何口令，
|
网络上也不传任何厂商库凭据——认证在那台机器内部完成。
|
2. 只执行固定的 SELECT。不写、不建、不改，不调用存储过程。
|
3. 把读到的行**原样**推到我们的接收端点（HTTP，令牌放在请求头里）：不做单位换算、
|
不改列名、不做业务判断。解释（身份映射、时间换算、幂等键）全部在后端完成。
|
4. 只有收到 **2xx** 应答之后才推进自己对该表的读取位置。推送失败或没应答时，
|
整批下一轮重推；端点按来源行身份幂等，重推不会产生重复记录。
|
5. 不开任何入站端口，只发起出站连接。
|
|
查询纪律（SQL 为什么写成这样）：
|
* 所有命令都经同一个 Invoke-Query，它给**每一条**命令设 CommandTimeout；连接串里设
|
Connect Timeout。SQL Server 服务端没有查询超时——客户端不设就是无限，
|
所以两处都必须显式设置，且代码里不存在绕过它的查询路径。
|
* 不加锁读、不长事务：全文没有 FOR UPDATE / HOLDLOCK / XLOCK / UPDLOCK / TABLOCK，
|
没有 BEGIN TRAN，也不读会长期持有的视图。每条语句即取即断。
|
* 增量条件是 record_time > @since，且**刻意不排序**：对没有索引的列排序会让服务端
|
把整张表排序一遍，而那台机器是生产机。排序与截断都在本程序里做（客户端侧）。
|
* 唯一一条 order by 落在 sys.databases 这个系统目录视图上，不碰任何测试表。
|
* 活跃库探测读的是 sys.dm_db_index_usage_stats（目录视图，不是测试表；该实例只有
|
5131 个库、每库一张表，聚合代价有限）。现场可用 sp_help 复核：
|
若发现这轮探测或增量查询的代价偏高，就调整这里的策略。
|
|
发送的列：厂商 IV 表自己的列，原样。要改列清单只改这一处；
|
后端负责一切解释。
|
|
交付注意事项：本文件是 **UTF-8 with BOM**。Windows PowerShell 5.1 会把没有 BOM 的
|
UTF-8 按 GBK 解码，中文注释会破坏脚本（报"字符串缺少终止符"，或者静默什么都不做、
|
退出码却是 0）。**重新保存本文件时必须保留 BOM。**
|
另有两条坑：多行脚本不能粘贴执行（一律跑 .ps1 文件）；record_time 是 Int64，
|
超过 JavaScript 安全整数范围，必须以字符串传输。
|
#>
|
param(
|
# 推送目标，例如 http://<服务器>:8080/device-ingest/j22
|
[Parameter(Mandatory = $true)][string]$Endpoint,
|
# 该来源的令牌；对应值放在后端进程工作目录的 config/vendor-credentials.properties（不入库）
|
[Parameter(Mandatory = $true)][string]$Token,
|
|
[string]$SqlInstance = '.\SQLEXPRESS',
|
[int]$IntervalSec = 60,
|
# 只在第一次看到某张表时生效：从这里往前这么久开始取，而不是读整表历史——
|
# 在生产库上读整表正是查询纪律要避免的
|
[int]$LookbackMinutes = 10,
|
[int]$MaxRowsPerChannel = 500,
|
[int]$ConnectTimeoutSec = 10,
|
[int]$CommandTimeoutSec = 15,
|
[int]$HttpTimeoutSec = 30,
|
[string]$StateFile = (Join-Path $PSScriptRoot 'j22-collector-state.json'),
|
[string]$LogFile = (Join-Path $PSScriptRoot 'j22-collector.log'),
|
[switch]$Once
|
)
|
|
$ErrorActionPreference = 'Continue'
|
|
# ---------------------------------------------------------------- 日志
|
function Write-Log([string]$message) {
|
$line = (Get-Date -Format 'yyyy-MM-dd HH:mm:ss') + ' ' + $message
|
try { Add-Content -LiteralPath $LogFile -Value $line -Encoding UTF8 } catch { }
|
Write-Host $line
|
}
|
|
# ---------------------------------------------------------------- 读取位置
|
# 读取位置按「库里哪张表」记："DB_<...>/Ws<N>_IV_0" -> 端点已确认的最后一条 record_time（FILETIME 刻度）
|
function Get-ReadPositions {
|
if (Test-Path -LiteralPath $StateFile) {
|
try { return Get-Content -LiteralPath $StateFile -Raw | ConvertFrom-Json } catch { }
|
}
|
return New-Object psobject
|
}
|
|
function Save-ReadPositions($positions) {
|
try { $positions | ConvertTo-Json -Depth 4 | Set-Content -LiteralPath $StateFile -Encoding UTF8 }
|
catch { Write-Log ('读取位置未能落盘: ' + $_.Exception.Message) }
|
}
|
|
# ---------------------------------------------------------------- 厂商库访问
|
# 全程序唯一构造 SqlCommand 的地方，所以命令超时不可能被漏掉。所有查询都走这里。
|
# 取值在 reader 打开期间拷出，不留 reader 本身。
|
function Invoke-Query([System.Data.SqlClient.SqlConnection]$connection, [string]$sql, $parameters) {
|
$command = $connection.CreateCommand()
|
$command.CommandText = $sql
|
$command.CommandTimeout = $CommandTimeoutSec
|
if ($parameters) {
|
foreach ($name in $parameters.Keys) { [void]$command.Parameters.AddWithValue($name, $parameters[$name]) }
|
}
|
$reader = $command.ExecuteReader()
|
try {
|
$rows = New-Object System.Collections.ArrayList
|
while ($reader.Read()) {
|
$values = New-Object object[] $reader.FieldCount
|
[void]$reader.GetValues($values)
|
[void]$rows.Add($values)
|
}
|
return $rows
|
}
|
finally { $reader.Close() }
|
}
|
|
function Open-VendorConnection {
|
$connection = New-Object System.Data.SqlClient.SqlConnection(
|
"Server=$SqlInstance;Database=master;Integrated Security=True;Connect Timeout=$ConnectTimeoutSec")
|
$connection.Open()
|
return $connection
|
}
|
|
# 候选库：只要最近写入过的。DMV 读不到时退回按建库时间取最近 60 个。
|
# 两条查询都失败就抛出——那说明这一轮**没有读到厂商库**，因此这一轮不许上报心跳。
|
function Find-ActiveDatabases([System.Data.SqlClient.SqlConnection]$connection) {
|
$databases = New-Object System.Collections.ArrayList
|
$failure = $null
|
try {
|
$rows = Invoke-Query $connection @"
|
select DB_NAME(database_id) as db
|
from sys.dm_db_index_usage_stats
|
where database_id > 4 and DB_NAME(database_id) like 'DB[_]%'
|
group by database_id
|
having max(last_user_update) > dateadd(minute, -$LookbackMinutes, getdate())
|
"@ $null
|
foreach ($row in $rows) { [void]$databases.Add([string]$row[0]) }
|
return $databases
|
}
|
catch {
|
$failure = $_.Exception.Message
|
Write-Log (' 活跃库探测失败，退回按建库时间取库列表: ' + $failure)
|
}
|
|
try {
|
$rows = Invoke-Query $connection @"
|
select top 60 name from sys.databases where name like 'DB[_]%' order by create_date desc
|
"@ $null
|
foreach ($row in $rows) { [void]$databases.Add([string]$row[0]) }
|
return $databases
|
}
|
catch {
|
throw "vendor database not readable: $failure / " + $_.Exception.Message
|
}
|
}
|
|
function Find-ChannelTables([System.Data.SqlClient.SqlConnection]$connection, [string]$database) {
|
$tables = New-Object System.Collections.ArrayList
|
$rows = Invoke-Query $connection "select name from [$database].sys.tables where name like 'Ws%[_]IV[_]0'" $null
|
foreach ($row in $rows) { [void]$tables.Add([string]$row[0]) }
|
return $tables
|
}
|
|
# ---------------------------------------------------------------- 推送
|
# 只有 2xx 才返回 $true：调用方据此决定是否推进读取位置，所以没被确认的批次整批下一轮重推
|
function Send-Payload([string]$json) {
|
try {
|
Invoke-RestMethod -Uri $Endpoint -Method Post -Body $json `
|
-ContentType 'application/json; charset=utf-8' `
|
-Headers @{ 'X-Device-Ingest-Token' = $Token } `
|
-TimeoutSec $HttpTimeoutSec | Out-Null
|
return $true
|
}
|
catch {
|
$status = ''
|
try { $status = [int]$_.Exception.Response.StatusCode } catch { }
|
Write-Log (" 推送失败（状态=$status）: " + $_.Exception.Message)
|
return $false
|
}
|
}
|
|
function Send-Samples([string]$database, [string]$table, [int]$channel, $rows) {
|
$payload = [pscustomobject]@{
|
kind = 'SAMPLE'
|
database = $database
|
table = $table
|
channel = $channel
|
rows = $rows
|
} | ConvertTo-Json -Depth 6 -Compress
|
return Send-Payload $payload
|
}
|
|
function Send-Topology($channels) {
|
if ($channels.Count -eq 0) { return $true }
|
$rows = @()
|
foreach ($channel in $channels) { $rows += [pscustomobject]@{ channel = $channel } }
|
$payload = [pscustomobject]@{ kind = 'TOPOLOGY'; rows = $rows } | ConvertTo-Json -Depth 4 -Compress
|
return Send-Payload $payload
|
}
|
|
# 心跳只表示"程序还活着"。它在一轮成功读到厂商库之后才发，
|
# 所以厂商库连不上（或那台机器权限/网络有问题）时来源显示离线，而不是"绿色但没数据"。
|
function Send-Heartbeat {
|
return Send-Payload '{"kind":"HEARTBEAT"}'
|
}
|
|
# ---------------------------------------------------------------- 一轮
|
function Invoke-Cycle([System.Data.SqlClient.SqlConnection]$connection) {
|
$positions = Get-ReadPositions
|
$seenChannels = New-Object System.Collections.ArrayList
|
$sentRows = 0
|
|
foreach ($database in (Find-ActiveDatabases $connection)) {
|
foreach ($table in (Find-ChannelTables $connection $database)) {
|
if ($table -notmatch '^Ws(\d+)_IV_0$') { continue }
|
$channel = [int]$Matches[1]
|
if (-not $seenChannels.Contains($channel)) { [void]$seenChannels.Add($channel) }
|
|
$key = "$database/$table"
|
$since = 0L
|
if ($positions.PSObject.Properties.Name -contains $key) { $since = [long]$positions.$key }
|
else {
|
# 第一次看到这张表：从"现在往前 LookbackMinutes"开始，不读整表历史
|
$since = [DateTime]::UtcNow.AddMinutes(-$LookbackMinutes).ToFileTimeUtc()
|
}
|
|
$rows = @()
|
try {
|
$read = Invoke-Query $connection @"
|
select record_time, Data_Point, Schedule_Index, Step_Index, Cycle_Index, Step_Time,
|
Voltage, eCurrent, ChargeCapacity, DischargeCapacity, ChargeEnergy, DischargeEnergy
|
from [$database].dbo.[$table]
|
where record_time > @since
|
"@ @{ '@since' = $since }
|
foreach ($row in $read) {
|
# 厂商列可能为 NULL：直接强转 [double]/[int] 会抛异常、让整个通道从此不再推进。
|
# 缺列一律转成 JSON null，后端把测量列的缺失当"设备没给"，不写零代替缺失。
|
$recordTime = To-Text $row[0]
|
$dataPoint = To-Int $row[1]
|
if ($null -eq $recordTime -or $null -eq $dataPoint) {
|
# 这两个是记录位置与幂等键：缺了无法归属，只能跳过并留下日志
|
Write-Log " 跳过一个缺少 record_time 或 Data_Point 的行 [$key]"
|
continue
|
}
|
# record_time 是 Int64，超过 JavaScript 安全整数范围：必须以字符串传输
|
$rows += [pscustomobject]@{
|
record_time = $recordTime
|
Data_Point = $dataPoint
|
Schedule_Index = To-Text $row[2]
|
Step_Index = To-Int $row[3]
|
Cycle_Index = To-Int $row[4]
|
Step_Time = To-Num $row[5]
|
Voltage = To-Num $row[6]
|
eCurrent = To-Num $row[7]
|
ChargeCapacity = To-Num $row[8]
|
DischargeCapacity = To-Num $row[9]
|
ChargeEnergy = To-Num $row[10]
|
DischargeEnergy = To-Num $row[11]
|
}
|
}
|
}
|
catch { Write-Log (" 读取失败 [$key]: " + $_.Exception.Message); continue }
|
|
if ($rows.Count -eq 0) { continue }
|
|
# 排序与截断在这里做：服务端不做排序
|
$rows = @($rows | Sort-Object { [long]$_.record_time } | Select-Object -First $MaxRowsPerChannel)
|
$highest = [long]$rows[-1].record_time
|
|
if (-not (Send-Samples $database $table $channel $rows)) { continue }
|
$positions | Add-Member -NotePropertyName $key -NotePropertyValue $highest -Force
|
$sentRows += $rows.Count
|
Write-Log (" 已推 {0,4} 行 ch={1} {2}" -f $rows.Count, $channel, $key)
|
}
|
}
|
|
# 拓扑：这台机器当前有数据表的通道。端点忽略重复声明，所以每轮都发没有副作用；
|
# 它同时是候选"最近数据时间"的来源之外的通道清单（最近数据时间取采样行）。
|
[void](Send-Topology $seenChannels)
|
Save-ReadPositions $positions
|
return $sentRows
|
}
|
|
# ---------------------------------------------------------------- 主循环
|
Write-Log '===== 取数程序启动 ====='
|
Write-Log "端点 = $Endpoint"
|
Write-Log "间隔 = ${IntervalSec}s，回溯 = ${LookbackMinutes}min，实例 = $SqlInstance"
|
|
$connection = $null
|
while ($true) {
|
try {
|
if ($null -eq $connection -or $connection.State -ne 'Open') {
|
$connection = Open-VendorConnection
|
Write-Log '已连上本机 SQL Server（只读，Windows 认证）'
|
}
|
$sent = Invoke-Cycle $connection
|
# 这一轮读到了厂商库：这是"来源还在"的证据
|
if (Send-Heartbeat) { Write-Log "本轮结束，共推 $sent 行" }
|
}
|
catch {
|
Write-Log ('本轮出错: ' + $_.Exception.Message)
|
try { if ($connection) { $connection.Close() } } catch { }
|
$connection = $null
|
}
|
|
if ($Once) { break }
|
Start-Sleep -Seconds $IntervalSec
|
}
|
