工业级全功能 LPeg CSV 解析与面向对象操作库

csv.lua

-- =====================================================================================
-- Filename:    csv.lua
-- Description: 工业级全功能 LPeg CSV 解析与面向对象操作库 (终极无懈可击版)
-- Architecture: 基于 Parsing Expression Grammars (LPeg) 的纯 Lua 极速解析引擎
--
-- 核心工业级特性：
-- 1. 极致性能优化：将常用全局函数全面局部变量化 (Localize Globals)，消除虚拟机高频查找开销。
-- 2. 强力防静默截断：在 LPeg 主文法尾部引入严格的 EOF (* -1) 锚定，杜绝语法畸变时静默产出残缺数据。
-- 3. 幽灵空行斩断：采用负向前瞻断言 -(nl + -1) 与空行海绵 nl^1，从文法底层斩断 {""} 幽灵行。
-- 4. 防死循环编译保护：标准行强制绑定 nl^1，静态保证 ^0 循环至少消耗 1 字符，彻底杜绝 LPeg 编译报错。
-- 5. 全面换行兼容：完美兼容 Windows (\r\n)、Linux/macOS (\n) 以及老式 Mac (\r) 格式。
-- 6. 无状态迭代器：重构 lines() 为零内存分配 (Zero-Allocation) 迭代器，消除高频遍历时的 GC 压力。
-- 7. 二进制安全 I/O：文件读写强制采用 "rb" 和 "wb" 模式，规避操作系统对换行符的隐式篡改。
-- 8. 流式分片追加：核心解析引擎支持对现有实例多次追加解析内容，增强了网络流与分片处理能力。
-- =====================================================================================

local lpeg = require("lpeg")

-- =====================================================================================
-- [性能优化] 将常用全局函数与标准库函数缓存为局部变量 (Localize Globals)
-- 虚拟机在高频循环访问局部变量时直接读取寄存器 (GETLOCAL)，速度远快于查询全局哈希表
-- =====================================================================================
local type, tostring, ipairs, error = type, tostring, ipairs, error
local table_insert  = table.insert
local table_concat  = table.concat
local table_sort    = table.sort
local string_format = string.format
local io_open       = io.open
local unpack        = unpack or table.unpack -- 完美兼容 Lua 5.1 至 Lua 5.4+

-- =====================================================================================
-- 第一部分：底层 LPeg 核心状态机文法定义 (零宽防漏、安全锁定)
-- =====================================================================================
local P, C, Cs, Ct, S = lpeg.P, lpeg.C, lpeg.Cs, lpeg.Ct, lpeg.S

local comma = P(",")

-- 跨平台换行符兼容规则：优先匹配 \r\n，其次匹配 \n，最后兼容旧式 Mac 的 \r
local nl    = P("\r")^-1 * P("\n") + P("\r") 
local quote = P('"')

-- 规则 A: 普通无引号字段 (匹配并捕获不包含逗号、双引号、换行符的连续字符)
local plain_field = C((1 - S(',"\r\n'))^0)

-- 规则 B: 引号包裹字段 (支持原生零临时表脱壳转义 "" -> ")
-- Cs (替换捕获) 配合 / 能够在 LPeg 匹配层面直接完成符号清洗，杜绝 string.gsub 产生的内存碎片
local escaped_quote = P('""') / '"'
local non_quote     = 1 - quote
local qcontent      = Cs((escaped_quote + non_quote)^0)
local quoted_field  = quote * qcontent * quote

-- 单元格字段综合规则 (优先匹配引号包裹字段，若不匹配则视为普通字段)
local field       = quoted_field + plain_field

-- 单行记录组装规则 (标准 CSV 单行：首字段 + 0到多个“逗号+字段”)
local record      = Ct(field * (comma * field)^0)

-- =====================================================================================
-- 【核心杀招：幽灵空行封杀 & 防零字符死循环 & 防静默截断】
-- =====================================================================================
-- 1. 负向前瞻断言 -(nl + -1)：
--    解决 0 字符可匹配陷阱。如果当前游标正对着换行符或文件结尾(EOF)，说明眼前是个纯空行或文件终点！
--    在起手瞬间直接拒绝匹配，绝对不给 record 零宽匹配生成 {""} 幽灵行的任何机会。
local valid_record = -(nl + -1) * record

-- 2. 标准行 (std_record)：
--    有效记录 + 1到多个连续换行符 (nl^1 化身海绵，顺便吞吃行尾所有多余空行)。
--    【编译保护】：因为 nl^1 静态保证了每次匹配至少消耗 1 个换行字符！
--    所以将它放入 ^0 循环中时，LPeg 绝对不会再报 "may accept empty string" 错误！
local std_record   = valid_record * nl^1

-- 3. 尾部行 (tail_record)：
--    有效记录 + 文件末尾 (-1)。专门捕获最后一行没有结尾换行符的特殊情况。
--    【防漏保护】：因为它使用 ^-1 (最多只匹配一次，不是无限循环)，所以允许零宽结尾！
local tail_record  = valid_record * -1

-- 4. 终极自适应主文法树：
--    nl^0吃掉头部空行 -> 循环匹配标准行 -> 匹配可选的无换行尾行 -> 锚定EOF防截断
--    * -1 为核心安全锁：要求状态机必须 100% 无损匹配完整个字符串，
--    若中途存在未闭合的引号或严重语法断裂，立刻报错返回 nil，彻底杜绝隐式截断！
local csv_grammar  = Ct( nl^0 * std_record^0 * tail_record^-1 ) * -1


-- =====================================================================================
-- 第二部分：面向对象 (OO) 元表与无损序列化辅助逻辑
-- =====================================================================================
local csv_mt = {}
csv_mt.__index = csv_mt

--- [内部辅助函数] 序列化单个单元格字段为标准 CSV 规范格式
--- @param val any 需要格式化的字段值
--- @return string 规范化并转义后的 CSV 字段文本
local function escape_csv_field(val)
    local str = tostring(val or "")
    -- 若字段包含逗号、双引号或换行符，必须进行外层双引号包裹，并将内部双引号转义为 ""
    if str:find('[,"\r\n]') then
        return '"' .. str:gsub('"', '""') .. '"'
    end
    return str
end

--- [元方法] 当调用 tostring(csv_obj) 或进行文件写入时，自动将二维 Table 无损还原为标准 CSV 文本
--- @return string 格式化后的标准多行 CSV 字符串
function csv_mt:__tostring()
    local out = {}
    for i = 1, #self do
        local row = self[i]
        if type(row) == "table" then
            local row_out = {}
            for j = 1, #row do
                row_out[j] = escape_csv_field(row[j])
            end
            out[i] = table_concat(row_out, ",")
        end
    end
    -- 按照工业规范，CSV 文件结尾应保持一个标准的换行符
    return table_concat(out, "\n") .. "\n"
end


-- =====================================================================================
-- 第三部分：业务调用与数据操作方法 (OO API)
-- =====================================================================================

--- [核心解析接口] 解析 CSV 格式字符串并高效追加到当前实例对象中
--- 该设计支持流式追加与分片加载，适合处理大型网络数据块
--- @param s string 待解析的 CSV 原始文本
--- @return table|nil csv_obj 成功则返回挂载了方法的自身，失败则返回 nil
--- @return string|nil err_msg 解析失败时的详细语法诊断描述
function csv_mt:parse_string(s)
    if type(s) ~= "string" or s == "" then
        return self
    end

    local parsed_rows = lpeg.match(csv_grammar, s)
    if not parsed_rows then
        return nil, "[LPeg Parse Error] CSV 语法严重断裂：发现未闭合的双引号、错乱转义或文件格式损坏！"
    end

    -- 将 LPeg 生成的新行数据高效合并到当前对象末尾
    local start_idx = #self
    for i, row in ipairs(parsed_rows) do
        self[start_idx + i] = row
    end

    return self
end

--- [内部辅助函数] 无状态迭代器核心驱动
--- 依靠泛型 for 的底层寄存器状态控制，避免每次迭代生成闭包和 upvalue，完全释放 GC 压力
local function read_line_stateless(t, index)
    index = index + 1
    local row = t[index]
    if not row then return nil end
    -- 注意：Lua 的泛型 for 要求第一个返回值为控制变量 (即当前行号 index)
    return index, unpack(row)
end

--- [业务方法] 极速无状态迭代器，专为大数据量泛型 for 循环打造
--- 用法示例: for row_idx, id, name, val in my_csv:lines() do ... end
--- @return function 迭代器驱动函数
--- @return table 实例自身
--- @return number 初始游标
function csv_mt:lines()
    return read_line_stateless, self, 0
end

--- [业务方法] 向当前表格尾部追加一行新数据
--- @param t table 必须是一个数组型的 Lua Table，例如: {"1002", "测试道具", "99"}
function csv_mt:insert(t)
    if type(t) == 'table' then
        table_insert(self, t)
    else
        error(string_format('csv:insert expected table row, got %s', type(t)), 2)
    end
end

--- [业务方法] 对当前表格内的数据行进行排序
--- @param func function|nil 排序比较函数。若留空，默认按照第一列(通常为主键ID)的升序排列
function csv_mt:sort(func)
    func = func or function (a, b) 
        return (a[1] or "") < (b[1] or "") 
    end
    table_sort(self, func)
end

--- [业务方法] 将当前表格内容以严谨的二进制覆盖模式持久化保存到本地硬盘
--- @param filepath string 目标文件的物理保存路径
--- @return boolean success 是否成功写入
--- @return string|nil error_msg 失败时的详细错误描述
function csv_mt:write(filepath)
    -- 强制采用 "wb" 二进制写入模式，杜绝 Windows 平台对 \n 自动转义篡改为 \r\n 的隐式破坏
    local f, err = io_open(filepath, 'wb')
    if f then
        f:write(tostring(self))
        f:close()
        return true
    else
        return false, string_format("[LPeg IO Error] 无法写入文件 '%s': %s", tostring(filepath), tostring(err))
    end
end

--- [业务方法] 格式化输出当前 CSV 结构到控制台，便于开发调试与直观核对
function csv_mt:print()
    for row_idx, row in ipairs(self) do
        if type(row) == "table" then
            print(string_format("Row %d: %s", row_idx, table_concat(row, " | ")))
        else
            print(string_format("Row %d: %s", row_idx, tostring(row)))
        end
    end
end


-- =====================================================================================
-- 第四部分：模块静态工厂接口 (静态 API 导出)
-- =====================================================================================
local csv = {}

--- [工厂方法] 创建一个全新的、空的 CSV 实例对象
--- @return table csv_obj 具有完整面向对象方法的空数据对象
function csv.new()
    return setmetatable({}, csv_mt)
end

--- [静态接口] 一步到位解析多行 CSV 格式原始字符串并返回标准实例
--- @param s string 内存中的 CSV 文本字符串
--- @return table|nil csv_obj 成功返回对象，语法断裂则返回 nil
--- @return string|nil err_msg 失败时的错误原因
function csv.parse_string(s)
    local obj = csv.new()
    return obj:parse_string(s)
end

--- [静态接口] 严谨地从磁盘加载并完整解析指定的本地 CSV 物理文件
--- @param filepath string 本地 CSV 文件的物理路径
--- @return table|nil csv_obj 成功返回对象，失败返回 nil
--- @return string|nil err_msg 失败时的详细原因描述
function csv.open(filepath)
    -- 强制采用 "rb" 二进制安全模式读取，保证文件流原汁原味投喂给 LPeg 状态机
    local f, err = io_open(filepath, "rb")
    if not f then
        return nil, string_format("[LPeg IO Error] 无法读取文件 '%s': %s", tostring(filepath), tostring(err))
    end
    
    -- 一次性高吞吐量将整个文件内容加载到内存中匹配，在处理大表时具有极高的 I/O 效率
    local s = f:read('*a')
    f:close()
    
    return csv.parse_string(s)
end

--- [工具方法] 高效集合运算：快速找出 csv2 中存在但 csv1 中不存在的“新增行数据”
--- 核心采用哈希映射结构，将算法复杂度从传统的双重循环 O(N*M) 降至线性 O(N+M)
--- @param csv1 table 基准对照老表
--- @param csv2 table 待比对的新表
--- @return table diff_table 仅包含新增数据行的全新 CSV 实例
function csv.diff(csv1, csv2)
    local t = csv.new()
    local lookup_set = {}
    
    -- 为原表的第一列(主键键值)建立 O(1) 极速查找哈希索引集
    for _, v1 in ipairs(csv1) do
        if type(v1) == "table" and v1[1] ~= nil then 
            lookup_set[v1[1]] = true 
        end
    end
    
    -- 线性扫描新表，快速筛出未曾记录过的新行并插入结果集
    for _, v2 in ipairs(csv2) do
        if type(v2) == "table" and (v2[1] == nil or not lookup_set[v2[1]]) then
            t:insert(v2)
        end
    end
    
    return t
end

-- 兼容老版本的调用习惯：保留 parse 作为 parse_string 的较短别名
csv.parse = csv.parse_string

return csv

test-csv.lua

--
--         FILE:  test-csv.lua
--       AUTHOR:  LeeoNix
--  DESCRIPTION:
--        NOTES:  ---
--

local csv = require 'csv'

function test()
    local t, err = csv.open(arg[1] or '_list.csv')
    if not t then
        print(err)
    end -- end if
    print('--------------------------------------------------------------------------------')
    for i, name, start, to, title in t:lines() do
        print(table.concat({i, name, start, to, title}, ','))
--        print(i, name, start, to, title)
    end -- end for
    print('--------------------------------------------------------------------------------')
    for i, v in ipairs(t) do
        print(i .. ',' .. table.concat(v, ','))
    end -- end for
    print('--------------------------------------------------------------------------------')
    t:print()
    print('--------------------------------------------------------------------------------')
    print(t)
    print('--------------------------------------------------------------------------------')
    t:write('_test.csv')
end -- end function

test()

test_csv.lua

-- =====================================================================================
-- Filename:    test_csv.lua
-- Description: 工业级 LPeg CSV 解析引擎自动化单元测试套件 (Test Suite)
-- Author:      Gemini
-- Target:      对 csv.lua 进行全方位边界测试、异常拦截与性能压测
-- =====================================================================================

local csv = require("csv")

-- =====================================================================================
-- [测试框架核心] 轻量级单元测试驱动
-- =====================================================================================
local pass_count = 0
local fail_count = 0
local total_duration = 0

local function print_header(title)
    print("\n" .. string.rep("=", 70))
    print(" >>> TEST SUITE: " .. title)
    print(string.rep("=", 70))
end

local function run_test(name, test_func)
    io.write(string.format(" [TEST] %-48s ", name .. " ..."))
    local start_time = os.clock()
    local status, err = pcall(test_func)
    local cost_time = (os.clock() - start_time) * 1000
    total_duration = total_duration + cost_time

    if status then
        pass_count = pass_count + 1
        io.write(string.format("[ PASS ] (%.2f ms)\n", cost_time))
    else
        fail_count = fail_count + 1
        io.write(string.format("[ FAIL ] (%.2f ms)\n", cost_time))
        print("        错误详情: " .. tostring(err))
    end
end

local function assert_eq(actual, expected, msg)
    if actual ~= expected then
        error(string.format("%s (期望值: '%s', 实际值: '%s')", msg or "断言失败", tostring(expected), tostring(actual)), 2)
    end
end

-- =====================================================================================
-- [测试用例 1] 基础常规解析与字段提取
-- =====================================================================================
print_header("1. 基础解析与常规数据校验")

run_test("标准逗号分隔与多行解析", function()
    local raw_str = "ID,Name,Level\n1001,战士,99\n1002,法师,85"
    local t = csv.parse_string(raw_str)
    assert_eq(#t, 3, "总行数应该为 3 行")
    assert_eq(t[1][2], "Name", "表头第2列需为 Name")
    assert_eq(t[2][2], "战士", "第2行第2列需为 战士")
    assert_eq(t[3][3], "85", "第3行第3列需为 85")
end)

-- =====================================================================================
-- [测试用例 2] 极端边界情况：双引号、转义脱壳与特殊字符
-- =====================================================================================
print_header("2. 极端转义与特殊语法边界 (Edge Cases)")

run_test("字段内嵌逗号与双引号脱壳 ( \"\" -> \" )", function()
    -- 模拟：1003,"这是一个,带有逗号和""引号""的字段",100
    local raw_str = '1003,"这是一个,带有逗号和""引号""的字段",100'
    local t = csv.parse_string(raw_str)
    assert_eq(t[1][2], '这是一个,带有逗号和"引号"的字段', "转义脱壳失败，未正确清洗双引号与逗号")
end)

run_test("字段内嵌多行换行符 (配置表多行描述文本)", function()
    local raw_str = '1004,"第一行文本\n第二行文本\r\n第三行文本",VIP'
    local t = csv.parse_string(raw_str)
    assert_eq(#t, 1, "包含内嵌换行符时，不应被拆分为多行记录")
    assert_eq(t[1][2], "第一行文本\n第二行文本\r\n第三行文本", "多行文本字段提取完整性校验失败")
    assert_eq(t[1][3], "VIP", "多行文本后的尾部字段定位错乱")
end)

run_test("连续空逗号与首尾空字段", function()
    local raw_str = ",,,,\n1,,3,,"
    local t = csv.parse_string(raw_str)
    assert_eq(#t[1], 5, "第一行5个空字段解析数量不符")
    assert_eq(t[1][1], "", "首列空字段值不为 empty string")
    assert_eq(t[2][2], "", "中间连续空字段解析失败")
end)

-- =====================================================================================
-- [测试用例 3] 跨平台换行符全兼容测试
-- =====================================================================================
print_header("3. 跨平台换行符兼容性测试")

run_test("混合换行符兼容 (Win \\r\\n + Unix \\n + OldMac \\r)", function()
    local raw_str = "Row1\r\nRow2\nRow3\rRow4"
    local t = csv.parse_string(raw_str)
    assert_eq(#t, 4, "未能正确识别并解析所有4种换行风格的记录")
    assert_eq(t[1][1], "Row1", "Win换行解析失败")
    assert_eq(t[2][1], "Row2", "Unix换行解析失败")
    assert_eq(t[3][1], "Row3", "旧式Mac换行解析失败")
    assert_eq(t[4][1], "Row4", "无换行尾行解析失败")
end)

-- =====================================================================================
-- [测试用例 4] 核心杀招：幽灵空行拦截与防静默截断
-- =====================================================================================
print_header("4. 幽灵空行拦截 & 防静默截断安全锁")

run_test("幽灵空行拦截 (文件尾部带有换行符时绝不生成 {\"\"} 空表)", function()
    local raw_str_with_nl = "A,B,C\n1,2,3\n\n" -- 尾部有连续换行
    local t = csv.parse_string(raw_str_with_nl)
    -- 注意：最后的 \n 不应该在结尾生成一个包含 "" 的新行
    local last_row = t[#t]
    if #last_row == 1 and last_row[1] == "" then
        error("拦截失败！在表尾生成了幽灵空行 {\"\"}")
    end
end)

run_test("防静默截断安全锁 (遭遇未闭合双引号时必须显式报错)", function()
    -- 模拟一个语法断裂的 CSV：第二行的双引号没有闭合
    local broken_str = 'ID,Desc\n101,"这个描述忘记闭合双引号了\n102,正常数据'
    local t, err = csv.parse_string(broken_str)
    assert_eq(t, nil, "静默截断防御失败！在语法严重错误时依然返回了半截残缺数据表")
    if not string.find(err or "", "Parse Error") then
        error("未能抛出预期的 LPeg Parse Error 错误信息")
    end
end)

-- =====================================================================================
-- [测试用例 5] 面向对象 (OO) 接口 & 无状态迭代器
-- =====================================================================================
print_header("5. OO 业务方法与无状态迭代器测试")

run_test("insert() 追加与 sort() 排序机制", function()
    local t = csv.new()
    t:insert({"300", "弓箭手", "C"})
    t:insert({"100", "狂战士", "S"})
    t:insert({"200", "大司祭", "A"})
    
    assert_eq(#t, 3, "insert 插入数量不正确")
    
    -- 默认按第一列 ASCII/主键 升序排序
    t:sort()
    assert_eq(t[1][1], "100", "默认升序排序后，第一行主键应为 100")
    assert_eq(t[3][1], "300", "默认升序排序后，最后一行主键应为 300")
    
    -- 自定义按第三列降序排序
    t:sort(function(a, b) return (a[3] or "") > (b[3] or "") end)
    assert_eq(t[1][1], "100", "按列3降序，评分S的战士应在第一位")
end)

run_test("lines() 无状态零 GC 迭代器解构遍历", function()
    local raw_str = "101,苹果,50\n102,香蕉,30\n103,西瓜,10"
    local t = csv.parse_string(raw_str)
    
    local count = 0
    local names = {}
    for row_idx, id, name, price in t:lines() do
        count = count + 1
        names[row_idx] = name
    end
    
    assert_eq(count, 3, "迭代器遍历的行数总数不匹配")
    assert_eq(names[1], "苹果", "迭代器第1行解构数据异常")
    assert_eq(names[3], "西瓜", "迭代器第3行解构数据异常")
end)

-- =====================================================================================
-- [测试用例 6] Diff 差异比对算法
-- =====================================================================================
print_header("6. 高效集合运算：csv.diff 比对")

run_test("基于 O(1) 哈希表的新增数据行快速过滤", function()
    local old_table = csv.parse_string("ID,Val\n1,A\n2,B\n3,C")
    local new_table = csv.parse_string("ID,Val\n1,A\n2,B_mod\n3,C\n4,D_new\n5,E_new")
    
    local diff_table = csv.diff(old_table, new_table)
    assert_eq(#diff_table, 2, "Diff 运算未准确筛选出 2 条新增行")
    assert_eq(diff_table[1][1], "4", "新增行第1条ID应为 4")
    assert_eq(diff_table[2][2], "E_new", "新增行第2条内容应为 E_new")
end)

-- =====================================================================================
-- [测试用例 7] 二进制安全 I/O 与无损序列化还原
-- =====================================================================================
print_header("7. 文件 I/O 与无损序列化 (Round-Trip Test)")

run_test("write() 与 open() 的二进制无损循环回弹", function()
    local temp_file = "test_temp_output.csv"
    local original_str = 'ID,Name,Memo\n1,"包含,逗号",Normal\n2,"包含""引号""和\n换行",Special\n'
    
    local t1 = csv.parse_string(original_str)
    local write_ok, w_err = t1:write(temp_file)
    assert_eq(write_ok, true, "文件写入硬盘失败: " .. tostring(w_err))
    
    -- 再次从硬盘读取刚才写入的文件
    local t2, r_err = csv.open(temp_file)
    assert_eq(t2 ~= nil, true, "从硬盘读取已写入文件失败: " .. tostring(r_err))
    
    -- 校验转义后的字符串序列化是否完美一致
    assert_eq(tostring(t1), tostring(t2), "无损序列化 Round-Trip 验证失败！写出再读入的数据发生了畸变")
    
    -- 清理临时文件
    os.remove(temp_file)
end)

-- =====================================================================================
-- [测试用例 8] 工业级极限性能压测 (Benchmark)
-- =====================================================================================
print_header("8. 极限吞吐量性能压测 (Benchmark)")

run_test("50,000 行包含转义与特殊字符的大表极速解析", function()
    -- 在内存中快速构建 5 万行数据测试集
    local lines = {"ID,Title,Description,Price"}
    for i = 1, 50000 do
        -- 故意混入带有双引号和逗号的复杂文本
        lines[i + 1] = string.format('%d,"商品_%d","这是描述,包含""转义引号""测试",%.2f', i, i, i * 1.5)
    end
    local big_csv_str = table.concat(lines, "\n")
    
    local start_t = os.clock()
    local big_table, err = csv.parse_string(big_csv_str)
    local cost_ms = (os.clock() - start_t) * 1000
    
    assert_eq(#big_table, 50001, "大表总行数解析遗漏！")
    assert_eq(big_table[50001][1], "50000", "最后一列主键数据提取核对失败")
    
    io.write(string.format("\n        [性能指标] 50,001 行复杂表解析耗时: %.2f ms | 吞吐速率: %.0f 行/秒\n        ", 
             cost_ms, 50000 / (cost_ms / 1000)))
end)

-- =====================================================================================
-- [测试报告总结]
-- =====================================================================================
print("\n" .. string.rep("=", 70))
print(string.format(" >>> 测试总结 (Total Duration: %.2f ms)", total_duration))
print(string.rep("=", 70))
print(string.format(" 成功通过 (PASS) : %d", pass_count))
if fail_count > 0 then
    print(string.format(" 失败挂掉 (FAIL) : %d", fail_count))
    print("\n 警告: 您的 CSV 库存在未通过的测试用例，请重点排查！")
else
    print(string.format(" 失败挂掉 (FAIL) : %d", fail_count))
    print("\n 恭喜您！所有极其严苛的单元测试和压力测试均顺利通过！")
    print("  您的 csv.lua 现已达到高度强壮的工业级水准，可放心投入生产环境！")
end
print(string.rep("=", 70) .. "\n")

test_lpeg.lua

local csv = require "csv"

-- 测试用例：包含多行文本、"" 转义、以及带有结尾换行符的标准 CSV
local raw_text = [[
id,name,desc
1001,LeeoNix,"Hacker ""Programmer"""
1002,LPeg,"This is a multiline
description with a , comma and ""quotes"" inside!"
1003,TotalCommander,No newline at the very end of file]]

local table_data, err = csv.parse(raw_text)

if not table_data then
    print("解析失败: " .. err)
else
    print("=== 解析成功！总行数: " .. #table_data .. " ===")
    for row_idx, row in ipairs(table_data) do
        print(string.format("Row %d [Col Count: %d]: %s | %s | %s", 
              row_idx, #row, row[1], row[2], string.gsub(row[3] or "", "\n", "\\n")))
    end
end