package repository import ( "database/sql" "errors" "fmt" "strings" "cmautobuy/admin/model" ) // InnerCodeSybSnapshot 是本地顺运宝明细用于解析货运单 stock id 的最小快照。 type InnerCodeSybSnapshot struct { OrderNumber string SybData string } // InnerCodeListFilter 是独立页面可组合的查询条件。 type InnerCodeListFilter struct { BusinessDate string Status string Keyword string } // InnerCodeStatusCounts 是当前业务日期的底栏摘要。 type InnerCodeStatusCounts struct { Total int Ready int NeedsCheck int } // InnerCodeImportOutcome 说明幂等导入是新增还是更新。 type InnerCodeImportOutcome string const ( InnerCodeImportCreated InnerCodeImportOutcome = "created" InnerCodeImportUpdated InnerCodeImportOutcome = "updated" ) // UpsertInnerCodeImportRow 按已确认业务键写入一行。 // 必须在事务中调用;先锁定业务键,避免另一个唯一键冲突时更新错行。 func UpsertInnerCodeImportRow(tx *sql.Tx, row model.InnerCodeImportRow, now string) (InnerCodeImportOutcome, error) { var id int64 err := tx.QueryRow(` SELECT id FROM syb_inner_code_records WHERE business_date=? AND order_number=? AND stall=? AND spec_key=? FOR UPDATE`, row.BusinessDate, row.OrderNumber, row.Stall, row.SpecKey).Scan(&id) if err != nil && !errors.Is(err, sql.ErrNoRows) { return "", fmt.Errorf("锁定档口入库码业务键失败: %w", err) } if errors.Is(err, sql.ErrNoRows) { _, err = tx.Exec(` INSERT INTO syb_inner_code_records ( business_date,source_row,print_sequence,order_number,shop_name,stall, spec_raw,spec_key,inner_code,source_duplicate_count,status, created_by_user_id,created_at,updated_at ) VALUES (?,?,?,?,?,?,?,?,?,?,'pending',?,?,?)`, row.BusinessDate, row.SourceRow, nullablePositiveInt(row.PrintSequence), row.OrderNumber, nullableString(row.ShopName), row.Stall, row.SpecRaw, row.SpecKey, row.InnerCode, row.SourceDuplicateCount, row.CreatedByUserID, now, now) if err != nil { return "", fmt.Errorf("新增档口入库码记录失败: %w", err) } return InnerCodeImportCreated, nil } _, err = tx.Exec(` UPDATE syb_inner_code_records SET source_row=?,print_sequence=?,shop_name=?,spec_raw=?,inner_code=?, source_duplicate_count=?, status=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN status ELSE 'pending' END, stock_id=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN stock_id ELSE NULL END, detail_id=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN detail_id ELSE NULL END, syb_spec=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_spec ELSE NULL END, syb_sku=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_sku ELSE NULL END, syb_variation_sku=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_variation_sku ELSE NULL END, purchase_platform=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN purchase_platform ELSE NULL END, purchase_code=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN purchase_code ELSE NULL END, remote_inner_code=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN remote_inner_code ELSE NULL END, result_message=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN result_message ELSE NULL END, planned_at=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN planned_at ELSE NULL END, updated_at=? WHERE id=?`, row.SourceRow, nullablePositiveInt(row.PrintSequence), nullableString(row.ShopName), row.SpecRaw, row.InnerCode, row.SourceDuplicateCount, now, id) if err != nil { return "", fmt.Errorf("更新档口入库码导入记录失败: %w", err) } return InnerCodeImportUpdated, nil } // ListInnerCodeRecordsForPlanning 返回某日允许重新规划的记录。 func ListInnerCodeRecordsForPlanning(q Execer, businessDate string) ([]model.InnerCodeRecord, error) { rows, err := q.Query(` SELECT id,business_date,source_row,COALESCE(print_sequence,0),order_number, COALESCE(shop_name,''),stall,spec_raw,spec_key,inner_code,source_duplicate_count, status,COALESCE(result_message,''),created_by_user_id,created_at,updated_at FROM syb_inner_code_records WHERE business_date=? AND status IN ('pending','ready','skipped','failed') ORDER BY source_row,id`, businessDate) if err != nil { return nil, fmt.Errorf("查询待规划档口入库码失败: %w", err) } defer rows.Close() result := make([]model.InnerCodeRecord, 0) for rows.Next() { var row model.InnerCodeRecord if err := rows.Scan(&row.ID, &row.BusinessDate, &row.SourceRow, &row.PrintSequence, &row.OrderNumber, &row.ShopName, &row.Stall, &row.SpecRaw, &row.SpecKey, &row.InnerCode, &row.SourceDuplicateCount, &row.Status, &row.ResultMessage, &row.CreatedByUserID, &row.CreatedAt, &row.UpdatedAt); err != nil { return nil, fmt.Errorf("读取待规划档口入库码失败: %w", err) } result = append(result, row) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("遍历待规划档口入库码失败: %w", err) } return result, nil } // ListInnerCodeSybSnapshots 查询订单号对应的本地 SYB 原始快照。 func ListInnerCodeSybSnapshots(q Execer, orderNumbers []string) ([]InnerCodeSybSnapshot, error) { if len(orderNumbers) == 0 { return nil, nil } placeholders := make([]string, len(orderNumbers)) args := make([]any, len(orderNumbers)) for index, orderNumber := range orderNumbers { placeholders[index] = "?" args[index] = orderNumber } rows, err := q.Query(`SELECT order_no,syb_data FROM syb_orders WHERE order_no IN (`+ strings.Join(placeholders, ",")+`) ORDER BY order_no,syb_id`, args...) if err != nil { return nil, fmt.Errorf("查询档口入库码对应顺运宝快照失败: %w", err) } defer rows.Close() result := make([]InnerCodeSybSnapshot, 0) for rows.Next() { var row InnerCodeSybSnapshot if err := rows.Scan(&row.OrderNumber, &row.SybData); err != nil { return nil, fmt.Errorf("读取顺运宝快照失败: %w", err) } result = append(result, row) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("遍历顺运宝快照失败: %w", err) } return result, nil } // SaveInnerCodePlans 在同一事务中保存一批只读规划结果。 func SaveInnerCodePlans(db *sql.DB, plans []model.InnerCodeRecord, plannedAt string) error { tx, err := db.Begin() if err != nil { return fmt.Errorf("开始保存档口入库码规划事务失败: %w", err) } defer tx.Rollback() for _, plan := range plans { result, err := tx.Exec(` UPDATE syb_inner_code_records SET stock_id=?,detail_id=?,syb_spec=?,syb_sku=?,syb_variation_sku=?, purchase_platform=?,purchase_code=?,remote_inner_code=?,status=?, result_message=?,planned_at=?,updated_at=? WHERE id=? AND status IN ('pending','ready','skipped','failed')`, nullablePositiveInt64(plan.StockID), nullablePositiveInt64(plan.DetailID), nullableString(plan.SybSpec), nullableString(plan.SybSKU), nullableString(plan.SybVariationSKU), nullableString(plan.PurchasePlatform), nullableString(plan.PurchaseCode), nullableString(plan.RemoteInnerCode), plan.Status, nullableString(plan.ResultMessage), plannedAt, plannedAt, plan.ID) if err != nil { return fmt.Errorf("保存档口入库码记录 %d 规划失败: %w", plan.ID, err) } if affected, err := result.RowsAffected(); err != nil || affected != 1 { return fmt.Errorf("档口入库码记录 %d 状态已变化,规划整体未保存", plan.ID) } } if err := tx.Commit(); err != nil { return fmt.Errorf("提交档口入库码规划失败: %w", err) } return nil } const innerCodeListColumns = `id,business_date,source_row,COALESCE(print_sequence,0),order_number, COALESCE(shop_name,''),stall,spec_raw,spec_key,inner_code,source_duplicate_count, COALESCE(stock_id,0),COALESCE(detail_id,0),COALESCE(syb_spec,''),COALESCE(syb_sku,''), COALESCE(syb_variation_sku,''),COALESCE(purchase_platform,''),COALESCE(purchase_code,''), COALESCE(remote_inner_code,''),status,COALESCE(result_message,''),created_by_user_id, COALESCE(applied_by_user_id,''),COALESCE(planned_at,''),COALESCE(apply_started_at,''), COALESCE(applied_at,''),created_at,updated_at` func innerCodeFilterClause(filter InnerCodeListFilter) (string, []any) { clauses := []string{"business_date=?"} args := []any{filter.BusinessDate} if filter.Status != "" { clauses = append(clauses, "status=?") args = append(args, filter.Status) } if filter.Keyword != "" { like := "%" + escapeLike(filter.Keyword) + "%" clauses = append(clauses, `(order_number LIKE ? ESCAPE '!' OR stall LIKE ? ESCAPE '!' OR spec_raw LIKE ? ESCAPE '!' OR inner_code LIKE ? ESCAPE '!')`) args = append(args, like, like, like, like) } return " WHERE " + strings.Join(clauses, " AND "), args } // ListInnerCodeRecords 分页读取独立页面记录。 func ListInnerCodeRecords(q Execer, filter InnerCodeListFilter, limit, offset int) ([]model.InnerCodeRecord, error) { where, args := innerCodeFilterClause(filter) query := `SELECT ` + innerCodeListColumns + ` FROM syb_inner_code_records` + where + ` ORDER BY source_row,id LIMIT ? OFFSET ?` args = append(args, limit, offset) rows, err := q.Query(query, args...) if err != nil { return nil, fmt.Errorf("查询档口入库码列表失败: %w", err) } defer rows.Close() result := make([]model.InnerCodeRecord, 0) for rows.Next() { row, err := scanInnerCodeRecord(rows) if err != nil { return nil, err } result = append(result, row) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("遍历档口入库码列表失败: %w", err) } return result, nil } // CountInnerCodeRecords 统计与列表完全相同的筛选结果。 func CountInnerCodeRecords(q Execer, filter InnerCodeListFilter) (int, error) { where, args := innerCodeFilterClause(filter) var count int if err := q.QueryRow(`SELECT COUNT(*) FROM syb_inner_code_records`+where, args...).Scan(&count); err != nil { return 0, fmt.Errorf("统计档口入库码列表失败: %w", err) } return count, nil } // CountInnerCodeStatuses 统计当前业务日期总量、可回写和需核对数量。 func CountInnerCodeStatuses(q Execer, businessDate string) (InnerCodeStatusCounts, error) { var result InnerCodeStatusCounts err := q.QueryRow(`SELECT COUNT(*), COALESCE(SUM(status='ready'),0),COALESCE(SUM(status='needs_check'),0) FROM syb_inner_code_records WHERE business_date=?`, businessDate). Scan(&result.Total, &result.Ready, &result.NeedsCheck) if err != nil { return result, fmt.Errorf("统计档口入库码状态失败: %w", err) } return result, nil } type innerCodeRowScanner interface { Scan(...any) error } func scanInnerCodeRecord(scanner innerCodeRowScanner) (model.InnerCodeRecord, error) { var row model.InnerCodeRecord err := scanner.Scan(&row.ID, &row.BusinessDate, &row.SourceRow, &row.PrintSequence, &row.OrderNumber, &row.ShopName, &row.Stall, &row.SpecRaw, &row.SpecKey, &row.InnerCode, &row.SourceDuplicateCount, &row.StockID, &row.DetailID, &row.SybSpec, &row.SybSKU, &row.SybVariationSKU, &row.PurchasePlatform, &row.PurchaseCode, &row.RemoteInnerCode, &row.Status, &row.ResultMessage, &row.CreatedByUserID, &row.AppliedByUserID, &row.PlannedAt, &row.ApplyStartedAt, &row.AppliedAt, &row.CreatedAt, &row.UpdatedAt) if err != nil { return row, fmt.Errorf("读取档口入库码记录失败: %w", err) } return row, nil } // ClaimInnerCodeForApply 原子领取一条 ready 记录。返回 claimed=false 表示状态已变化。 func ClaimInnerCodeForApply(db *sql.DB, id int64, actorUserID, now string) (*model.InnerCodeRecord, bool, error) { tx, err := db.Begin() if err != nil { return nil, false, fmt.Errorf("开始领取档口入库码事务失败: %w", err) } defer tx.Rollback() record, err := scanInnerCodeRecord(tx.QueryRow(`SELECT `+innerCodeListColumns+ ` FROM syb_inner_code_records WHERE id=? FOR UPDATE`, id)) if errors.Is(err, sql.ErrNoRows) { return nil, false, nil } if err != nil { return nil, false, err } if record.Status != model.InnerCodeReady { return &record, false, nil } result, err := tx.Exec(`UPDATE syb_inner_code_records SET status='applying',applied_by_user_id=?,apply_started_at=?,updated_at=? WHERE id=? AND status='ready'`, actorUserID, now, now, id) if err != nil { return nil, false, fmt.Errorf("领取档口入库码记录失败: %w", err) } if affected, err := result.RowsAffected(); err != nil || affected != 1 { return &record, false, nil } if err := tx.Commit(); err != nil { return nil, false, fmt.Errorf("提交档口入库码领取失败: %w", err) } record.Status = model.InnerCodeApplying record.AppliedByUserID = actorUserID record.ApplyStartedAt = now record.UpdatedAt = now return &record, true, nil } // FinishInnerCodeApply 保存一条已领取记录的最终结果。 func FinishInnerCodeApply(q Execer, id int64, status model.InnerCodeStatus, message, remoteCode, finishedAt string) error { result, err := q.Exec(`UPDATE syb_inner_code_records SET status=?,result_message=?,remote_inner_code=?, applied_at=CASE WHEN ? IN ('updated','already_filled') THEN ? ELSE applied_at END, updated_at=? WHERE id=? AND status='applying'`, status, message, nullableString(remoteCode), status, finishedAt, finishedAt, id) if err != nil { return fmt.Errorf("保存档口入库码回写结果失败: %w", err) } if affected, err := result.RowsAffected(); err != nil || affected != 1 { return fmt.Errorf("档口入库码记录 %d 已不在回写中,拒绝覆盖结果", id) } return nil } // GetInnerCodeForRecheck 读取一条需核对记录。 func GetInnerCodeForRecheck(q Execer, id int64) (*model.InnerCodeRecord, error) { record, err := scanInnerCodeRecord(q.QueryRow(`SELECT `+innerCodeListColumns+ ` FROM syb_inner_code_records WHERE id=?`, id)) if errors.Is(err, sql.ErrNoRows) { return nil, nil } if err != nil { return nil, err } return &record, nil } // SaveInnerCodeRecheck 保存只读重新核对的远端结果,不执行状态领取或写入。 func SaveInnerCodeRecheck(q Execer, id int64, status model.InnerCodeStatus, message, remoteCode, checkedAt string) error { result, err := q.Exec(`UPDATE syb_inner_code_records SET status=?,result_message=?,remote_inner_code=?, applied_at=CASE WHEN ?='updated' THEN ? ELSE applied_at END,updated_at=? WHERE id=? AND status='needs_check'`, status, message, nullableString(remoteCode), status, checkedAt, checkedAt, id) if err != nil { return fmt.Errorf("保存档口入库码核对结果失败: %w", err) } if affected, err := result.RowsAffected(); err != nil || affected != 1 { return fmt.Errorf("档口入库码记录 %d 已不需要核对", id) } return nil } // InterruptApplyingInnerCodes 在启动时把未知结果的 applying 收敛为 needs_check。 func InterruptApplyingInnerCodes(q Execer, interruptedAt string) (int, error) { result, err := q.Exec(`UPDATE syb_inner_code_records SET status='needs_check',result_message='Admin 在回写完成前退出,请重新核对远端结果;系统不会自动重写', updated_at=? WHERE status='applying'`, interruptedAt) if err != nil { return 0, fmt.Errorf("恢复中断的档口入库码回写失败: %w", err) } affected, err := result.RowsAffected() if err != nil { return 0, fmt.Errorf("读取中断档口入库码数量失败: %w", err) } return int(affected), nil } func nullablePositiveInt(value int) any { if value <= 0 { return nil } return value } func nullablePositiveInt64(value int64) any { if value <= 0 { return nil } return value } func nullableString(value string) any { if value == "" { return nil } return value }