1273 lines
28 KiB
Go
1273 lines
28 KiB
Go
package main
|
||
|
||
/*
|
||
#include <stdlib.h>
|
||
*/
|
||
import "C"
|
||
import (
|
||
"encoding/csv"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
"unsafe"
|
||
)
|
||
|
||
// CSVManager CSV全局管理器
|
||
type CSVManager struct {
|
||
files sync.Map // handle -> *CSVFile (并发安全的映射)
|
||
nextHandle int64 // 生成唯一句柄
|
||
errorMsg string // 全局错误信息
|
||
mu sync.RWMutex // 管理器级别的锁
|
||
}
|
||
|
||
// CSVFile 文件结构
|
||
type CSVFile struct {
|
||
filename string // 实际文件路径
|
||
handle int64 // 唯一句柄
|
||
delimiter rune // 分隔符（如 ','）
|
||
hasHeader bool // 是否有标题行
|
||
|
||
// 内存数据缓存
|
||
data [][]string // 内存中的数据行
|
||
header []string // 标题行（如果有）
|
||
fileSize int64 // 文件大小
|
||
modified bool // 标记是否被修改
|
||
|
||
// 并发控制
|
||
mu sync.RWMutex // 文件级锁
|
||
rowLocks []*sync.RWMutex // 行级锁
|
||
rowMu sync.Mutex // 行锁数组保护
|
||
readers int32 // 活跃读取器计数
|
||
writers int32 // 活跃写入器计数
|
||
saving int32 // 正在保存的goroutine计数
|
||
saveErr chan error // 保存错误通道
|
||
done chan struct{} // 关闭信号
|
||
}
|
||
|
||
// 单例模式的管理器
|
||
var manager *CSVManager
|
||
var once sync.Once
|
||
|
||
// newCSVFile 创建新的CSVFile对象
|
||
func newCSVFile(filename string, delimiter rune, hasHeader bool) *CSVFile {
|
||
return &CSVFile{
|
||
filename: filename,
|
||
delimiter: delimiter,
|
||
hasHeader: hasHeader,
|
||
data: make([][]string, 0),
|
||
rowLocks: make([]*sync.RWMutex, 0),
|
||
saveErr: make(chan error, 1),
|
||
done: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// getManager 获取全局管理器(单例)
|
||
func getManager() *CSVManager {
|
||
once.Do(func() {
|
||
manager = &CSVManager{
|
||
nextHandle: 1,
|
||
}
|
||
})
|
||
return manager
|
||
}
|
||
|
||
// setError 设置错误信息
|
||
func setError(err string) {
|
||
mgr := getManager()
|
||
mgr.mu.Lock()
|
||
mgr.errorMsg = err
|
||
mgr.mu.Unlock()
|
||
}
|
||
|
||
// 初始化管理器
|
||
func initCSVManager() int64 {
|
||
_ = getManager()
|
||
return 0 // 成功
|
||
}
|
||
|
||
// 打开/创建CSV文件(句柄)
|
||
func openCSVFile(filename string, delimiter rune, hasHeader bool) int64 {
|
||
// 创建CSV全局管理器
|
||
mgr := getManager()
|
||
|
||
// 先检查文件是否已经打开
|
||
var existingHandle int64 = -1
|
||
mgr.files.Range(func(key, value interface{}) bool {
|
||
file := value.(*CSVFile)
|
||
// 比较文件路径（使用绝对路径确保一致性）
|
||
absFilename, _ := filepath.Abs(filename)
|
||
absExisting, _ := filepath.Abs(file.filename)
|
||
|
||
if absFilename == absExisting {
|
||
fmt.Println("absFilename", absFilename)
|
||
fmt.Println("absExisting", absExisting)
|
||
fmt.Println("bool", absFilename == absExisting)
|
||
|
||
existingHandle = key.(int64)
|
||
return false // 停止遍历
|
||
}
|
||
return true // 继续遍历
|
||
})
|
||
|
||
// 如果文件已经打开，返回现有句柄
|
||
if existingHandle != -1 {
|
||
fmt.Printf("文件已打开，返回现有句柄: %d\n", existingHandle)
|
||
return existingHandle
|
||
}
|
||
|
||
// 生成句柄
|
||
handle := atomic.AddInt64(&mgr.nextHandle, 1)
|
||
|
||
// 创建文件对象
|
||
file := newCSVFile(filename, delimiter, hasHeader)
|
||
file.handle = handle
|
||
|
||
// 加载文件数据
|
||
if err := file.load(); err != nil {
|
||
setError(err.Error())
|
||
return -1
|
||
}
|
||
|
||
// 存储到管理器
|
||
mgr.files.Store(handle, file)
|
||
return handle
|
||
}
|
||
|
||
// 读取多行数据
|
||
func readRows(handle int64, startRow, count int64, buffer []byte) int64 {
|
||
mgr := getManager()
|
||
|
||
// 获取文件对象
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("Invalid file handle")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
atomic.AddInt32(&file.readers, 1)
|
||
defer atomic.AddInt32(&file.readers, -1)
|
||
|
||
// 获取读取锁
|
||
file.mu.RLock()
|
||
defer file.mu.RUnlock()
|
||
|
||
totalRows := int64(len(file.data))
|
||
if startRow < 0 || startRow >= totalRows {
|
||
return 0
|
||
}
|
||
|
||
// 计算实际读取行数
|
||
endRow := startRow + count
|
||
if endRow > totalRows {
|
||
endRow = totalRows
|
||
}
|
||
|
||
rowsToRead := endRow - startRow
|
||
|
||
// 将数据复制到缓冲区
|
||
bytesWritten := 0
|
||
|
||
for i := startRow; i < endRow; i++ {
|
||
row := file.data[i]
|
||
|
||
// 获取行读锁
|
||
if rowLock := file.getRowLock(int(i)); rowLock != nil {
|
||
rowLock.RLock()
|
||
}
|
||
|
||
for _, cell := range row {
|
||
// 写入单元格长度
|
||
cellLen := len(cell)
|
||
if bytesWritten+4 > len(buffer) {
|
||
break
|
||
}
|
||
|
||
// 写入4字节长度
|
||
buffer[bytesWritten] = byte(cellLen & 0xFF)
|
||
buffer[bytesWritten+1] = byte((cellLen >> 8) & 0xFF)
|
||
buffer[bytesWritten+2] = byte((cellLen >> 16) & 0xFF)
|
||
buffer[bytesWritten+3] = byte((cellLen >> 24) & 0xFF)
|
||
bytesWritten += 4
|
||
|
||
// 写入单元格数据
|
||
if bytesWritten+cellLen > len(buffer) {
|
||
// 缓冲区不足，回退长度写入
|
||
bytesWritten -= 4
|
||
break
|
||
}
|
||
|
||
copy(buffer[bytesWritten:bytesWritten+cellLen], cell)
|
||
bytesWritten += cellLen
|
||
}
|
||
|
||
// 写入行结束标记
|
||
if bytesWritten+4 <= len(buffer) {
|
||
// 4字节的0表示行结束
|
||
buffer[bytesWritten] = 0
|
||
buffer[bytesWritten+1] = 0
|
||
buffer[bytesWritten+2] = 0
|
||
buffer[bytesWritten+3] = 0
|
||
bytesWritten += 4
|
||
}
|
||
|
||
// 释放行锁
|
||
if rowLock := file.getRowLock(int(i)); rowLock != nil {
|
||
rowLock.RUnlock()
|
||
}
|
||
|
||
if bytesWritten >= len(buffer) {
|
||
break
|
||
}
|
||
}
|
||
|
||
return rowsToRead
|
||
}
|
||
|
||
// 写入/覆盖行数据
|
||
func writeRows(handle int64, rowData [][]string, header int) int64 {
|
||
mgr := getManager()
|
||
|
||
// 获取文件对象
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("文件无效句柄！")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
|
||
// 获取写入锁
|
||
file.mu.Lock()
|
||
defer file.mu.Unlock()
|
||
|
||
atomic.AddInt32(&file.writers, 1)
|
||
defer atomic.AddInt32(&file.writers, -1)
|
||
|
||
// 清空现有数据（因为这是覆盖写入）
|
||
file.data = make([][]string, 0, len(rowData))
|
||
if header == 0 {
|
||
file.header = rowData[0]
|
||
// 添加新数据
|
||
file.data = rowData[1:]
|
||
} else {
|
||
// 添加新数据
|
||
for _, row := range rowData {
|
||
file.data = append(file.data, row)
|
||
}
|
||
}
|
||
|
||
// 扩展行锁数组
|
||
file.rowMu.Lock()
|
||
file.rowLocks = make([]*sync.RWMutex, len(file.data))
|
||
for i := range file.rowLocks {
|
||
file.rowLocks[i] = &sync.RWMutex{}
|
||
}
|
||
file.rowMu.Unlock()
|
||
|
||
file.modified = true
|
||
// 异步保存到文件
|
||
go file.saveAsync()
|
||
return int64(len(file.data))
|
||
}
|
||
|
||
// 加载CSV文件到内存
|
||
func (f *CSVFile) load() error {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
|
||
// 打开文件
|
||
file, err := os.Open(f.filename)
|
||
if err != nil {
|
||
// 文件不存在则创建空文件
|
||
if os.IsNotExist(err) {
|
||
f.data = make([][]string, 0) // 初始化空数据
|
||
f.rowLocks = make([]*sync.RWMutex, 0) // 初始化空行锁数组
|
||
return nil
|
||
}
|
||
return fmt.Errorf("文件不存在: %v", err)
|
||
}
|
||
// 确保函数结束时关闭文件
|
||
defer file.Close()
|
||
|
||
// 获取文件大小
|
||
stat, _ := file.Stat() // 获取文件信息
|
||
f.fileSize = stat.Size() // 获取文件大小
|
||
|
||
// 创建CSV读取器
|
||
reader := csv.NewReader(file)
|
||
reader.Comma = f.delimiter // 设置分隔符（默认逗号）
|
||
reader.LazyQuotes = true // 允许宽松的引号解析
|
||
reader.TrimLeadingSpace = true // 去除字段前的空格
|
||
|
||
// 关键修改：允许变长记录，不强制检查字段数量
|
||
reader.FieldsPerRecord = -1
|
||
firstRecord, err := reader.Read()
|
||
if err != nil {
|
||
if err == io.EOF {
|
||
// 空文件
|
||
f.data = make([][]string, 0)
|
||
f.rowLocks = make([]*sync.RWMutex, 0)
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
maxColumns := len(firstRecord)
|
||
allData := [][]string{firstRecord}
|
||
|
||
// 读取剩余行
|
||
for {
|
||
record, err := reader.Read()
|
||
if err == io.EOF {
|
||
break
|
||
}
|
||
if err != nil {
|
||
// 对于有问题的行，可以填充或跳过
|
||
continue
|
||
}
|
||
|
||
// 确保所有行都有相同的列数
|
||
if len(record) < maxColumns {
|
||
// 填充缺失的列为空字符串
|
||
paddedRecord := make([]string, maxColumns)
|
||
copy(paddedRecord, record)
|
||
for i := len(record); i < maxColumns; i++ {
|
||
paddedRecord[i] = ""
|
||
}
|
||
record = paddedRecord
|
||
} else if len(record) > maxColumns {
|
||
// 如果行有更多列，更新最大列数
|
||
maxColumns = len(record)
|
||
// 重新处理之前的所有行
|
||
for i := range allData {
|
||
if len(allData[i]) < maxColumns {
|
||
paddedRecord := make([]string, maxColumns)
|
||
copy(paddedRecord, allData[i])
|
||
allData[i] = paddedRecord
|
||
}
|
||
}
|
||
}
|
||
|
||
allData = append(allData, record)
|
||
}
|
||
|
||
if len(allData) == 0 {
|
||
f.data = make([][]string, 0)
|
||
f.rowLocks = make([]*sync.RWMutex, 0)
|
||
return nil
|
||
}
|
||
|
||
// 处理表头
|
||
if f.hasHeader && len(allData) > 0 {
|
||
f.header = allData[0]
|
||
f.data = allData[1:]
|
||
} else {
|
||
f.data = allData
|
||
}
|
||
|
||
// 初始化行锁
|
||
f.initRowLocks()
|
||
|
||
return nil
|
||
}
|
||
|
||
// initRowLocks 初始化行锁数组
|
||
func (f *CSVFile) initRowLocks() {
|
||
count := len(f.data)
|
||
f.rowLocks = make([]*sync.RWMutex, count)
|
||
for i := 0; i < count; i++ {
|
||
f.rowLocks[i] = &sync.RWMutex{}
|
||
}
|
||
}
|
||
|
||
// getRowLock 获取行锁（线程安全）
|
||
func (f *CSVFile) getRowLock(rowIndex int) *sync.RWMutex {
|
||
if rowIndex < 0 {
|
||
return nil
|
||
}
|
||
|
||
f.rowMu.Lock()
|
||
defer f.rowMu.Unlock()
|
||
|
||
// 确保行锁存在
|
||
if rowIndex >= len(f.rowLocks) {
|
||
// 扩展行锁数组
|
||
newLocks := make([]*sync.RWMutex, rowIndex+1)
|
||
copy(newLocks, f.rowLocks)
|
||
for i := len(f.rowLocks); i <= rowIndex; i++ {
|
||
newLocks[i] = &sync.RWMutex{}
|
||
}
|
||
f.rowLocks = newLocks
|
||
}
|
||
|
||
return f.rowLocks[rowIndex]
|
||
}
|
||
|
||
// save 保存到文件
|
||
func (f *CSVFile) save() error {
|
||
// 标记正在保存
|
||
if !atomic.CompareAndSwapInt32(&f.saving, 0, 1) {
|
||
// 已经在保存中，直接返回
|
||
return nil
|
||
}
|
||
defer atomic.StoreInt32(&f.saving, 0)
|
||
|
||
// 如果没有修改，直接返回
|
||
f.mu.RLock()
|
||
if !f.modified {
|
||
f.mu.RUnlock()
|
||
return nil
|
||
}
|
||
// 复制数据
|
||
dataCopy := make([][]string, len(f.data))
|
||
|
||
for i := range f.data {
|
||
dataCopy[i] = make([]string, len(f.data[i]))
|
||
copy(dataCopy[i], f.data[i])
|
||
}
|
||
// 复制表头
|
||
headerCopy := make([]string, len(f.header))
|
||
copy(headerCopy, f.header)
|
||
// 复制配置(值类型，直接赋值)
|
||
hasHeader := f.hasHeader
|
||
delimiter := f.delimiter
|
||
filename := f.filename
|
||
f.mu.RUnlock() // 释放读锁
|
||
|
||
// 创建临时文件(使用不同的扩展名避免冲突)
|
||
tempFile := filename + ".tmp"
|
||
file, err := os.Create(tempFile)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 创建CSV写入器
|
||
writer := csv.NewWriter(file)
|
||
writer.Comma = delimiter
|
||
|
||
// 写入数据
|
||
if hasHeader && len(headerCopy) > 0 {
|
||
if err := writer.Write(headerCopy); err != nil {
|
||
file.Close()
|
||
os.Remove(tempFile)
|
||
return err
|
||
}
|
||
}
|
||
|
||
if err := writer.WriteAll(dataCopy); err != nil {
|
||
file.Close()
|
||
os.Remove(tempFile)
|
||
return err
|
||
}
|
||
|
||
writer.Flush()
|
||
if err := writer.Error(); err != nil {
|
||
file.Close()
|
||
os.Remove(tempFile)
|
||
return err
|
||
}
|
||
|
||
// 关闭文件，确保数据写入磁盘
|
||
if err := file.Close(); err != nil {
|
||
os.Remove(tempFile)
|
||
return err
|
||
}
|
||
|
||
// 尝试重命名，如果失败可能是文件被占用
|
||
var renameErr error
|
||
for retry := 0; retry < 3; retry++ {
|
||
renameErr = os.Rename(tempFile, filename)
|
||
if renameErr == nil {
|
||
break
|
||
}
|
||
time.Sleep(100 * time.Millisecond) // 等待重试
|
||
}
|
||
|
||
if renameErr != nil {
|
||
os.Remove(tempFile)
|
||
return renameErr
|
||
}
|
||
|
||
// 标记为已保存
|
||
f.mu.Lock()
|
||
f.modified = false
|
||
f.mu.Unlock()
|
||
|
||
return nil
|
||
}
|
||
|
||
// saveAsync 异步保存，带错误处理
|
||
func (f *CSVFile) saveAsync() {
|
||
select {
|
||
case f.saveErr <- f.save():
|
||
// 保存完成
|
||
default:
|
||
// 通道已满，忽略错误
|
||
}
|
||
}
|
||
|
||
// 追加行数据
|
||
func appendRows(handle int64, rowsData []byte, rowCount int64) int {
|
||
mgr := getManager()
|
||
|
||
// 获取文件对象
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("Invalid file handle")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
|
||
// 获取写入锁
|
||
file.mu.Lock()
|
||
atomic.AddInt32(&file.writers, 1)
|
||
|
||
// 解析行数据
|
||
offset := 0
|
||
|
||
for i := int64(0); i < rowCount; i++ {
|
||
var row []string
|
||
|
||
// 读取行数据
|
||
for {
|
||
if offset+4 > len(rowsData) {
|
||
break
|
||
}
|
||
|
||
cellLen := int(uint32(rowsData[offset]) |
|
||
uint32(rowsData[offset+1])<<8 |
|
||
uint32(rowsData[offset+2])<<16 |
|
||
uint32(rowsData[offset+3])<<24)
|
||
offset += 4
|
||
|
||
if cellLen == 0 {
|
||
break
|
||
}
|
||
|
||
if offset+cellLen > len(rowsData) {
|
||
break
|
||
}
|
||
|
||
cell := string(rowsData[offset : offset+cellLen])
|
||
offset += cellLen
|
||
row = append(row, cell)
|
||
}
|
||
|
||
// 追加到数据
|
||
file.data = append(file.data, row)
|
||
}
|
||
|
||
file.modified = true
|
||
|
||
atomic.AddInt32(&file.writers, -1)
|
||
file.mu.Unlock()
|
||
|
||
// 异步保存
|
||
go file.saveAsync()
|
||
|
||
// 获取总行数
|
||
count := getRowCount(handle)
|
||
fmt.Println("count:", count)
|
||
return int(count)
|
||
}
|
||
|
||
// 打开/创建CSV文件(句柄)
|
||
func createOpenCSVFile(filename string, delimiter rune, hasHeader bool) int64 {
|
||
// 创建CSV全局管理器
|
||
mgr := getManager()
|
||
|
||
// 生成句柄
|
||
handle := atomic.AddInt64(&mgr.nextHandle, 1)
|
||
|
||
// 创建文件对象
|
||
file := newCSVFile(filename, delimiter, hasHeader)
|
||
file.handle = handle
|
||
|
||
//// 加载文件数据
|
||
//if err := file.load(); err != nil {
|
||
// setError(err.Error())
|
||
// return -1
|
||
//}
|
||
|
||
// 存储到管理器
|
||
mgr.files.Store(handle, file)
|
||
|
||
return handle
|
||
}
|
||
|
||
// updateCSVRowSafe 修改csv文件行数据
|
||
func updateCSVRowSafe(handleID int64, rowNum int, newRow []string) error {
|
||
mgr := getManager()
|
||
|
||
// 获取文件对象
|
||
val, ok := mgr.files.Load(handleID)
|
||
if !ok {
|
||
setError("Invalid file handle")
|
||
return fmt.Errorf("无效句柄: %d", handleID)
|
||
}
|
||
file := val.(*CSVFile)
|
||
|
||
// 1. 创建临时文件
|
||
tempDir := filepath.Dir(file.filename)
|
||
if tempDir == "" {
|
||
tempDir = "."
|
||
}
|
||
|
||
tempFile, err := os.CreateTemp(tempDir, "temp_*.csv")
|
||
if err != nil {
|
||
return fmt.Errorf("创建临时文件失败: %v", err)
|
||
}
|
||
tempFileName := tempFile.Name()
|
||
|
||
// 确保临时文件被关闭和清理
|
||
defer func() {
|
||
tempFile.Close()
|
||
// 如果临时文件还存在（替换失败），则清理它
|
||
if _, err := os.Stat(tempFileName); err == nil {
|
||
os.Remove(tempFileName)
|
||
}
|
||
}()
|
||
|
||
// 2. 读取原始文件
|
||
sourceFile, err := os.Open(file.filename)
|
||
if err != nil {
|
||
return fmt.Errorf("打开源文件失败: %v", err)
|
||
}
|
||
defer sourceFile.Close()
|
||
|
||
// 3. 读取并处理数据
|
||
reader := csv.NewReader(sourceFile)
|
||
allRows, err := reader.ReadAll()
|
||
if err != nil {
|
||
return fmt.Errorf("读取CSV失败: %v", err)
|
||
}
|
||
|
||
// 4. 验证并更新
|
||
if rowNum < 1 || rowNum > len(allRows) {
|
||
return fmt.Errorf("行号 %d 超出范围 (1-%d)", rowNum, len(allRows))
|
||
}
|
||
|
||
// 显示修改前后的对比
|
||
oldRow := allRows[rowNum-1]
|
||
fmt.Printf("修改前第 %d 行: %v\n", rowNum, oldRow)
|
||
fmt.Printf("修改后第 %d 行: %v\n", rowNum, newRow)
|
||
|
||
allRows[rowNum-1] = newRow
|
||
|
||
// 5. 写入临时文件
|
||
writer := csv.NewWriter(tempFile)
|
||
if err := writer.WriteAll(allRows); err != nil {
|
||
return fmt.Errorf("写入临时文件失败: %v", err)
|
||
}
|
||
writer.Flush()
|
||
|
||
if err := writer.Error(); err != nil {
|
||
return fmt.Errorf("刷新写入失败: %v", err)
|
||
}
|
||
|
||
// 6. 确保数据写入磁盘
|
||
if err := tempFile.Sync(); err != nil {
|
||
return fmt.Errorf("同步文件失败: %v", err)
|
||
}
|
||
tempFile.Close()
|
||
|
||
// 7. 使用复制而不是重命名（解决Windows文件占用问题）
|
||
if err := copyFile(tempFileName, file.filename); err != nil {
|
||
return fmt.Errorf("复制文件失败: %v", err)
|
||
}
|
||
|
||
fmt.Printf("✅ 成功更新第 %d 行，文件已直接更新\n", rowNum)
|
||
|
||
return nil
|
||
}
|
||
|
||
// copyFile 复制文件内容
|
||
func copyFile(src, dst string) error {
|
||
// 打开源文件
|
||
source, err := os.Open(src)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer source.Close()
|
||
|
||
// 创建目标文件
|
||
destination, err := os.Create(dst)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer destination.Close()
|
||
|
||
// 复制内容
|
||
_, err = io.Copy(destination, source)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 确保数据写入磁盘
|
||
err = destination.Sync()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// 获取总行数
|
||
func getRowCount(handle int64) int64 {
|
||
mgr := getManager()
|
||
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("Invalid file handle")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
file.mu.RLock()
|
||
defer file.mu.RUnlock()
|
||
if file.header != nil {
|
||
return int64(len(file.data) + 1)
|
||
}
|
||
return int64(len(file.data))
|
||
}
|
||
|
||
// 搜索行
|
||
func findRows(handle int64, searchText string, columnIndex int64, resultBuffer []byte, maxResults int64) int64 {
|
||
mgr := getManager()
|
||
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("Invalid file handle")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
|
||
file.mu.RLock()
|
||
defer file.mu.RUnlock()
|
||
|
||
atomic.AddInt32(&file.readers, 1)
|
||
defer atomic.AddInt32(&file.readers, -1)
|
||
|
||
var foundRows []int64
|
||
|
||
// 搜索行
|
||
for i, row := range file.data {
|
||
if maxResults > 0 && int64(len(foundRows)) >= maxResults {
|
||
break
|
||
}
|
||
|
||
// 获取行读锁
|
||
if rowLock := file.getRowLock(i); rowLock != nil {
|
||
rowLock.RLock()
|
||
}
|
||
|
||
// 检查列
|
||
if columnIndex < 0 || columnIndex >= int64(len(row)) {
|
||
// 搜索所有列
|
||
for _, cell := range row {
|
||
if cell == searchText {
|
||
foundRows = append(foundRows, int64(i))
|
||
break
|
||
}
|
||
}
|
||
} else if row[columnIndex] == searchText {
|
||
foundRows = append(foundRows, int64(i))
|
||
}
|
||
|
||
// 释放行锁
|
||
if rowLock := file.getRowLock(i); rowLock != nil {
|
||
rowLock.RUnlock()
|
||
}
|
||
}
|
||
|
||
// 写入结果到缓冲区
|
||
if resultBuffer != nil && len(foundRows) > 0 {
|
||
bytesWritten := 0
|
||
|
||
for _, rowIndex := range foundRows {
|
||
if bytesWritten+8 > len(resultBuffer) {
|
||
break
|
||
}
|
||
|
||
// 写入行索引（8字节）
|
||
for j := 0; j < 8; j++ {
|
||
resultBuffer[bytesWritten] = byte((rowIndex >> (uint(j) * 8)) & 0xFF)
|
||
bytesWritten++
|
||
}
|
||
}
|
||
}
|
||
|
||
return int64(len(foundRows))
|
||
}
|
||
|
||
// 合并多个CSV文件（线程安全）
|
||
func mergeCSVFiles(handles []int64, outputFilename string, delimiter rune, hasHeader bool) int64 {
|
||
mgr := getManager()
|
||
|
||
// 验证所有句柄并获取文件对象
|
||
files := make([]*CSVFile, 0, len(handles))
|
||
for _, handle := range handles {
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("Invalid file handle: " + string(handle))
|
||
return -1
|
||
}
|
||
files = append(files, val.(*CSVFile))
|
||
}
|
||
|
||
// 创建输出文件对象
|
||
outputFile := newCSVFile(outputFilename, delimiter, hasHeader)
|
||
outputFile.modified = true // 标记为需要保存
|
||
|
||
// 第一步：合并表头
|
||
mergedHeader := make([]string, 0)
|
||
headerSet := make(map[string]bool)
|
||
|
||
if hasHeader {
|
||
// 收集所有不重复的表头
|
||
for _, file := range files {
|
||
file.mu.RLock()
|
||
if file.hasHeader && len(file.header) > 0 {
|
||
for _, h := range file.header {
|
||
if !headerSet[h] {
|
||
headerSet[h] = true
|
||
mergedHeader = append(mergedHeader, h)
|
||
}
|
||
}
|
||
}
|
||
file.mu.RUnlock()
|
||
}
|
||
|
||
// 如果没有找到表头，创建一个默认的表头
|
||
if len(mergedHeader) == 0 && len(files) > 0 {
|
||
files[0].mu.RLock()
|
||
maxColumns := len(files[0].data[0])
|
||
files[0].mu.RUnlock()
|
||
|
||
for i := 0; i < maxColumns; i++ {
|
||
mergedHeader = append(mergedHeader, fmt.Sprintf("Column%d", i+1))
|
||
}
|
||
}
|
||
}
|
||
outputFile.header = mergedHeader
|
||
|
||
// 第二步：合并数据
|
||
mergedData := make([][]string, 0)
|
||
|
||
// 为每个输入文件创建读取锁并并发读取
|
||
var wg sync.WaitGroup
|
||
dataChan := make(chan [][]string, len(files))
|
||
errorChan := make(chan error, len(files))
|
||
|
||
for idx, file := range files {
|
||
wg.Add(1)
|
||
go func(fileIdx int, f *CSVFile) {
|
||
defer wg.Done()
|
||
|
||
// 获取文件的读取锁
|
||
f.mu.RLock()
|
||
defer f.mu.RUnlock()
|
||
|
||
// 增加活跃读取器计数
|
||
atomic.AddInt32(&f.readers, 1)
|
||
defer atomic.AddInt32(&f.readers, -1)
|
||
|
||
// 读取数据
|
||
fileData := make([][]string, len(f.data))
|
||
for i := 0; i < len(f.data); i++ {
|
||
// 获取行读锁
|
||
if rowLock := f.getRowLock(i); rowLock != nil {
|
||
rowLock.RLock()
|
||
}
|
||
|
||
// 复制行数据
|
||
row := make([]string, len(f.data[i]))
|
||
copy(row, f.data[i])
|
||
|
||
// 释放行锁
|
||
if rowLock := f.getRowLock(i); rowLock != nil {
|
||
rowLock.RUnlock()
|
||
}
|
||
|
||
fileData[i] = row
|
||
}
|
||
|
||
// 如果需要处理表头映射
|
||
if hasHeader && f.hasHeader && len(f.header) > 0 {
|
||
// 创建列映射：源列 -> 目标列
|
||
columnMapping := make(map[int]int)
|
||
for srcIdx, srcHeader := range f.header {
|
||
for dstIdx, dstHeader := range mergedHeader {
|
||
if srcHeader == dstHeader {
|
||
columnMapping[srcIdx] = dstIdx
|
||
break
|
||
}
|
||
}
|
||
}
|
||
|
||
// 重新排列数据以匹配合并后的表头
|
||
for i := 0; i < len(fileData); i++ {
|
||
newRow := make([]string, len(mergedHeader))
|
||
for srcIdx, dstIdx := range columnMapping {
|
||
if srcIdx < len(fileData[i]) {
|
||
newRow[dstIdx] = fileData[i][srcIdx]
|
||
}
|
||
}
|
||
fileData[i] = newRow
|
||
}
|
||
} else if hasHeader && len(mergedHeader) > 0 {
|
||
// 文件没有表头，但输出需要表头
|
||
// 简单地将数据填充到对应位置
|
||
for i := 0; i < len(fileData); i++ {
|
||
newRow := make([]string, len(mergedHeader))
|
||
for j := 0; j < len(fileData[i]) && j < len(newRow); j++ {
|
||
newRow[j] = fileData[i][j]
|
||
}
|
||
fileData[i] = newRow
|
||
}
|
||
}
|
||
|
||
// 将处理后的数据发送到通道
|
||
dataChan <- fileData
|
||
}(idx, file)
|
||
}
|
||
|
||
// 等待所有goroutine完成
|
||
go func() {
|
||
wg.Wait()
|
||
close(dataChan)
|
||
close(errorChan)
|
||
}()
|
||
|
||
// 收集所有数据
|
||
for data := range dataChan {
|
||
mergedData = append(mergedData, data...)
|
||
}
|
||
|
||
// 检查是否有错误
|
||
select {
|
||
case err := <-errorChan:
|
||
if err != nil {
|
||
setError("Error merging files: " + err.Error())
|
||
return -1
|
||
}
|
||
default:
|
||
}
|
||
|
||
// 第三步：设置输出文件的数据
|
||
outputFile.data = mergedData
|
||
outputFile.initRowLocks()
|
||
|
||
// 第四步：保存到文件
|
||
if err := outputFile.save(); err != nil {
|
||
setError("Error saving merged file: " + err.Error())
|
||
return -1
|
||
}
|
||
|
||
// 第五步：将输出文件添加到管理器
|
||
handle := atomic.AddInt64(&mgr.nextHandle, 1)
|
||
outputFile.handle = handle
|
||
mgr.files.Store(handle, outputFile)
|
||
|
||
return handle
|
||
}
|
||
|
||
// 关闭文件
|
||
func closeCSVFile(handle int64) int64 {
|
||
mgr := getManager()
|
||
|
||
val, ok := mgr.files.Load(handle)
|
||
if !ok {
|
||
setError("文件句柄无效！")
|
||
return -1
|
||
}
|
||
|
||
file := val.(*CSVFile)
|
||
|
||
// 等待所有读写操作完成
|
||
for atomic.LoadInt32(&file.readers) > 0 || atomic.LoadInt32(&file.writers) > 0 {
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
|
||
// 等待异步保存完成
|
||
for i := 0; i < 100; i++ { // 最多等待100ms
|
||
if atomic.LoadInt32(&file.saving) == 0 {
|
||
break
|
||
}
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
|
||
// 如果有正在进行的保存，等待一小段时间
|
||
if atomic.LoadInt32(&file.saving) > 0 {
|
||
time.Sleep(50 * time.Millisecond)
|
||
}
|
||
|
||
// 检查是否需要保存
|
||
file.mu.RLock()
|
||
needSave := file.modified
|
||
file.mu.RUnlock()
|
||
|
||
if needSave {
|
||
// 同步保存修改
|
||
if err := file.save(); err != nil {
|
||
setError(err.Error())
|
||
return -1
|
||
}
|
||
}
|
||
|
||
// 从管理器移除
|
||
mgr.files.Delete(handle)
|
||
|
||
//// 安全关闭通道（如果存在）
|
||
//if file.done != nil {
|
||
// close(file.done)
|
||
//}
|
||
|
||
return 0
|
||
}
|
||
|
||
// 获取错误信息
|
||
func getError() string {
|
||
mgr := getManager()
|
||
mgr.mu.RLock()
|
||
defer mgr.mu.RUnlock()
|
||
|
||
if mgr.errorMsg == "" {
|
||
return ""
|
||
}
|
||
|
||
err := mgr.errorMsg
|
||
mgr.errorMsg = "" // 清空错误
|
||
|
||
return err
|
||
}
|
||
|
||
func parseSimpleTable(goData string) [][]string {
|
||
lines := strings.Split(strings.TrimSpace(goData), "\n")
|
||
result := make([][]string, len(lines))
|
||
|
||
for i, line := range lines {
|
||
// 根据你的分隔符分割，这里用逗号举例
|
||
fields := strings.Split(line, ",")
|
||
// 如果需要去除每个字段的空格
|
||
for j := range fields {
|
||
fields[j] = strings.TrimSpace(fields[j])
|
||
}
|
||
result[i] = fields
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
// ============ C 导出接口 ============
|
||
|
||
//export InitCSVManager
|
||
func InitCSVManager() C.longlong {
|
||
return C.longlong(initCSVManager())
|
||
}
|
||
|
||
// OpenCSVFile 打开/创建CSV文件
|
||
//
|
||
//export OpenCSVFile
|
||
func OpenCSVFile(filename *C.char, delimiter C.char, hasHeader C.int) C.longlong {
|
||
return C.longlong(openCSVFile(C.GoString(filename), rune(delimiter), hasHeader != 0))
|
||
}
|
||
|
||
// CSV响应结构体
|
||
type CSVResponse struct {
|
||
Success bool `json:"success"`
|
||
Message string `json:"message,omitempty"`
|
||
Data interface{} `json:"data,omitempty"`
|
||
}
|
||
|
||
// UpdateCSVRowSafe 修改csv文件行数据
|
||
//
|
||
//export UpdateCSVRowSafe
|
||
func UpdateCSVRowSafe(handleID C.longlong, rowNum C.int, newRow *C.char) *C.char {
|
||
goHandle := int64(handleID)
|
||
goRowNum := int(rowNum)
|
||
goNewRow := C.GoString(newRow)
|
||
var row []string
|
||
var csvResponse CSVResponse
|
||
// 解析JSON
|
||
err := json.Unmarshal([]byte(goNewRow), &row)
|
||
if err != nil {
|
||
csvResponse = CSVResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf("解析JSON失败: %v", err),
|
||
}
|
||
errorJson, _ := json.Marshal(csvResponse)
|
||
return C.CString(string(errorJson))
|
||
}
|
||
// 修改csv行数据
|
||
err = updateCSVRowSafe(goHandle, goRowNum, row)
|
||
if err != nil {
|
||
csvResponse = CSVResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf(err.Error()),
|
||
}
|
||
errorJson, _ := json.Marshal(csvResponse)
|
||
return C.CString(string(errorJson))
|
||
} else {
|
||
csvResponse = CSVResponse{
|
||
Success: true,
|
||
}
|
||
}
|
||
// 转换为JSON字符串
|
||
jsonData, err := json.Marshal(csvResponse)
|
||
if err != nil {
|
||
// 如果JSON序列化失败，返回错误信息
|
||
csvResponse = CSVResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf("JSON序列化失败: %v", err),
|
||
}
|
||
errorJson, _ := json.Marshal(csvResponse)
|
||
return C.CString(string(errorJson))
|
||
}
|
||
return C.CString(string(jsonData))
|
||
}
|
||
|
||
//export CreateOpenCSVFile
|
||
func CreateOpenCSVFile(filename *C.char, delimiter C.char, hasHeader C.int) *C.char {
|
||
goFilename := C.GoString(filename)
|
||
goDelimiter := rune(delimiter)
|
||
var goHasHeader bool
|
||
if int(hasHeader) == 0 {
|
||
goHasHeader = true
|
||
}
|
||
goHasHeader = false
|
||
handle := createOpenCSVFile(goFilename, goDelimiter, goHasHeader)
|
||
var csvResponse CSVResponse
|
||
response := struct {
|
||
HandleID int64 `json:"handleID"`
|
||
}{
|
||
HandleID: handle,
|
||
}
|
||
csvResponse.Success = true
|
||
csvResponse.Data = response
|
||
jsonData, _ := json.Marshal(csvResponse)
|
||
return C.CString(string(jsonData))
|
||
}
|
||
|
||
// ReadRows 读取多行数据
|
||
//
|
||
//export ReadRows
|
||
func ReadRows(handle C.longlong, startRow C.longlong, count C.longlong, buffer *C.char, bufferSize C.int) C.longlong {
|
||
// 将 C 缓冲区转换为 Go 的字节切片
|
||
goBuffer := unsafe.Slice((*byte)(unsafe.Pointer(buffer)), int(bufferSize))
|
||
result := readRows(int64(handle), int64(startRow), int64(count), goBuffer)
|
||
return C.longlong(result)
|
||
}
|
||
|
||
// WriteRows 写入/覆盖行数据
|
||
//
|
||
//export WriteRows
|
||
func WriteRows(handle C.longlong, rowsData *C.char, header C.int) C.int {
|
||
goData := C.GoString(rowsData)
|
||
goHeader := int(header)
|
||
//data := parseSimpleTable(goData)
|
||
var data [][]string
|
||
err := json.Unmarshal([]byte(goData), &data)
|
||
if err != nil {
|
||
setError(err.Error())
|
||
}
|
||
result := writeRows(int64(handle), data, goHeader)
|
||
return C.int(result)
|
||
}
|
||
|
||
// AppendRows 追加行数据
|
||
//
|
||
//export AppendRows
|
||
func AppendRows(handle C.longlong, rowsData *C.char, dataSize C.int, rowCount C.longlong) C.int {
|
||
goData := unsafe.Slice((*byte)(unsafe.Pointer(rowsData)), int(dataSize))
|
||
result := appendRows(int64(handle), goData, int64(rowCount))
|
||
return C.int(result)
|
||
}
|
||
|
||
// GetRowCount 获取总行数
|
||
//
|
||
//export GetRowCount
|
||
func GetRowCount(handle C.longlong) C.longlong {
|
||
result := getRowCount(int64(handle))
|
||
return C.longlong(result)
|
||
}
|
||
|
||
// FindRows 搜索行
|
||
//
|
||
//export FindRows
|
||
func FindRows(handle C.longlong, searchText *C.char, columnIndex C.longlong, resultBuffer *C.char, bufferSize C.int, maxResults C.longlong) C.longlong {
|
||
goSearchText := C.GoString(searchText)
|
||
goResultBuffer := unsafe.Slice((*byte)(unsafe.Pointer(resultBuffer)), int(bufferSize))
|
||
result := findRows(int64(handle), goSearchText, int64(columnIndex), goResultBuffer, int64(maxResults))
|
||
return C.longlong(result)
|
||
}
|
||
|
||
// MergeCSVFiles 合并多个CSV文件(线程安全)
|
||
//
|
||
//export MergeCSVFiles
|
||
func MergeCSVFiles(handlesPtr *C.longlong, handlesCount C.int, outputFilename *C.char, delimiter C.char, hasHeader C.int) C.longlong {
|
||
// 将C数组转换为Go切片
|
||
goHandles := unsafe.Slice(handlesPtr, int(handlesCount))
|
||
handles := make([]int64, len(goHandles))
|
||
for i := 0; i < len(goHandles); i++ {
|
||
handles[i] = int64(goHandles[i])
|
||
}
|
||
// 调用合并函数
|
||
result := mergeCSVFiles(handles, C.GoString(outputFilename), rune(delimiter), hasHeader != 0)
|
||
return C.longlong(result)
|
||
}
|
||
|
||
// CloseCSVFile 关闭文件
|
||
//
|
||
//export CloseCSVFile
|
||
func CloseCSVFile(handle C.longlong) C.int {
|
||
result := closeCSVFile(int64(handle))
|
||
return C.int(result)
|
||
}
|
||
|
||
// GetError 获取错误信息
|
||
//
|
||
//export GetError
|
||
func GetError(buffer *C.char, bufferSize C.int) C.int {
|
||
errMsg := getError()
|
||
if errMsg == "" {
|
||
return 0
|
||
}
|
||
|
||
// 将错误信息复制到缓冲区
|
||
goBuffer := unsafe.Slice((*byte)(unsafe.Pointer(buffer)), int(bufferSize))
|
||
n := copy(goBuffer, errMsg)
|
||
return C.int(n)
|
||
}
|
||
|
||
// 导出函数：释放C字符串内存
|
||
//
|
||
//export FreeCString
|
||
func FreeCString(str *C.char) {
|
||
C.free(unsafe.Pointer(str))
|
||
}
|
||
|
||
// CSV_VERSION 版本号
|
||
const (
|
||
CSV_VERSION = "v1"
|
||
)
|
||
|
||
// 获取版本信息
|
||
//
|
||
//export GetVersion
|
||
func GetVersion() *C.char {
|
||
return C.CString(CSV_VERSION)
|
||
}
|
||
|
||
// main 函数是必需的，即使为空
|
||
func main() {
|
||
}
|