Go语言CSV数据导入MS SQL的健壮性实践：解决记录丢失问题

package main

import (
    "database/sql" // 假设go-odbc兼容database/sql接口，或直接使用odbc.Connection
    "fmt"
    "log"

    _ "github.com/alexbrainman/odbc" // 根据实际使用的驱动导入
)

// insertRecord 负责执行单个记录的插入操作
func insertRecord(conn *sql.DB, query string, params []interface{}) error {
    // 注意：此处假设 conn 是 *sql.DB 类型，如果直接使用 go-odbc 的 *odbc.Connection
    // 则需要调整函数签名和 Prepare/Execute 方法的调用。
    // 为简化示例，我们统一使用 *sql.DB 接口。

    stmt, err := conn.Prepare(query)
    if err != nil {
        return fmt.Errorf("prepare statement failed: %w", err)
    }
    // 使用 defer 确保语句在函数返回前关闭，无论成功与否
    defer func() {
        if stmt != nil {
            if closeErr := stmt.Close(); closeErr != nil {
                log.Printf("Error closing statement: %v", closeErr)
            }
        }
    }()

    _, err = stmt.Exec(params...) // 对于 INSERT 操作，通常使用 Exec
    if err != nil {
        return fmt.Errorf("execute statement failed: %w", err)
    }
    return nil
}

package main

import (
    "encoding/csv"
    "fmt"
    "io"
    "log"
    "regexp"
    "strings"
    // "github.com/alexbrainman/odbc" // 如果直接使用 go-odbc
    "database/sql" // 如果使用 database/sql 接口
)

// 假设这些变量在实际应用中已正确初始化
var (
    filename    = "data.csv"
    tablename   = "YourTable"
    numElements = 5 // 假设每行有5个有效数据字段
    fieldNames  = []string{"Col1", "Col2", "Col3", "Col4", "Col5"}
    // dest        *odbc.Connection // 如果直接使用 go-odbc
    dest        *sql.DB // 如果使用 database/sql 接口
)

func main() {
    // ... 初始化 dest (数据库连接) 和 c (csv.Reader) ...
    // 示例：
    // dest, err := sql.Open("odbc", "DSN=YourDSN")
    // if err != nil { log.Fatal(err) }
    // defer dest.Close()
    //
    // file, err := os.Open(filename)
    // if err != nil { log.Fatal(err) }
    // defer file.Close()
    // c := csv.NewReader(file)

    // 模拟初始化 dest 和 c
    // 请替换为实际的数据库连接和CSV文件读取逻辑
    // 假设 dest 已经是一个有效的 *sql.DB 连接
    // 假设 c 已经是一个有效的 *csv.Reader
    // 这里仅为编译通过提供占位符
    log.Println("Initializing dummy database connection and CSV reader...")
    // dest = &sql.DB{} // 替换为实际的数据库连接
    // c = csv.NewReader(strings.NewReader("val1,val2,val3,val4,val5\n'a','b','c','d','e'\n'f','g','h','i','j'"))


    // 实际的CSV读取和数据导入循环
    for {
        record, err := c.Read()
        if err != nil {
            // 如果不是EOF错误，说明读取文件本身出现了问题，应该中断循环
            if err != io.EOF {
                log.Printf("Error while reading %s: %s\n", filename, err)
                break
            }
            // 如果是EOF错误，并且 record 为空，说明文件已经完全读取完毕，可以安全退出
            // 注意：csv.Reader 在遇到 EOF 时，可能会返回一个空 record 和 io.EOF
            // 也可能在返回最后一个有效 record 后，下一次调用才返回 io.EOF
            // 这种处理方式确保了所有有效 record 都会被处理
            if len(record) == 0 {
                break
            }
        }

        // 处理CSV记录，构建SQL插入参数
        re, err := regexp.Compile("^'|'$") // 移除字符串首尾的单引号
        if err != nil {
            log.Printf("Error compiling regex: %v", err)
            continue
        }

        params := make([]interface{}, 0, numElements)
        valueHolders := make([]string, 0, numElements)
        tmpFields := make([]string, 0, numElements)
        count := 0

        // 从 record 中提取有效数据
        for i := 1; i <= numElements; i++ { // 假设 CSV 文件的第一列是行号或不相关数据，从第二列开始处理
            if i >= len(record) { // 防止索引越界
                log.Printf("Record has fewer elements than expected. Expected %d, got %d. Record: %v", numElements, len(record)-1, record)
                break
            }
            tmp := re.ReplaceAllString(record[i], "")

            // 只插入非空值
            if len(tmp) > 0 {
                params = append(params, tmp)
                valueHolders = append(valueHolders, "?")
                tmpFields = append(tmpFields, fieldNames[i-1]) // fieldNames 索引从0开始
                count++
            }
        }

        // 构建SQL插入查询
        query := "insert into [l2test].[dbo]." + tablename +
            " (" + strings.Join(tmpFields, ",") + ")" +
            " values (" + strings.Join(valueHolders, ",") + ")"

        // 调用封装好的插入函数
        err = insertRecord(dest, query, params)
        if err != nil {
            // 记录详细的错误信息，包括查询、参数和原始记录，便于调试
            log.Printf("Failed to insert record:\nError: %v\nQuery: %s\nParams: %v\nRecord: %s\n",
                err, query, params, strings.Join(record, "||"))
            // 根据业务需求，可以选择跳过当前记录继续处理，或直接中断
            continue // 继续处理下一条记录
        }
    }
    log.Println("CSV data import complete.")
}