-- Module:PythonScript/Librarys/Csv
-- 本檔案包含了CSV和JSON的編碼與解碼模型。
local CsvMod = {}
local PyClass = require("Module:PythonScript/Class")
local Runtime = require("Module:PythonScript/Runtime")
-- 定義 csv 專屬例外類別
local CsvErrorClass = PyClass.make_class("Error", {Runtime.builtins.Exception}, {})
-- ==========================================
-- CSV 解析核心 (跨行引號狀態機)
-- ==========================================
local function parse_csv_line(iter, delimiter, quotechar)
local parsed_row = {}
local current_val = {}
local quote_open = false
local line_str = ""
while true do
-- 嘗試透過 Python 迭代器拿取下一行
local success, res = pcall(Runtime.builtins.next, {iter})
if not success then
-- 若遇到 StopIteration，代表資料結束
if type(res) == "table" and res.__class and res.__class.__name == "StopIteration" then
if line_str == "" and not quote_open then
Runtime.raise("StopIteration")
else
-- 如果資料已經結束，但引號還沒閉合，就拋出 csv.Error
if quote_open then
Runtime.raise(CsvErrorClass, "unexpected end of data")
end
end
end
-- 其他預期外錯誤則向上拋出
error(res, 0)
end
-- 確保拿到的物件能轉為字串
local success_str, s = pcall(Runtime.builtins.str.__call, {res})
local chunk = success_str and s or tostring(res)
-- 濾除原生的結尾換行，因為 CSV 模組的解析會自行補上跨行換行
chunk = chunk:gsub("\r?\n$", "")
if line_str == "" then
line_str = chunk
else
line_str = line_str .. "\n" .. chunk
end
local i = 1
local n = #line_str
parsed_row = {}
current_val = {}
quote_open = false
while i <= n do
local c = line_str:sub(i, i)
if c == quotechar then
-- 處理連續兩個引號的跳脫字元 ("")
if quote_open and i < n and line_str:sub(i+1, i+1) == quotechar then
table.insert(current_val, quotechar)
i = i + 1
else
quote_open = not quote_open
end
elseif c == delimiter and not quote_open then
table.insert(parsed_row, table.concat(current_val))
current_val = {}
else
table.insert(current_val, c)
end
i = i + 1
end
if not quote_open then
table.insert(parsed_row, table.concat(current_val))
return parsed_row
else
-- 引號依然開啟，代表這是一個跨行欄位，繼續進入下一個迴圈拿取下一行
end
end
end
-- ==========================================
-- reader 類別
-- ==========================================
local reader_dict = {
__init__ = {
__type = "builtin_method",
__call = function(args, kwargs)
local self = args[1]
local csvfile = args[2]
local kw = kwargs or {}
self.__dict.delimiter = kw.delimiter or ","
self.__dict.quotechar = kw.quotechar or '"'
if type(self.__dict.delimiter) ~= "string" or #self.__dict.delimiter ~= 1 then
Runtime.raise("TypeError", '"delimiter" must be a 1-character string')
end
if type(self.__dict.quotechar) ~= "string" or #self.__dict.quotechar ~= 1 then
Runtime.raise("TypeError", '"quotechar" must be a 1-character string')
end
self.__dict.iterator = Runtime.builtins.iter({csvfile})
self.__dict.line_num = 0
return nil
end
},
__iter__ = {
__type = "builtin_method",
__call = function(args) return args[1] end
},
__next__ = {
__type = "builtin_method",
__call = function(args)
local self = args[1]
local row_table = parse_csv_line(self.__dict.iterator, self.__dict.delimiter, self.__dict.quotechar)
self.__dict.line_num = self.__dict.line_num + 1
local elements = {}
for _, v in ipairs(row_table) do table.insert(elements, v) end
return { __type = "list", elements = elements }
end
},
line_num = {
__type = "property",
fget = { __type = "builtin_method", __call = function(args) return args[1].__dict.line_num end }
}
}
local ReaderClass = PyClass.make_class("reader", {}, reader_dict)
-- ==========================================
-- writer 類別 (實作錯誤防護)
-- ==========================================
local writer_dict = {
__init__ = {
__type = "builtin_method",
__call = function(args, kwargs)
-- 初始化不報錯，直到實際 write 才會擋下
return nil
end
},
writerow = {
__type = "builtin_method",
__call = function() Runtime.raise("OSError", "write not supported in this environment") end
},
writerows = {
__type = "builtin_method",
__call = function() Runtime.raise("OSError", "write not supported in this environment") end
}
}
local WriterClass = PyClass.make_class("writer", {}, writer_dict)
-- ==========================================
-- DictReader 類別
-- ==========================================
local DictReader_dict = {
__init__ = {
__type = "builtin_method",
__call = function(args, kwargs)
local self = args[1]; local f = args[2]
local kw = kwargs or {}
self.__dict.fieldnames = (args[3] ~= nil and args[3]) or kw.fieldnames
self.__dict.restkey = (args[4] ~= nil and args[4]) or kw.restkey or Runtime.builtins.None
self.__dict.restval = (args[5] ~= nil and args[5]) or kw.restval or Runtime.builtins.None
self.__dict.reader = PyClass.make_instance(ReaderClass)
-- 將 f 以及其他參數轉傳給原生的 reader 初始化
ReaderClass.__dict.__init__.__call({self.__dict.reader, f}, kwargs)
return nil
end
},
__iter__ = {
__type = "builtin_method",
__call = function(args) return args[1] end
},
__next__ = {
__type = "builtin_method",
__call = function(args)
local self = args[1]
local row_list = ReaderClass.__dict.__next__.__call({self.__dict.reader})
local row = row_list.elements
-- 若未提供 fieldnames，則將第一行作為 fieldnames
if self.__dict.fieldnames == nil or self.__dict.fieldnames == Runtime.builtins.None then
self.__dict.fieldnames = row
row_list = ReaderClass.__dict.__next__.__call({self.__dict.reader})
row = row_list.elements
end
local fieldnames = (type(self.__dict.fieldnames) == "table" and self.__dict.fieldnames.elements) or self.__dict.fieldnames
local d = { __type = "dict", pairs = {} }
for i, field in ipairs(fieldnames) do
local val = row[i]
if val == nil then val = self.__dict.restval end
table.insert(d.pairs, { key = field, value = val })
end
-- 欄位多於標題時，歸入 restkey
if #row > #fieldnames then
local rest = { __type = "list", elements = {} }
for i = #fieldnames + 1, #row do
table.insert(rest.elements, row[i])
end
table.insert(d.pairs, { key = self.__dict.restkey, value = rest })
end
return d
end
},
fieldnames = {
__type = "property",
fget = { __type = "builtin_method", __call = function(args) return args[1].__dict.fieldnames end },
fset = { __type = "builtin_method", __call = function(args) args[1].__dict.fieldnames = args[2]; return nil end }
},
line_num = {
__type = "property",
fget = { __type = "builtin_method", __call = function(args) return args[1].__dict.reader.__dict.line_num end }
}
}
local DictReaderClass = PyClass.make_class("DictReader", {}, DictReader_dict)
-- ==========================================
-- DictWriter 類別 (實作錯誤防護)
-- ==========================================
local DictWriter_dict = {
__init__ = {
__type = "builtin_method",
__call = function(args, kwargs) return nil end
},
writeheader = {
__type = "builtin_method",
__call = function() Runtime.raise("OSError", "write not supported in this environment") end
},
writerow = {
__type = "builtin_method",
__call = function() Runtime.raise("OSError", "write not supported in this environment") end
},
writerows = {
__type = "builtin_method",
__call = function() Runtime.raise("OSError", "write not supported in this environment") end
}
}
local DictWriterClass = PyClass.make_class("DictWriter", {}, DictWriter_dict)
-- ==========================================
-- 模組打包
-- ==========================================
function CsvMod.get_module()
local dict = {
QUOTE_MINIMAL = 0,
QUOTE_ALL = 1,
QUOTE_NONNUMERIC = 2,
QUOTE_NONE = 3,
Error = CsvErrorClass,
reader = ReaderClass,
writer = WriterClass,
DictReader = DictReaderClass,
DictWriter = DictWriterClass
}
local inst = PyClass.make_instance(PyClass.make_class("module", {}, dict))
inst.__dict["__name__"] = "csv"
return inst
end
-- ==========================================
-- 整合Json解碼於此
-- ==========================================
local JsonMod = {}
local initialized = false
local cache = {}
local function init_json()
local Evaluator = require("Module:PythonScript/Evaluator")
local EvalUtils = require("Module:PythonScript/Evaluator/Utils")
local RUtils = require("Module:PythonScript/Runtime/Utils")
local function is_none(v)
return v == nil or (type(v) == "table" and v.__class and v.__class.__name == "NoneType")
end
-- ==========================================
-- 1. 字串跳脫與單次遍歷 JSON 編碼核心 (Encoder)
-- ==========================================
local escape_map = {
['"'] = '\\"',
['\\'] = '\\\\',
['\b'] = '\\b',
['\f'] = '\\f',
['\n'] = '\\n',
['\r'] = '\\r',
['\t'] = '\\t'
}
local function encode_json_string(s, ensure_ascii)
local out = { '"' }
if not ensure_ascii then
local escaped = s:gsub('[%z\1-\31"\\]', function(c)
return escape_map[c] or string.format("\\u%04x", string.byte(c))
end)
out[#out + 1] = escaped
else
for cp in mw.ustring.gcodepoint(s) do
if cp == 34 then out[#out + 1] = '\\"'
elseif cp == 92 then out[#out + 1] = '\\\\'
elseif cp == 8 then out[#out + 1] = '\\b'
elseif cp == 12 then out[#out + 1] = '\\f'
elseif cp == 10 then out[#out + 1] = '\\n'
elseif cp == 13 then out[#out + 1] = '\\r'
elseif cp == 9 then out[#out + 1] = '\\t'
elseif cp < 32 or (cp >= 127 and cp <= 0xFFFF) then
out[#out + 1] = string.format("\\u%04x", cp)
elseif cp > 0xFFFF then
local sub = cp - 0x10000
local high = 0xD800 + math.floor(sub / 1024)
local low = 0xDC00 + (sub % 1024)
out[#out + 1] = string.format("\\u%04x\\u%04x", high, low)
else
out[#out + 1] = string.char(cp)
end
end
end
out[#out + 1] = '"'
return table.concat(out)
end
-- 格式化字典鍵 (JSON 規範要求 key 必須轉為字串)
local function coerce_dict_key(k, skipkeys, ensure_ascii)
if type(k) == "string" then return k end
if type(k) == "boolean" then return k and "true" or "false" end
if is_none(k) then return "null" end
if type(k) == "number" then
if k ~= k then return "NaN"
elseif k == math.huge then return "Infinity"
elseif k == -math.huge then return "-Infinity" end
return tostring(k)
end
if type(k) == "table" then
if k.__type == "int" and k.is_big then
local MathCore = require("Module:PythonScript/Runtime/Cmath")
return MathCore.BigInt.to_string(k.sign, k.limbs)
elseif k.__type == "instance" and k.__builtin_value ~= nil then
return coerce_dict_key(k.__builtin_value, skipkeys, ensure_ascii)
end
end
if skipkeys then return nil end
Runtime.raise("TypeError", "keys must be str, int, float, bool or None, not " .. RUtils.get_type_str(k))
end
local function serialize_value(val, opts, depth, seen, buf)
-- 處理自定義實例但底層綁定原生型別 (如 IntEnum / StrEnum / NamedTuple)
if type(val) == "table" and val.__type == "instance" and val.__builtin_value ~= nil then
local bv = val.__builtin_value
if type(bv) == "number" or type(bv) == "string" or type(bv) == "boolean" or (type(bv) == "table" and (bv.__type == "tuple" or bv.__type == "list" or bv.__type == "dict" or bv.__type == "int")) then
val = bv
end
end
if is_none(val) then
buf[#buf + 1] = "null"
return
elseif type(val) == "boolean" then
buf[#buf + 1] = val and "true" or "false"
return
elseif type(val) == "number" then
if val ~= val then
if not opts.allow_nan then Runtime.raise("ValueError", "Out of range float values are not JSON compliant") end
buf[#buf + 1] = "NaN"
elseif val == math.huge then
if not opts.allow_nan then Runtime.raise("ValueError", "Out of range float values are not JSON compliant") end
buf[#buf + 1] = "Infinity"
elseif val == -math.huge then
if not opts.allow_nan then Runtime.raise("ValueError", "Out of range float values are not JSON compliant") end
buf[#buf + 1] = "-Infinity"
else
buf[#buf + 1] = tostring(val)
end
return
elseif type(val) == "string" then
buf[#buf + 1] = encode_json_string(val, opts.ensure_ascii)
return
elseif type(val) == "table" then
if val.__type == "int" and val.is_big then
local MathCore = require("Module:PythonScript/Runtime/Cmath")
buf[#buf + 1] = MathCore.BigInt.to_string(val.sign, val.limbs)
return
elseif val.__type == "list" or val.__type == "tuple" then
local elts = val.elements
if #elts == 0 then
buf[#buf + 1] = "[]"
return
end
if opts.check_circular then
if seen[val] then Runtime.raise("ValueError", "Circular reference detected") end
seen[val] = true
end
buf[#buf + 1] = "["
local next_depth = depth + 1
local item_sep = opts.item_separator
local indent_str = opts.indent_str
local step_indent = indent_str and ("\n" .. string.rep(indent_str, next_depth)) or nil
for i = 1, #elts do
if i > 1 then buf[#buf + 1] = item_sep end
if step_indent then buf[#buf + 1] = step_indent end
serialize_value(elts[i], opts, next_depth, seen, buf)
end
if indent_str then
buf[#buf + 1] = "\n" .. string.rep(indent_str, depth)
end
buf[#buf + 1] = "]"
if opts.check_circular then seen[val] = nil end
return
elseif val.__type == "dict" then
local pairs_arr = val.pairs
if #pairs_arr == 0 then
buf[#buf + 1] = "{}"
return
end
if opts.check_circular then
if seen[val] then Runtime.raise("ValueError", "Circular reference detected") end
seen[val] = true
end
local valid_items = {}
for i = 1, #pairs_arr do
local p = pairs_arr[i]
local k_str = coerce_dict_key(p.key, opts.skipkeys, opts.ensure_ascii)
if k_str ~= nil then
valid_items[#valid_items + 1] = { key = k_str, value = p.value }
end
end
if #valid_items == 0 then
if opts.check_circular then seen[val] = nil end
buf[#buf + 1] = "{}"
return
end
if opts.sort_keys then
table.sort(valid_items, function(a, b) return a.key < b.key end)
end
buf[#buf + 1] = "{"
local next_depth = depth + 1
local item_sep = opts.item_separator
local kv_sep = opts.key_separator
local indent_str = opts.indent_str
local step_indent = indent_str and ("\n" .. string.rep(indent_str, next_depth)) or nil
for i = 1, #valid_items do
if i > 1 then buf[#buf + 1] = item_sep end
if step_indent then buf[#buf + 1] = step_indent end
buf[#buf + 1] = encode_json_string(valid_items[i].key, opts.ensure_ascii)
buf[#buf + 1] = kv_sep
serialize_value(valid_items[i].value, opts, next_depth, seen, buf)
end
if indent_str then
buf[#buf + 1] = "\n" .. string.rep(indent_str, depth)
end
buf[#buf + 1] = "}"
if opts.check_circular then seen[val] = nil end
return
end
end
-- 遇到預設不支援的型別：委派給 default 回呼或 JSONEncoder.default 方法
if opts.check_circular and type(val) == "table" then
if seen[val] then Runtime.raise("ValueError", "Circular reference detected") end
seen[val] = true
end
local replacement
if opts.encoder_self then
local def_m = PyClass.get_attr(opts.encoder_self, "default")
replacement = Evaluator.invoke(def_m, { val }, {})
elseif opts.default_fn and not is_none(opts.default_fn) then
replacement = Evaluator.invoke(opts.default_fn, { val }, {})
else
Runtime.raise("TypeError", "Object of type " .. RUtils.get_type_str(val) .. " is not JSON serializable")
end
serialize_value(replacement, opts, depth, seen, buf)
if opts.check_circular and type(val) == "table" then
seen[val] = nil
end
end
local function build_encode_opts(kw, encoder_self)
kw = kw or {}
local d = encoder_self and encoder_self.__dict or {}
local function pick_bool(key, def_val)
if kw[key] ~= nil then return EvalUtils.is_truthy(kw[key]) end
if d[key] ~= nil then return EvalUtils.is_truthy(d[key]) end
return def_val
end
local indent_raw = (kw.indent ~= nil) and kw.indent or d.indent
local indent_str = nil
if not is_none(indent_raw) then
if type(indent_raw) == "number" then
indent_str = string.rep(" ", math.max(0, math.floor(indent_raw)))
elseif type(indent_raw) == "string" then
indent_str = indent_raw
end
end
local sep_raw = (kw.separators ~= nil) and kw.separators or d.separators
local item_sep = indent_str and "," or ", "
local key_sep = ": "
if not is_none(sep_raw) and type(sep_raw) == "table" and sep_raw.elements and #sep_raw.elements >= 2 then
item_sep = tostring(sep_raw.elements[1])
key_sep = tostring(sep_raw.elements[2])
end
local default_fn = (kw.default ~= nil) and kw.default or d._default_fn
return {
skipkeys = pick_bool("skipkeys", false),
ensure_ascii = pick_bool("ensure_ascii", true),
check_circular = pick_bool("check_circular", true),
allow_nan = pick_bool("allow_nan", true),
sort_keys = pick_bool("sort_keys", false),
indent_str = indent_str,
item_separator = item_sep,
key_separator = key_sep,
default_fn = default_fn,
encoder_self = encoder_self
}
end
-- ==========================================
-- 2. 單次遍歷高速游標 JSON 解碼核心 (Decoder)
-- ==========================================
local function raise_decode_error(msg, doc, pos)
local B = Runtime.builtins
local sub_doc = doc:sub(1, math.max(0, pos - 1))
local lineno = 1
for _ in sub_doc:gmatch("\n") do lineno = lineno + 1 end
local last_nl = sub_doc:match(".*()\n")
local colno = last_nl and (pos - last_nl) or pos
local full_msg = string.format("%s: line %d column %d (char %d)", msg, lineno, colno, pos - 1)
local exc = PyClass.make_instance(cache.JSONDecodeError)
exc.__dict.msg = full_msg
exc.__dict.doc = doc
exc.__dict.pos = pos - 1
exc.__dict.lineno = lineno
exc.__dict.colno = colno
exc.__dict.args = { __type = "tuple", elements = { full_msg, doc, pos - 1 } }
Runtime.raise(exc)
end
local function skip_ws(s, pos, len)
local _, e = s:find("^[ \t\r\n]+", pos)
return e and (e + 1) or pos
end
local function parse_json_string(s, pos, len, strict)
-- pos 指向開頭的 '"'
local i = pos + 1
local out = {}
local chunk_start = i
while i <= len do
local b = string.byte(s, i)
if b == 34 then -- '"'
if i > chunk_start then out[#out + 1] = s:sub(chunk_start, i - 1) end
return table.concat(out), i + 1
elseif b == 92 then -- '\'
if i > chunk_start then out[#out + 1] = s:sub(chunk_start, i - 1) end
i = i + 1
if i > len then raise_decode_error("Unterminated string starting at", s, pos) end
local esc = string.sub(s, i, i)
if esc == '"' or esc == '\\' or esc == '/' then out[#out + 1] = esc; i = i + 1
elseif esc == 'b' then out[#out + 1] = '\b'; i = i + 1
elseif esc == 'f' then out[#out + 1] = '\f'; i = i + 1
elseif esc == 'n' then out[#out + 1] = '\n'; i = i + 1
elseif esc == 'r' then out[#out + 1] = '\r'; i = i + 1
elseif esc == 't' then out[#out + 1] = '\t'; i = i + 1
elseif esc == 'u' then
local hex = s:sub(i + 1, i + 4)
if #hex < 4 or not hex:match("^%x%x%x%x$") then
raise_decode_error("Invalid \\uXXXX escape", s, i - 1)
end
local cp = tonumber(hex, 16)
i = i + 5
-- 處理 UTF-16 Surrogate Pair (\uD800..\uDBFF 搭配 \uDC00..\uDFFF)
if cp >= 0xD800 and cp <= 0xDBFF and s:sub(i, i + 1) == "\\u" then
local hex2 = s:sub(i + 2, i + 5)
if #hex2 == 4 and hex2:match("^%x%x%x%x$") then
local cp2 = tonumber(hex2, 16)
if cp2 >= 0xDC00 and cp2 <= 0xDFFF then
cp = 0x10000 + (cp - 0xD800) * 1024 + (cp2 - 0xDC00)
i = i + 6
end
end
end
out[#out + 1] = mw.ustring.char(cp)
else
raise_decode_error("Invalid \\escape", s, i - 1)
end
chunk_start = i
elseif strict and b < 32 then
raise_decode_error("Invalid control character at", s, i)
else
i = i + 1
end
end
raise_decode_error("Unterminated string starting at", s, pos)
end
local parse_json_value
local function parse_json_array(s, pos, len, hooks)
local i = skip_ws(s, pos + 1, len)
local elts = {}
if i <= len and string.byte(s, i) == 93 then -- ']'
return { __type = "list", elements = elts }, i + 1
end
while true do
if i > len then raise_decode_error("Expecting value", s, i) end
local val
val, i = parse_json_value(s, i, len, hooks)
elts[#elts + 1] = val
i = skip_ws(s, i, len)
local b = string.byte(s, i)
if b == 93 then -- ']'
return { __type = "list", elements = elts }, i + 1
elseif b == 44 then -- ','
i = skip_ws(s, i + 1, len)
else
raise_decode_error("Expecting ',' delimiter", s, i)
end
end
end
local function parse_json_object(s, pos, len, hooks)
local i = skip_ws(s, pos + 1, len)
local kv_pairs = {}
if i <= len and string.byte(s, i) == 125 then -- '}'
local empty_dict = { __type = "dict", pairs = kv_pairs }
if hooks.object_pairs_hook then
return Evaluator.invoke(hooks.object_pairs_hook, { { __type = "list", elements = {} } }, {}), i + 1
elseif hooks.object_hook then
return Evaluator.invoke(hooks.object_hook, { empty_dict }, {}), i + 1
end
return empty_dict, i + 1
end
while true do
if i > len or string.byte(s, i) ~= 34 then -- '"'
raise_decode_error("Expecting property name enclosed in double quotes", s, i)
end
local key
key, i = parse_json_string(s, i, len, hooks.strict)
i = skip_ws(s, i, len)
if i > len or string.byte(s, i) ~= 58 then -- ':'
raise_decode_error("Expecting ':' delimiter", s, i)
end
i = skip_ws(s, i + 1, len)
if i > len then raise_decode_error("Expecting value", s, i) end
local val
val, i = parse_json_value(s, i, len, hooks)
kv_pairs[#kv_pairs + 1] = { key = key, value = val }
i = skip_ws(s, i, len)
local b = string.byte(s, i)
if b == 125 then -- '}'
i = i + 1
if hooks.object_pairs_hook then
local tup_list = {}
for idx = 1, #kv_pairs do
local p = kv_pairs[idx]
tup_list[idx] = { __type = "tuple", elements = { p.key, p.value } }
end
return Evaluator.invoke(hooks.object_pairs_hook, { { __type = "list", elements = tup_list } }, {}), i
end
-- 若有重複 key，後出現的覆蓋先出現的 (維持 CPython dict 行為)
local dedup_pairs = {}
local key_idx = {}
for idx = 1, #kv_pairs do
local p = kv_pairs[idx]
local existing = key_idx[p.key]
if existing then
dedup_pairs[existing].value = p.value
else
local new_i = #dedup_pairs + 1
key_idx[p.key] = new_i
dedup_pairs[new_i] = p
end
end
local dict_obj = { __type = "dict", pairs = dedup_pairs }
if hooks.object_hook then
return Evaluator.invoke(hooks.object_hook, { dict_obj }, {}), i
end
return dict_obj, i
elseif b == 44 then -- ','
i = skip_ws(s, i + 1, len)
else
raise_decode_error("Expecting ',' delimiter", s, i)
end
end
end
parse_json_value = function(s, pos, len, hooks)
pos = skip_ws(s, pos, len)
if pos > len then raise_decode_error("Expecting value", s, pos) end
local b = string.byte(s, pos)
if b == 34 then -- '"'
return parse_json_string(s, pos, len, hooks.strict)
elseif b == 123 then -- '{'
return parse_json_object(s, pos, len, hooks)
elseif b == 91 then -- '['
return parse_json_array(s, pos, len, hooks)
elseif b == 116 and s:sub(pos, pos + 3) == "true" then
return true, pos + 4
elseif b == 102 and s:sub(pos, pos + 4) == "false" then
return false, pos + 5
elseif b == 110 and s:sub(pos, pos + 3) == "null" then
return Runtime.builtins.None, pos + 4
elseif b == 78 and s:sub(pos, pos + 2) == "NaN" then
if hooks.parse_constant then
return Evaluator.invoke(hooks.parse_constant, { "NaN" }, {}), pos + 3
end
return 0/0, pos + 3
elseif b == 73 and s:sub(pos, pos + 7) == "Infinity" then
if hooks.parse_constant then
return Evaluator.invoke(hooks.parse_constant, { "Infinity" }, {}), pos + 8
end
return math.huge, pos + 8
elseif s:sub(pos, pos + 8) == "-Infinity" then
if hooks.parse_constant then
return Evaluator.invoke(hooks.parse_constant, { "-Infinity" }, {}), pos + 9
end
return -math.huge, pos + 9
else
-- 匹配 JSON 數字字面量
local _, num_end, num_str = s:find("^(-?%d+%.?%d*[eE]?[%+%-]?%d*)", pos)
if num_str and num_str ~= "" then
-- 檢查前導零 (如 01 非法，但 0 或 0.1 合法)
local int_part = num_str:match("^-?(%d+)")
if int_part and #int_part > 1 and int_part:sub(1, 1) == "0" then
raise_decode_error("Expecting value", s, pos)
end
local is_float_token = num_str:find("[%.eE]") ~= nil
if is_float_token then
if hooks.parse_float then
return Evaluator.invoke(hooks.parse_float, { num_str }, {}), num_end + 1
end
local n = tonumber(num_str)
if not n then raise_decode_error("Invalid number", s, pos) end
return n, num_end + 1
else
if hooks.parse_int then
return Evaluator.invoke(hooks.parse_int, { num_str }, {}), num_end + 1
end
return Runtime.builtins.int.__call({ num_str }), num_end + 1
end
end
raise_decode_error("Expecting value", s, pos)
end
end
local function build_decode_hooks(kw, decoder_self)
kw = kw or {}
local d = decoder_self and decoder_self.__dict or {}
local function pick_hook(name)
local h = (kw[name] ~= nil) and kw[name] or d[name]
if is_none(h) then return nil end
return h
end
local strict = true
if kw.strict ~= nil then strict = EvalUtils.is_truthy(kw.strict)
elseif d.strict ~= nil then strict = EvalUtils.is_truthy(d.strict) end
return {
object_hook = pick_hook("object_hook"),
object_pairs_hook = pick_hook("object_pairs_hook"),
parse_float = pick_hook("parse_float"),
parse_int = pick_hook("parse_int"),
parse_constant = pick_hook("parse_constant"),
strict = strict
}
end
local function extract_json_input(s)
if type(s) == "string" then
-- 剝除 UTF-8 BOM (若有)
if s:sub(1, 3) == "\239\187\191" then s = s:sub(4) end
return s
elseif type(s) == "table" then
if s.__type == "bytes" or s.__type == "bytearray" then
return s._data
elseif s.__type == "memoryview" then
if s._released then Runtime.raise("ValueError", "operation forbidden on released memoryview object") end
return s._data
end
end
Runtime.raise("TypeError", "the JSON object must be str, bytes or bytearray, not " .. RUtils.get_type_str(s))
end
-- ==========================================
-- 3. 初始化 JSONEncoder / JSONDecoder / 頂層函式
-- ==========================================
local function ensure_init()
if initialized then return end
initialized = true
local B = Runtime.builtins
-- [A] JSONDecodeError 例外類別
local jde_dict = {
__init__ = { __type = "builtin_method", __call = function(a)
local self, msg, doc, pos = a[1], tostring(a[2] or ""), tostring(a[3] or ""), tonumber(a[4]) or 0
local sub_doc = doc:sub(1, pos)
local lineno = 1
for _ in sub_doc:gmatch("\n") do lineno = lineno + 1 end
local last_nl = sub_doc:match(".*()\n")
local colno = last_nl and (pos - last_nl + 1) or (pos + 1)
local full_msg = string.format("%s: line %d column %d (char %d)", msg, lineno, colno, pos)
self.__dict.msg = msg
self.__dict.doc = doc
self.__dict.pos = pos
self.__dict.lineno = lineno
self.__dict.colno = colno
self.__dict.args = { __type = "tuple", elements = { full_msg } }
return nil
end },
__str__ = { __type = "builtin_method", __call = function(a)
local d = a[1].__dict
if d.args and d.args.elements and d.args.elements[1] then return tostring(d.args.elements[1]) end
return tostring(d.msg or "")
end }
}
cache.JSONDecodeError = PyClass.make_class("JSONDecodeError", { B.ValueError }, jde_dict)
cache.JSONDecodeError.__dict["__module__"] = "json.decoder"
-- [B] JSONEncoder 類別 (支援子類別繼承與 super().default(obj))
local encoder_dict = {
item_separator = ", ",
key_separator = ": ",
__init__ = { __type = "builtin_method", __call = function(a, kw)
local self = a[1]
kw = kw or {}
local d = self.__dict
d.skipkeys = (a[2] ~= nil and EvalUtils.is_truthy(a[2])) or (kw.skipkeys ~= nil and EvalUtils.is_truthy(kw.skipkeys)) or false
d.ensure_ascii = (a[3] ~= nil and EvalUtils.is_truthy(a[3])) or (kw.ensure_ascii ~= nil and EvalUtils.is_truthy(kw.ensure_ascii)) or (a[3] == nil and kw.ensure_ascii == nil)
d.check_circular = (a[4] ~= nil and EvalUtils.is_truthy(a[4])) or (kw.check_circular ~= nil and EvalUtils.is_truthy(kw.check_circular)) or (a[4] == nil and kw.check_circular == nil)
d.allow_nan = (a[5] ~= nil and EvalUtils.is_truthy(a[5])) or (kw.allow_nan ~= nil and EvalUtils.is_truthy(kw.allow_nan)) or (a[5] == nil and kw.allow_nan == nil)
d.sort_keys = (a[6] ~= nil and EvalUtils.is_truthy(a[6])) or (kw.sort_keys ~= nil and EvalUtils.is_truthy(kw.sort_keys)) or false
d.indent = (a[7] ~= nil and a[7]) or kw.indent or B.None
d.separators = (a[8] ~= nil and a[8]) or kw.separators or B.None
local def_fn = (a[9] ~= nil and a[9]) or kw.default
if not is_none(def_fn) then d._default_fn = def_fn end
return nil
end },
default = { __type = "builtin_method", __name = "default", __call = function(a)
local self, o = a[1], a[2]
if self.__dict and self.__dict._default_fn and not is_none(self.__dict._default_fn) then
return Evaluator.invoke(self.__dict._default_fn, { o }, {})
end
Runtime.raise("TypeError", "Object of type " .. RUtils.get_type_str(o) .. " is not JSON serializable")
end },
encode = { __type = "builtin_method", __name = "encode", __call = function(a)
local self, o = a[1], a[2]
local opts = build_encode_opts(nil, self)
local buf = {}
serialize_value(o, opts, 0, {}, buf)
return table.concat(buf)
end },
iterencode = { __type = "builtin_method", __name = "iterencode", __call = function(a)
local self, o = a[1], a[2]
local enc_m = PyClass.get_attr(self, "encode")
local s = Evaluator.invoke(enc_m, { o }, {})
return B.iter({ { __type = "list", elements = { s } } })
end }
}
cache.JSONEncoder = PyClass.make_class("JSONEncoder", {}, encoder_dict)
cache.JSONEncoder.__dict["__module__"] = "json.encoder"
-- [C] JSONDecoder 類別
local decoder_dict = {
__init__ = { __type = "builtin_method", __call = function(a, kw)
local self = a[1]
kw = kw or {}
local d = self.__dict
d.object_hook = (a[2] ~= nil and a[2]) or kw.object_hook or B.None
d.parse_float = (a[3] ~= nil and a[3]) or kw.parse_float or B.None
d.parse_int = (a[4] ~= nil and a[4]) or kw.parse_int or B.None
d.parse_constant = (a[5] ~= nil and a[5]) or kw.parse_constant or B.None
d.strict = (a[6] ~= nil and EvalUtils.is_truthy(a[6])) or (kw.strict ~= nil and EvalUtils.is_truthy(kw.strict)) or (a[6] == nil and kw.strict == nil)
d.object_pairs_hook = (a[7] ~= nil and a[7]) or kw.object_pairs_hook or B.None
return nil
end },
raw_decode = { __type = "builtin_method", __name = "raw_decode", __call = function(a, kw)
local self = a[1]
local s = extract_json_input(a[2] or (kw and kw.s))
local idx = tonumber(a[3] or (kw and kw.idx)) or 0
if idx < 0 then raise_decode_error("Expecting value", s, 1) end
local hooks = build_decode_hooks(nil, self)
local obj, end_pos = parse_json_value(s, idx + 1, #s, hooks)
return { __type = "tuple", elements = { obj, end_pos - 1 } }
end },
decode = { __type = "builtin_method", __name = "decode", __call = function(a, kw)
local self = a[1]
local s = extract_json_input(a[2] or (kw and kw.s))
local hooks = build_decode_hooks(nil, self)
local len = #s
local pos = skip_ws(s, 1, len)
local obj, next_pos = parse_json_value(s, pos, len, hooks)
next_pos = skip_ws(s, next_pos, len)
if next_pos <= len then
raise_decode_error("Extra data", s, next_pos)
end
return obj
end }
}
cache.JSONDecoder = PyClass.make_class("JSONDecoder", {}, decoder_dict)
cache.JSONDecoder.__dict["__module__"] = "json.decoder"
-- [D] dumps / dump / loads / load 頂層函式
cache.dumps = {
__type = "builtin_method", __name = "dumps",
__call = function(a, kw)
local obj = a[1]
kw = kw or {}
local cls = kw.cls
if not is_none(cls) then
local init_kw = {}
for k, v in pairs(kw) do
if k ~= "cls" then init_kw[k] = v end
end
local encoder_inst = Evaluator.invoke(cls, {}, init_kw)
local encode_m = PyClass.get_attr(encoder_inst, "encode")
return Evaluator.invoke(encode_m, { obj }, {})
end
local opts = build_encode_opts(kw, nil)
local buf = {}
serialize_value(obj, opts, 0, {}, buf)
return table.concat(buf)
end
}
cache.dump = {
__type = "builtin_method", __name = "dump",
__call = function(a, kw)
local obj, fp = a[1], a[2]
if not fp then Runtime.raise("TypeError", "dump() missing required argument: 'fp'") end
local s = cache.dumps.__call({ obj }, kw)
local write_m = PyClass.get_attr(fp, "write")
Evaluator.invoke(write_m, { s }, {})
return B.None
end
}
cache.loads = {
__type = "builtin_method", __name = "loads",
__call = function(a, kw)
local raw_s = a[1] or (kw and kw.s)
local s = extract_json_input(raw_s)
kw = kw or {}
local cls = kw.cls
if not is_none(cls) then
local init_kw = {}
for k, v in pairs(kw) do
if k ~= "cls" then init_kw[k] = v end
end
local decoder_inst = Evaluator.invoke(cls, {}, init_kw)
local decode_m = PyClass.get_attr(decoder_inst, "decode")
return Evaluator.invoke(decode_m, { s }, {})
end
local hooks = build_decode_hooks(kw, nil)
local len = #s
local pos = skip_ws(s, 1, len)
local obj, next_pos = parse_json_value(s, pos, len, hooks)
next_pos = skip_ws(s, next_pos, len)
if next_pos <= len then
raise_decode_error("Extra data", s, next_pos)
end
return obj
end
}
cache.load = {
__type = "builtin_method", __name = "load",
__call = function(a, kw)
local fp = a[1]
if not fp then Runtime.raise("TypeError", "load() missing required argument: 'fp'") end
local read_m = PyClass.get_attr(fp, "read")
local content = Evaluator.invoke(read_m, {}, {})
return cache.loads.__call({ content }, kw)
end
}
end
function JsonMod.get_module()
ensure_init()
local mod_dict = {
__getattr__ = {
__type = "builtin_method",
__call = function(args)
local self, attr = args[1], args[2]
local val = cache[attr]
if val ~= nil then
self.__dict[attr] = val
return val
end
Runtime.raise("AttributeError", "module 'json' has no attribute '" .. tostring(attr) .. "'")
end
}
}
local mod = PyClass.make_instance(PyClass.make_class("module", {}, mod_dict))
mod.__dict["__name__"] = "json"
mod.__dict["__all__"] = {
__type = "list",
elements = { "dump", "dumps", "load", "loads", "JSONDecoder", "JSONDecodeError", "JSONEncoder" }
}
return mod
end
return JsonMod
end
return {
Csv = CsvMod,
Json = init_json
}
