-- Module:Graph/Wikidata
-- 專為維基百科圖表設計之輕量型 Wikidata SPARQL 直譯與資料萃取引擎
local p = {}
local Expr = require('Module:Graph/Expression')
local sub, find, match, gmatch, gsub, lower, upper = string.sub, string.find, string.match, string.gmatch, string.gsub, string.lower, string.upper
local tonumber, tostring, ipairs, pairs, type = tonumber, tostring, ipairs, pairs, type
-- 單一圖表最大實體查詢上限 (保護 Scribunto 500 次昂貴函式限制與 50MB 記憶體上限)
local MAX_ENTITIES = 100
-- 解碼 wikidatasparql:// URL 並剝離註解
local function decodeSparqlUrl(rawUrl)
local q = match(tostring(rawUrl or ""), "^wikidatasparql:///?%?query=(.+)$")
or match(tostring(rawUrl or ""), "^wikidatasparql:///?(.+)$")
or tostring(rawUrl or "")
-- 自動判斷 PATH 編碼 (%20) 或 QUERY 編碼 (+)
if find(q, "%%20") or not find(q, "%+") then
q = mw.uri.decode(q, "PATH")
else
q = mw.uri.decode(q, "QUERY")
end
q = mw.text.decode(q)
-- 剝離單行 # 註解 (避開引號與 <...> IRI 內的 #)
local lines = mw.text.split(q, "\n", true)
local cleanLines = {}
for i = 1, #lines do
local line = lines[i]
local inQuote, inIri = false, false
local cutPos = nil
for c = 1, #line do
local ch = sub(line, c, c)
if ch == '"' or ch == "'" then
inQuote = not inQuote
elseif not inQuote then
if ch == "<" then inIri = true
elseif ch == ">" then inIri = false
elseif ch == "#" and not inIri then
cutPos = c
break
end
end
end
cleanLines[#cleanLines + 1] = cutPos and sub(line, 1, cutPos - 1) or line
end
return table.concat(cleanLines, "\n")
end
-- 解析 Wikibase Snak 結構為純量值 (數值、ISO 日期字串、Q-ID 或文字)
local function parseSnakValue(snak)
if type(snak) ~= "table" or snak.snaktype ~= "value" then return nil end
local dv = snak.datavalue
if type(dv) ~= "table" then return nil end
local dvType = dv.type
local val = dv.value
if dvType == "quantity" and type(val) == "table" then
local amt = gsub(tostring(val.amount or ""), "^%+", "")
return tonumber(amt)
elseif dvType == "time" and type(val) == "table" then
local tStr = tostring(val.time or "")
-- 處理如 "+1985-00-00T00:00:00Z" 或 "+2016-10-05T00:00:00Z"
local sign, y, m, d = match(tStr, "^([%+%-]?)(%d+)%-(%d%d)%-(%d%d)")
if y then
local numY = tonumber((sign == "-" and "-" or "") .. y)
local cleanM = (m == "00") and "01" or m
local cleanD = (d == "00") and "01" or d
if sign == "-" then
return numY
end
return string.format("%04d-%s-%s", numY, cleanM, cleanD)
end
return tStr
elseif dvType == "wikibase-entityid" and type(val) == "table" then
return val.id
elseif dvType == "monolingualtext" and type(val) == "table" then
return val.text
elseif type(val) == "string" or type(val) == "number" then
return val
end
return nil
end
-- 依語言清單取得 Q-ID 標籤 (含本地快取)
local function getEntityLabel(qid, langList, cache)
if not qid or not mw.wikibase then return qid or "" end
if cache[qid] then return cache[qid] end
local lbl = nil
if mw.wikibase.getLabelByLang and type(langList) == "table" then
for _, lang in ipairs(langList) do
lbl = mw.wikibase.getLabelByLang(qid, lang)
if lbl and lbl ~= "" then break end
end
end
if (not lbl or lbl == "") and mw.wikibase.getLabel then
lbl = mw.wikibase.getLabel(qid)
end
lbl = (lbl and lbl ~= "") and lbl or qid
cache[qid] = lbl
return lbl
end
-- 將 SPARQL 運算式轉譯為 Module:Graph/Expression 可執行的語法
local function translateSparqlExpr(exprStr)
local s = mw.text.trim(tostring(exprStr or ""))
-- 移除外層多餘括號
if match(s, "^%b()$") then
s = sub(s, 2, -2)
end
-- 移除 ^^xsd:dateTime 等型別標記
s = gsub(s, "%^%^<?[%w_:%./#%-]+>?", "")
-- 轉譯 YEAR(?x), MONTH(?x), DAY(?x), LCASE(?x), UCASE(?x)
s = gsub(s, "[yY][eE][aA][rR]%s*%(", "year(")
s = gsub(s, "[mM][oO][nN][tT][hH]%s*%(", "month(")
s = gsub(s, "[dD][aA][yY]%s*%(", "date(")
s = gsub(s, "[lL][cC][aA][sS][eE]%s*%(", "lower(")
s = gsub(s, "[uU][cC][aA][sS][eE]%s*%(", "upper(")
-- 將 ?var 轉為 datum.var
s = gsub(s, "%?([%w_]+)", "datum.%1")
-- 將單一 '=' 比較運算子轉為 '==' (避開 <=, >=, !=, ==)
s = gsub(s, "([^<>!=])=([^=])", "%1==%2")
return s
end
-- 主查詢入口：傳入 wikidatasparql:// URL 或 SPARQL 字串，回傳 (records, shouldAbort)
function p.query(rawUrl)
if not mw.wikibase then return nil, true end
local sparql = decodeSparqlUrl(rawUrl)
if sparql == "" then return nil, true end
-- 1. 解析 SELECT 子句與聚合定義 (如 (sample(?number) as ?number))
local selectPart, whereAndTail = match(sparql, "[sS][eE][lL][eE][cC][tT]%s+(.-)%s+[wW][hH][eE][rR][eE]%s*(%b{}.*)$")
if not selectPart or not whereAndTail then return nil, true end
local whereBody, tailPart = match(whereAndTail, "^(%b{})(.*)$")
if not whereBody then return nil, true end
whereBody = sub(whereBody, 2, -2)
local aggregations = {}
local selectVars = {}
local cleanSelect = selectPart
cleanSelect = gsub(cleanSelect, "^[dD][iI][sS][tT][iI][nN][cC][tT]%s+", "")
cleanSelect = gsub(cleanSelect, "^[rR][eE][dD][uU][cC][eE][dD]%s+", "")
-- 萃取 (AGG(?src) AS ?alias) 或 (EXPR AS ?alias)
cleanSelect = gsub(cleanSelect, "%b()", function(parenBlock)
local inner = sub(parenBlock, 2, -2)
local fn, srcVar, aliasVar = match(inner, "^%s*([%a_]+)%s*%(%s*%?([%w_]+)%s*%)%s+[aA][sS]%s+%?([%w_]+)%s*$")
if fn and srcVar and aliasVar then
aggregations[#aggregations + 1] = {
op = lower(fn),
src = srcVar,
as = aliasVar
}
selectVars[#selectVars + 1] = aliasVar
end
return " "
end)
for v in gmatch(cleanSelect, "%?([%w_]+)") do
selectVars[#selectVars + 1] = v
end
-- 2. 解析服務語言 (SERVICE wikibase:label { bd:serviceParam wikibase:language "zh,en" })
local langList = { "zh-hant", "zh-tw", "zh", "en" }
local rawLangs = match(whereBody, "wikibase:language%s+[\"']([^\"']+)[\"']")
if rawLangs then
local parsedLangs = {}
for lg in gmatch(rawLangs, "[^,%s]+") do
if lg ~= "[AUTO_LANGUAGE]" then
parsedLangs[#parsedLangs + 1] = lg
end
end
if #parsedLangs > 0 then langList = parsedLangs end
end
-- 3. 萃取目標 Q-ID 清單與綁定變數名 (支援 VALUES ?item { wd:Q... } 或直接 wd:Q...)
local itemVar = "item"
local qids = {}
local qidSeen = {}
local function addQid(q)
if q and not qidSeen[q] and #qids < MAX_ENTITIES then
qidSeen[q] = true
qids[#qids + 1] = q
end
end
local valVar, valBlock = match(whereBody, "[vV][aA][lL][uU][eE][sS]%s+%?([%w_]+)%s*(%b{})")
if valVar and valBlock then
itemVar = valVar
for q in gmatch(valBlock, "wd:(Q%d+)") do addQid(q) end
end
if #qids == 0 then
for q in gmatch(whereBody, "wd:(Q%d+)%s+w?d?t?p?:P%d+") do addQid(q) end
end
if #qids == 0 and mw.wikibase.getEntityIdForCurrentPage then
local curQ = mw.wikibase.getEntityIdForCurrentPage()
if curQ then addQid(curQ) end
end
if #qids == 0 then return nil, true end
-- 4. 萃取 BIND 與 FILTER 並從 whereBody 移除，避免干擾三元組解析
local binds = {}
local filters = {}
local tripleBody = whereBody
tripleBody = gsub(tripleBody, "[mM][iI][nN][uU][sS]%s*%b{}", " ")
tripleBody = gsub(tripleBody, "[sS][eE][rR][vV][iI][cC][eE]%s+wikibase:label%s*%b{}", " ")
tripleBody = gsub(tripleBody, "[bB][iI][nN][dD]%s*(%b())", function(bBlock)
local inner = sub(bBlock, 2, -2)
local exprPart, asVar = match(inner, "^(.-)%s+[aA][sS]%s+%?([%w_]+)%s*$")
if exprPart and asVar then
binds[#binds + 1] = { expr = translateSparqlExpr(exprPart), as = asVar }
end
return " "
end)
tripleBody = gsub(tripleBody, "[fF][iI][lL][tT][eE][rR]%s*(%b())", function(fBlock)
filters[#filters + 1] = translateSparqlExpr(fBlock)
return " "
end)
-- 5. 解析三元組：支援聲明模式 (?item p:Pxxx ?stmt; ?stmt ps:Pxxx ?val; ?stmt pq:Pyyy ?qual) 與直接模式 (wdt:Pxxx)
local stmtPid = nil
local stmtVar = nil
local psVar = nil
local pqVars = {} -- { { pid = "P585", var = "time" }, ... }
local pqConstraints = {} -- { { pid = "P518", qid = "Q..." }, ... }
local wdtProps = {} -- { { pid = "P1082", var = "pop" }, ... }
local sItemVar, pPid, sVar = match(tripleBody, "%?([%w_]+)%s+p:(P%d+)%s+%?([%w_]+)")
if pPid and sVar then
itemVar = sItemVar or itemVar
stmtPid = pPid
stmtVar = sVar
else
local qSubj, pPid2, sVar2 = match(tripleBody, "wd:Q%d+%s+p:(P%d+)%s+%?([%w_]+)")
if pPid2 and sVar2 then
stmtPid = pPid2
stmtVar = sVar2
end
end
if stmtVar then
for pid, vName in gmatch(tripleBody, "%?" .. stmtVar .. "%s+psv?:(P%d+)%s+%?([%w_]+)") do
psVar = vName
end
for pid, vName in gmatch(tripleBody, "%?" .. stmtVar .. "%s+pqv?:(P%d+)%s+%?([%w_]+)") do
pqVars[#pqVars + 1] = { pid = pid, var = vName }
end
for pid, targetQ in gmatch(tripleBody, "%?" .. stmtVar .. "%s+pq:(P%d+)%s+wd:(Q%d+)") do
pqConstraints[#pqConstraints + 1] = { pid = pid, qid = targetQ }
end
end
for _, pid, vName in gmatch(tripleBody, "(%?[%w_]+)%s+wdt:(P%d+)%s+%?([%w_]+)") do
wdtProps[#wdtProps + 1] = { pid = pid, var = vName }
end
if not stmtPid and #wdtProps == 0 then
for pid, vName in gmatch(tripleBody, "wd:Q%d+%s+wdt:(P%d+)%s+%?([%w_]+)") do
wdtProps[#wdtProps + 1] = { pid = pid, var = vName }
end
end
if not stmtPid and #wdtProps == 0 then return nil, true end
-- 6. 批次讀取 Wikidata Statements 並組裝資料列
local labelCache = {}
local rawRows = {}
local getStmtsFn = mw.wikibase.getAllStatements or mw.wikibase.getBestStatements
for _, qid in ipairs(qids) do
local itemLabel = getEntityLabel(qid, langList, labelCache)
-- 先讀取附加的 wdt: 單值屬性
local baseWdtVals = {}
for _, wp in ipairs(wdtProps) do
local best = mw.wikibase.getBestStatements(qid, wp.pid)
if best and best[1] and best[1].mainsnak then
baseWdtVals[wp.var] = parseSnakValue(best[1].mainsnak)
end
end
if stmtPid then
local stmts = getStmtsFn(qid, stmtPid) or {}
for _, st in ipairs(stmts) do
if st.rank ~= "deprecated" then
local passConstraint = true
for _, cons in ipairs(pqConstraints) do
local qList = st.qualifiers and st.qualifiers[cons.pid]
local matched = false
if type(qList) == "table" then
for _, qSnak in ipairs(qList) do
if parseSnakValue(qSnak) == cons.qid then
matched = true
break
end
end
end
if not matched then passConstraint = false; break end
end
if passConstraint then
local mainVal = parseSnakValue(st.mainsnak)
if mainVal ~= nil or not psVar then
local row = {
[itemVar] = qid,
[itemVar .. "Label"] = itemLabel
}
for k, v in pairs(baseWdtVals) do row[k] = v end
if psVar then row[psVar] = mainVal end
local hasAllQuals = true
for _, pq in ipairs(pqVars) do
local qList = st.qualifiers and st.qualifiers[pq.pid]
local qVal = (type(qList) == "table" and qList[1]) and parseSnakValue(qList[1]) or nil
if qVal == nil then
hasAllQuals = false
break
end
row[pq.var] = qVal
if type(qVal) == "string" and match(qVal, "^Q%d+$") then
row[pq.var .. "Label"] = getEntityLabel(qVal, langList, labelCache)
end
end
if hasAllQuals then
rawRows[#rawRows + 1] = row
end
end
end
end
end
else
-- 純 wdt: 模式
local row = {
[itemVar] = qid,
[itemVar .. "Label"] = itemLabel
}
local hasAny = false
for k, v in pairs(baseWdtVals) do
row[k] = v
hasAny = true
end
if hasAny then rawRows[#rawRows + 1] = row end
end
end
-- 7. 執行 BIND 與 FILTER
local filteredRows = {}
for i = 1, #rawRows do
local r = rawRows[i]
for _, b in ipairs(binds) do
local okB, valB = Expr.evaluate(b.expr, { datum = r })
if okB then r[b.as] = valB end
end
local keep = true
for _, fExpr in ipairs(filters) do
local okF, resF = Expr.evaluate(fExpr, { datum = r })
if okF and not Expr.isTruthy(resF) then
keep = false
break
end
end
if keep then
filteredRows[#filteredRows + 1] = r
end
end
-- 8. 執行 GROUP BY 與聚合運算 (如 (sample(?number) as ?number))
local groupTail = match(tailPart or "", "[gG][rR][oO][uU][pP]%s+[bB][yY]%s+(.-)$")
if groupTail then
groupTail = gsub(groupTail, "[oO][rR][dD][eE][rR]%s+[bB][yY].*$", "")
groupTail = gsub(groupTail, "[lL][iI][mM][iI][tT]%s+%d+.*$", "")
end
if (groupTail and find(groupTail, "%?")) or #aggregations > 0 then
local groupVars = {}
if groupTail then
for gv in gmatch(groupTail, "%?([%w_]+)") do
groupVars[#groupVars + 1] = gv
end
end
local transforms = {
{
type = "aggregate",
groupby = groupVars,
summarize = {}
}
}
for _, agg in ipairs(aggregations) do
local opName = agg.op
if opName == "sample" then opName = "max"
elseif opName == "avg" then opName = "mean" end
transforms[1].summarize[#transforms[1].summarize + 1] = {
field = agg.src,
ops = { opName },
as = { agg.as }
}
end
local aggRes = Expr.applyTransforms(filteredRows, transforms)
if aggRes then filteredRows = aggRes end
end
-- 9. 執行 ORDER BY (支援多鍵與 ASC / DESC，例如 order by ?item desc (?year))
local orderTail = match(tailPart or "", "[oO][rR][dD][eE][rR]%s+[bB][yY]%s+(.-)$")
if orderTail then
orderTail = gsub(orderTail, "[lL][iI][mM][iI][tT]%s+%d+.*$", "")
local sortKeys = {}
local pos = 1
while pos <= #orderTail do
local s1, e1, dir, v1 = find(orderTail, "([dD][eE][sS][cC])%s*%(%s*%?([%w_]+)%s*%)", pos)
local s2, e2, _, v2 = find(orderTail, "([aA][sS][cC])%s*%(%s*%?([%w_]+)%s*%)", pos)
local s3, e3, v3 = find(orderTail, "%?([%w_]+)", pos)
if s1 and (not s3 or s1 <= s3) and (not s2 or s1 <= s2) then
sortKeys[#sortKeys + 1] = { field = v1, desc = true }
pos = e1 + 1
elseif s2 and (not s3 or s2 <= s3) then
sortKeys[#sortKeys + 1] = { field = v2, desc = false }
pos = e2 + 1
elseif s3 then
sortKeys[#sortKeys + 1] = { field = v3, desc = false }
pos = e3 + 1
else
break
end
end
if #sortKeys > 0 then
table.sort(filteredRows, function(a, b)
if type(a) ~= "table" or type(b) ~= "table" then return false end
for k = 1, #sortKeys do
local sk = sortKeys[k]
local va, vb = a[sk.field], b[sk.field]
if va ~= vb then
if va == nil then return false end
if vb == nil then return true end
local na, nb = tonumber(va), tonumber(vb)
if na and nb then
if na ~= nb then
if sk.desc then return na > nb else return na < nb end
end
else
local sa, sb = tostring(va), tostring(vb)
if sa ~= sb then
if sk.desc then return sa > sb else return sa < sb end
end
end
end
end
return false
end)
end
end
-- 10. 執行 LIMIT
local limitNum = tonumber(match(tailPart or "", "[lL][iI][mM][iI][tT]%s+(%d+)"))
if limitNum and limitNum > 0 and #filteredRows > limitNum then
local sliced = {}
for i = 1, limitNum do sliced[i] = filteredRows[i] end
filteredRows = sliced
end
return (#filteredRows > 0) and filteredRows or nil, (#filteredRows == 0)
end
return p
