117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""校验生成的 INSERT 脚本: 元组数量/字段个数/列序/与缓存数据一致性。用完可删。"""
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
|
from fetch_food_nutrition import to_num # noqa: E402
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
SQL = ROOT / "src" / "main" / "resources" / "db" / "data_food_nutrition.sql"
|
|
DDL = ROOT / "src" / "main" / "resources" / "db" / "migration_add_food_nutrition.sql"
|
|
CACHE = ROOT / "tools" / "_cache"
|
|
N_META, N_NUT, EXPECT = 9, 31, 41 # 9 元数据 + 31 营养素 + raw_values
|
|
|
|
|
|
def split_top(text):
|
|
"""按最外层逗号/分号切分, 尊重单引号字符串(含 '' 转义)与括号嵌套。"""
|
|
out, cur, depth, in_str, i = [], [], 0, False, 0
|
|
while i < len(text):
|
|
ch = text[i]
|
|
if in_str:
|
|
if ch == "'":
|
|
if i + 1 < len(text) and text[i + 1] == "'":
|
|
cur.append("''")
|
|
i += 2
|
|
continue
|
|
in_str = False
|
|
cur.append(ch)
|
|
elif ch == "'":
|
|
in_str = True
|
|
cur.append(ch)
|
|
elif ch in "([{":
|
|
depth += 1
|
|
cur.append(ch)
|
|
elif ch in ")]}":
|
|
depth -= 1
|
|
cur.append(ch)
|
|
elif depth == 0 and ch == ",":
|
|
out.append("".join(cur).strip())
|
|
cur = []
|
|
else:
|
|
cur.append(ch)
|
|
i += 1
|
|
if "".join(cur).strip():
|
|
out.append("".join(cur).strip())
|
|
return out
|
|
|
|
|
|
sql = SQL.read_text(encoding="utf-8")
|
|
body = "\n".join(l for l in sql.splitlines() if not l.lstrip().startswith("--"))
|
|
stmts = [s for s in body.split(";") if "INSERT INTO" in s]
|
|
ins_cols = re.findall(r"`(\w+)`", stmts[0][:stmts[0].index("VALUES")])[1:] # 去掉表名
|
|
print(f"INSERT 语句 {len(stmts)} 条, 列数 {len(ins_cols)}")
|
|
|
|
# 1) DDL 列序 vs INSERT 列序(排除 id/create_time/update_time)
|
|
ddl = DDL.read_text(encoding="utf-8")
|
|
block = ddl[ddl.index("CREATE TABLE"):ddl.index("ENGINE=InnoDB")]
|
|
ddl_cols = [c for c in re.findall(r"^\s*`(\w+)`\s+(?:BIGINT|INT|VARCHAR|DECIMAL|JSON|DATETIME)",
|
|
block, re.M)
|
|
if c not in ("id", "create_time", "update_time")]
|
|
print(f"DDL 列数 {len(ddl_cols)}; 列序与 INSERT 一致: {ddl_cols == ins_cols}")
|
|
if ddl_cols != ins_cols:
|
|
for a, b in zip(ddl_cols, ins_cols):
|
|
if a != b:
|
|
print(f" 首个差异: DDL={a} INSERT={b}")
|
|
break
|
|
|
|
# 2) 元组解析
|
|
rows, bad = [], []
|
|
for st in stmts:
|
|
for tup in split_top(st[st.index("VALUES") + 6:]):
|
|
t = tup.strip()
|
|
t = t[1:-1] if t.startswith("(") and t.endswith(")") else t
|
|
f = split_top(t)
|
|
if len(f) != EXPECT:
|
|
bad.append((f[0] if f else "?", len(f)))
|
|
else:
|
|
rows.append(f)
|
|
print(f"解析记录 {len(rows)} 条; 字段数非 {EXPECT} 的 {len(bad)} {bad[:3]}")
|
|
|
|
ids = [r[0] for r in rows]
|
|
print(f"food_id 唯一 {len(set(ids))}; 升序 {ids == sorted(ids, key=int)}; "
|
|
f"范围 {min(map(int, ids))}~{max(map(int, ids))}")
|
|
|
|
# 3) 与接口缓存比对
|
|
master = {}
|
|
for f in CACHE.glob("cat_0_0_p*.json"):
|
|
for row in json.loads(f.read_text(encoding="utf-8")).get("list") or []:
|
|
master[int(row[0])] = row
|
|
print(f"缓存 {len(master)} 条; ID 集合一致: {set(map(int, ids)) == set(master)}")
|
|
|
|
mis = badj = 0
|
|
for r in rows:
|
|
src = master[int(r[0])]
|
|
got = r[N_META:N_META + N_NUT]
|
|
for n, idx in enumerate(range(5, 36)):
|
|
exp = to_num(src[idx])
|
|
if (exp is None and got[n] != "NULL") or (exp is not None and got[n] != exp):
|
|
mis += 1
|
|
if mis <= 5:
|
|
print(f" 值不一致 id={r[0]} idx={idx} 期望 {exp} 实际 {got[n]}")
|
|
if json.loads(r[N_META + N_NUT].strip()[1:-1].replace("''", "'")) != src:
|
|
badj += 1
|
|
print(f"31 个营养素逐值比对: {'全部一致' if mis == 0 else f'{mis} 处不一致'}")
|
|
print(f"raw_values 为合法 JSON 且与原数组一致: {badj == 0} (异常 {badj})")
|
|
|
|
print("\n样例:")
|
|
for r in rows:
|
|
if r[1][:2] in ("小麦", "鸡肝", "花生") or r[1] == "小麦粉(标准粉)":
|
|
print(" " + " | ".join(r[:N_META + 8]))
|
|
if r[1].startswith("小麦粉(标准粉)"):
|
|
break
|
|
print(" 首行: " + " | ".join(rows[0][:N_META + 8]))
|
|
print(" raw_values(首行前 60 字): " + rows[0][N_META + N_NUT][:60])
|