SunYF
450a094f0e
Dashboard 移除 CNC/清洗机等不存在设备，改为纯装配线看板（扫码枪+拧紧枪+人工）；JWT/内部token统一为 Hardman_2026 写入各 etc/*.yaml，MES 建表 DSN 改读 yaml；event_log 增加 workOrderNo/operator，BOM/备料/PLC下发/拧紧/扫码报工/半成品/AGV 全链路埋点，日志页支持工单号/操作人筛选；MES/WMS 日志按天存储（dailyWriter），WMS SetWriter 用 logx.NewWriter 适配
403 lines
12 KiB
Go
403 lines
12 KiB
Go
package eventlog
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log/slog"
|
||
"sync"
|
||
"time"
|
||
|
||
"bj_power_mes/ent"
|
||
enteventlog "bj_power_mes/ent/eventlog"
|
||
"bj_power_mes/internal/eventbus"
|
||
)
|
||
|
||
// EventLogWriter 异步批量写入事件日志到 PostgreSQL
|
||
type EventLogWriter struct {
|
||
entClient *ent.Client
|
||
ch chan eventbus.Event
|
||
batchSize int
|
||
flushMs int
|
||
|
||
mu sync.Mutex
|
||
running bool
|
||
cancel context.CancelFunc
|
||
wg sync.WaitGroup
|
||
}
|
||
|
||
// NewEventLogWriter 创建事件日志写入器
|
||
func NewEventLogWriter(entClient *ent.Client, batchSize, flushMs int) *EventLogWriter {
|
||
if batchSize <= 0 {
|
||
batchSize = 20
|
||
}
|
||
if flushMs <= 0 {
|
||
flushMs = 1000
|
||
}
|
||
return &EventLogWriter{
|
||
entClient: entClient,
|
||
ch: make(chan eventbus.Event, 256),
|
||
batchSize: batchSize,
|
||
flushMs: flushMs,
|
||
}
|
||
}
|
||
|
||
// Start 启动异步写入循环
|
||
func (w *EventLogWriter) Start(ctx context.Context) {
|
||
w.mu.Lock()
|
||
if w.running {
|
||
w.mu.Unlock()
|
||
return
|
||
}
|
||
w.running = true
|
||
ctx, w.cancel = context.WithCancel(ctx)
|
||
w.mu.Unlock()
|
||
|
||
w.wg.Add(1)
|
||
go w.writeLoop(ctx)
|
||
}
|
||
|
||
// Stop 停止写入器
|
||
func (w *EventLogWriter) Stop() {
|
||
w.mu.Lock()
|
||
if !w.running {
|
||
w.mu.Unlock()
|
||
return
|
||
}
|
||
w.cancel()
|
||
w.mu.Unlock()
|
||
w.wg.Wait()
|
||
}
|
||
|
||
// Write 写入事件（非阻塞，满时丢弃）
|
||
func (w *EventLogWriter) Write(event eventbus.Event) {
|
||
select {
|
||
case w.ch <- event:
|
||
default:
|
||
slog.Warn("eventlog: channel full, dropping event", "type", event.Type)
|
||
}
|
||
}
|
||
|
||
func (w *EventLogWriter) writeLoop(ctx context.Context) {
|
||
defer w.wg.Done()
|
||
|
||
ticker := time.NewTicker(time.Duration(w.flushMs) * time.Millisecond)
|
||
defer ticker.Stop()
|
||
|
||
var batch []eventbus.Event
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
w.flush(batch)
|
||
return
|
||
case event := <-w.ch:
|
||
batch = append(batch, event)
|
||
if len(batch) >= w.batchSize {
|
||
w.flush(batch)
|
||
batch = batch[:0]
|
||
}
|
||
case <-ticker.C:
|
||
if len(batch) > 0 {
|
||
w.flush(batch)
|
||
batch = batch[:0]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (w *EventLogWriter) flush(events []eventbus.Event) {
|
||
if len(events) == 0 {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
|
||
var builders []*ent.EventLogCreate
|
||
for _, e := range events {
|
||
// 连接状态事件仅用于 SSE 实时推送，不持久化
|
||
if e.Type == eventbus.EventConnectionStatusUpdated {
|
||
continue
|
||
}
|
||
// 报警确认仅用于 SSE 实时推送，不持久化
|
||
if e.Type == eventbus.EventAlarmAcked {
|
||
continue
|
||
}
|
||
// MACHINE_DONE: 创建独立事件记录（与 MACHINE_START 配对）
|
||
// 同时更新对应 MACHINE_START 的 endedAt 用于耗时查询
|
||
if e.Type == eventbus.EventMachineDone {
|
||
if endedAt, ok := e.Payload["endedAt"].(time.Time); ok {
|
||
jobId, _ := getPayloadInt(e.Payload, "jobId")
|
||
equipId, _ := getPayloadInt(e.Payload, "equipmentId")
|
||
if jobId > 0 && equipId > 0 {
|
||
w.updateMachineStartEndedAt(ctx, jobId, equipId, endedAt)
|
||
}
|
||
}
|
||
}
|
||
// INSPECTION_RESULT: 同样更新对应 MACHINE_START 的 endedAt
|
||
if e.Type == eventbus.EventInspectionResult {
|
||
if endedAt, ok := e.Payload["endedAt"].(time.Time); ok {
|
||
jobId, _ := getPayloadInt(e.Payload, "jobId")
|
||
equipId, _ := getPayloadInt(e.Payload, "equipmentId")
|
||
if jobId > 0 && equipId > 0 {
|
||
w.updateMachineStartEndedAt(ctx, jobId, equipId, endedAt)
|
||
}
|
||
}
|
||
}
|
||
|
||
b := w.entClient.EventLog.Create().
|
||
SetEventType(string(e.Type)).
|
||
SetSourceId(e.Source).
|
||
SetEntityType(e.EntityType).
|
||
SetEntityId(e.EntityID).
|
||
SetEntityVersion(e.EntityVersion).
|
||
SetDescription(eventDescription(e)).
|
||
SetPayload(e.Payload)
|
||
if v, ok := getPayloadInt(e.Payload, "jobId"); ok {
|
||
b.SetJobId(v)
|
||
}
|
||
if v, ok := e.Payload["workpieceNo"].(string); ok && v != "" {
|
||
b.SetWorkpieceNo(v)
|
||
}
|
||
if v, ok := e.Payload["sn"].(string); ok && v != "" && e.Payload["workpieceNo"] == nil {
|
||
b.SetWorkpieceNo(v) // 工位终端上报以 sn 为工件号
|
||
}
|
||
if v, ok := e.Payload["workOrderNo"].(string); ok && v != "" {
|
||
b.SetWorkOrderNo(v)
|
||
} else if v, ok := e.Payload["orderNo"].(string); ok && v != "" {
|
||
b.SetWorkOrderNo(v)
|
||
}
|
||
if v, ok := e.Payload["operator"].(string); ok && v != "" {
|
||
b.SetOperator(v)
|
||
}
|
||
if v, ok := getPayloadInt(e.Payload, "equipmentId"); ok {
|
||
b.SetEquipmentId(v)
|
||
}
|
||
if v, ok := e.Payload["equipmentName"].(string); ok && v != "" {
|
||
b.SetEquipmentName(v)
|
||
}
|
||
if v, ok := getPayloadInt(e.Payload, "tempSlotNo"); ok {
|
||
b.SetTempSlotNo(v)
|
||
}
|
||
if v, ok := e.Payload["startedAt"].(time.Time); ok {
|
||
b.SetStartedAt(v)
|
||
}
|
||
if v, ok := e.Payload["endedAt"].(time.Time); ok {
|
||
b.SetEndedAt(v)
|
||
}
|
||
builders = append(builders, b)
|
||
}
|
||
if len(builders) > 0 {
|
||
if _, err := w.entClient.EventLog.CreateBulk(builders...).Save(ctx); err != nil {
|
||
slog.Error("eventlog: bulk save failed", "count", len(builders), "error", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// updateMachineStartEndedAt 更新对应 MACHINE_START 事件日志的 endedAt
|
||
func (w *EventLogWriter) updateMachineStartEndedAt(ctx context.Context, jobId, equipId int, endedAt time.Time) {
|
||
n, err := w.entClient.EventLog.Update().
|
||
Where(
|
||
enteventlog.EventTypeEQ(string(eventbus.EventMachineStart)),
|
||
enteventlog.JobIdEQ(jobId),
|
||
enteventlog.EquipmentIdEQ(equipId),
|
||
enteventlog.EndedAtIsNil(),
|
||
).
|
||
SetEndedAt(endedAt).
|
||
Save(ctx)
|
||
if err != nil {
|
||
slog.Error("eventlog: update MACHINE_START endedAt failed",
|
||
"jobId", jobId, "equipmentId", equipId, "error", err)
|
||
return
|
||
}
|
||
if n == 0 {
|
||
slog.Warn("eventlog: no matching MACHINE_START found for endedAt update",
|
||
"jobId", jobId, "equipmentId", equipId)
|
||
}
|
||
}
|
||
|
||
// eventDescription 根据事件类型生成中文描述
|
||
// EventDescription 根据事件类型生成中文描述（供 SSE 桥接等复用）
|
||
func EventDescription(e eventbus.Event) string {
|
||
return eventDescription(e)
|
||
}
|
||
|
||
func eventDescription(e eventbus.Event) string {
|
||
switch e.Type {
|
||
case eventbus.EventLoad:
|
||
return fmt.Sprintf("%s 从 暂存台#%s 上料至 %s", jobLabel(e), tempSlotLabel(e), equipLabel(e))
|
||
case eventbus.EventUnload:
|
||
return fmt.Sprintf("%s 从 %s 下料至 暂存台#%s", jobLabel(e), equipLabel(e), tempSlotLabel(e))
|
||
case eventbus.EventExchange:
|
||
return fmt.Sprintf("%s 在 %s 换料", jobLabel(e), equipLabel(e))
|
||
case eventbus.EventScan:
|
||
return jobLabel(e) + " 扫码"
|
||
case eventbus.EventMark:
|
||
return jobLabel(e) + " 打标"
|
||
case eventbus.EventReplenish:
|
||
return fmt.Sprintf("%s 从 %s 补料至 暂存台#%s", jobLabel(e), dockLabel(e), tempSlotLabel(e))
|
||
case eventbus.EventMachineStart:
|
||
return fmt.Sprintf("%s 在 %s 加工", jobLabel(e), equipLabel(e))
|
||
case eventbus.EventMachineDone:
|
||
desc := fmt.Sprintf("%s 在 %s 加工完成", jobLabel(e), equipLabel(e))
|
||
if dur, ok := e.Payload["duration"].(string); ok && dur != "" {
|
||
desc = fmt.Sprintf("%s 在 %s 加工完成（耗时%s）", jobLabel(e), equipLabel(e), dur)
|
||
}
|
||
return desc
|
||
case eventbus.EventInspectionResult:
|
||
if pass, ok := e.Payload["pass"].(bool); ok {
|
||
if pass {
|
||
return fmt.Sprintf("%s 在 %s 检测合格", jobLabel(e), equipLabel(e))
|
||
}
|
||
return fmt.Sprintf("%s 在 %s 检测不合格", jobLabel(e), equipLabel(e))
|
||
}
|
||
return fmt.Sprintf("%s 在 %s 检测", jobLabel(e), equipLabel(e))
|
||
case eventbus.EventJobCompleted:
|
||
return jobLabel(e) + " 全部工序完成"
|
||
case eventbus.EventJobScrapped:
|
||
return jobLabel(e) + " 报废"
|
||
case eventbus.EventJobDischarged:
|
||
return fmt.Sprintf("%s 从 暂存台#%s 下线至 %s", jobLabel(e), tempSlotLabel(e), dockLabel(e))
|
||
|
||
case eventbus.EventOrderStarted:
|
||
return fmt.Sprintf("%s 启动", orderLabel(e))
|
||
case eventbus.EventOrderPaused:
|
||
return fmt.Sprintf("%s 暂停", orderLabel(e))
|
||
case eventbus.EventOrderResumed:
|
||
return fmt.Sprintf("%s 恢复", orderLabel(e))
|
||
case eventbus.EventOrderCancelled:
|
||
return fmt.Sprintf("%s 取消", orderLabel(e))
|
||
case eventbus.EventOrderCompleted:
|
||
return fmt.Sprintf("%s 完成", orderLabel(e))
|
||
|
||
case eventbus.EventPalletCompleted:
|
||
dockNo, _ := getPayloadInt(e.Payload, "dockNo")
|
||
passQty, _ := getPayloadInt(e.Payload, "passQty")
|
||
ngQty, _ := getPayloadInt(e.Payload, "ngQty")
|
||
return fmt.Sprintf("接驳台#%d 托盘完工（良品%d 不良%d）", dockNo, passQty, ngQty)
|
||
|
||
case eventbus.EventAlarmRaised:
|
||
alarmCode, _ := e.Payload["alarmCode"].(string)
|
||
alarmMsg, _ := e.Payload["alarmMessage"].(string)
|
||
entityLabel := entityLabel(e.EntityType, e.EntityID)
|
||
return fmt.Sprintf("%s 报警: [%s] %s", entityLabel, alarmCode, alarmMsg)
|
||
|
||
case eventbus.EventAlarmAcked:
|
||
entityLabel := entityLabel(e.EntityType, e.EntityID)
|
||
return fmt.Sprintf("%s 报警已确认", entityLabel)
|
||
|
||
case eventbus.EventEquipmentAvailabilityChanged:
|
||
enabled, _ := e.Payload["enabled"].(bool)
|
||
if enabled {
|
||
return fmt.Sprintf("%s 已设为调度可用", equipLabel(e))
|
||
}
|
||
return fmt.Sprintf("%s 已设为调度不可用", equipLabel(e))
|
||
|
||
case eventbus.EventManualActionCreated:
|
||
actionType, _ := e.Payload["actionType"].(string)
|
||
entityLabel := entityLabel(e.EntityType, e.EntityID)
|
||
return fmt.Sprintf("%s 创建人工处理: %s", entityLabel, manualActionTypeDesc(actionType))
|
||
|
||
case eventbus.EventManualActionResolved:
|
||
entityLabel := entityLabel(e.EntityType, e.EntityID)
|
||
return fmt.Sprintf("%s 人工处理已解决", entityLabel)
|
||
|
||
default:
|
||
return fmt.Sprintf("%s: %s", entityLabel(e.EntityType, e.EntityID), string(e.Type))
|
||
}
|
||
}
|
||
|
||
// jobLabel 生成 "工件 #344 (M2026...)" 格式
|
||
func jobLabel(e eventbus.Event) string {
|
||
if v, ok := getPayloadInt(e.Payload, "jobId"); ok && v > 0 {
|
||
if wn, ok := e.Payload["workpieceNo"].(string); ok && wn != "" {
|
||
return fmt.Sprintf("工件 #%d (%s)", v, wn)
|
||
}
|
||
return fmt.Sprintf("工件 #%d", v)
|
||
}
|
||
return fmt.Sprintf("%s #%s", entityLabel(e.EntityType, e.EntityID), e.EntityID)
|
||
}
|
||
|
||
// equipLabel 生成 "CNC#3" 格式
|
||
func equipLabel(e eventbus.Event) string {
|
||
en, _ := e.Payload["equipmentName"].(string)
|
||
if v, ok := getPayloadInt(e.Payload, "equipmentId"); ok && v > 0 {
|
||
if en != "" {
|
||
return fmt.Sprintf("%s#%d", en, v)
|
||
}
|
||
return fmt.Sprintf("设备#%d", v)
|
||
}
|
||
if en != "" {
|
||
return en
|
||
}
|
||
return "设备"
|
||
}
|
||
|
||
// tempSlotLabel 生成 "7" 格式（纯槽位号）
|
||
func tempSlotLabel(e eventbus.Event) string {
|
||
if v, ok := getPayloadInt(e.Payload, "tempSlotNo"); ok && v > 0 {
|
||
return fmt.Sprintf("%d", v)
|
||
}
|
||
if v, ok := getPayloadInt(e.Payload, "slotNo"); ok && v > 0 {
|
||
return fmt.Sprintf("%d", v)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func getPayloadInt(payload map[string]any, key string) (int, bool) {
|
||
if v, ok := payload[key].(float64); ok {
|
||
return int(v), true
|
||
}
|
||
if v, ok := payload[key].(int); ok {
|
||
return v, true
|
||
}
|
||
return 0, false
|
||
}
|
||
|
||
// dockLabel 生成 接驳台1-3 格式
|
||
func dockLabel(e eventbus.Event) string {
|
||
dn, dok := getPayloadInt(e.Payload, "dockNo")
|
||
ds, sok := getPayloadInt(e.Payload, "dockSlotNo")
|
||
if dok && dn > 0 && sok && ds > 0 {
|
||
return fmt.Sprintf("接驳台#%d-%d", dn, ds)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// orderLabel 生成 "工单 #42" 格式
|
||
func orderLabel(e eventbus.Event) string {
|
||
if v, ok := getPayloadInt(e.Payload, "orderId"); ok && v > 0 {
|
||
return fmt.Sprintf("工单 #%d", v)
|
||
}
|
||
return fmt.Sprintf("%s #%s", entityLabel(e.EntityType, e.EntityID), e.EntityID)
|
||
}
|
||
|
||
func entityLabel(entityType, entityID string) string {
|
||
switch entityType {
|
||
case "job":
|
||
return fmt.Sprintf("工件 #%s", entityID)
|
||
case "workorder":
|
||
return fmt.Sprintf("工单 #%s", entityID)
|
||
case "equipment":
|
||
return fmt.Sprintf("设备 #%s", entityID)
|
||
default:
|
||
if entityType != "" && entityID != "" {
|
||
return fmt.Sprintf("%s #%s", entityType, entityID)
|
||
}
|
||
return "系统"
|
||
}
|
||
}
|
||
|
||
func manualActionTypeDesc(s string) string {
|
||
switch s {
|
||
case "SCAN_FAILED":
|
||
return "扫码失败"
|
||
case "RECOVERY_REQUIRED":
|
||
return "需恢复确认"
|
||
case "INSPECTION_DECISION":
|
||
return "检测判定"
|
||
case "FAULT_CONFIRM":
|
||
return "故障确认"
|
||
default:
|
||
return s
|
||
}
|
||
}
|