217 lines
7.9 KiB
Python
217 lines
7.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
远端修复：inq_quote_template_version.mapping_json 非法（PowerShell ConvertTo-Json/种子写入无引号）导致 previewSample 失败。
|
||
对每条版本：若 JSON 不可解析，则按上传 xlsx + 标签词典重算合法 mapping_json。
|
||
副作用：UPDATE MySQL mapping_json / mapping_point_count / parse_message；不改 object 文件。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
try:
|
||
import pymysql
|
||
except ImportError:
|
||
pymysql = None
|
||
|
||
try:
|
||
from openpyxl import load_workbook
|
||
from openpyxl.utils import get_column_letter
|
||
except ImportError:
|
||
load_workbook = None
|
||
get_column_letter = None
|
||
|
||
|
||
def load_env(path: Path) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
if not path.is_file():
|
||
return out
|
||
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, v = line.split("=", 1)
|
||
out[k.strip()] = v.strip().strip('"')
|
||
return out
|
||
|
||
|
||
def try_parse(s: str):
|
||
if not s:
|
||
return None
|
||
try:
|
||
return json.loads(s)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def repair_ps_style(s: str):
|
||
"""尽力把无引号对象字面量修成 JSON（仅用于种子脏数据）。"""
|
||
if not s or s[0] != "{":
|
||
return None
|
||
# 已是合法 JSON
|
||
if try_parse(s) is not None:
|
||
return try_parse(s)
|
||
# 给 key 加引号：{foo: 或 ,foo:
|
||
t = re.sub(r"([,{]\s*)([A-Za-z_][A-Za-z0-9_]*)\s*:", r'\1"\2":', s)
|
||
# 给像 Sheet2!B8 / quote-template-mapping-v1 / 纯标识 的裸字符串值加引号
|
||
# 匹配 :value 直到 ,} ]
|
||
def quote_val(m):
|
||
prefix, val, suffix = m.group(1), m.group(2), m.group(3)
|
||
val = val.strip()
|
||
if val in ("[]", "{}", "true", "false", "null") or re.match(r"^-?\d+(\.\d+)?$", val):
|
||
return prefix + val + suffix
|
||
if (val.startswith('"') and val.endswith('"')) or (val.startswith("{") or val.startswith("[")):
|
||
return prefix + val + suffix
|
||
return prefix + json.dumps(val, ensure_ascii=False) + suffix
|
||
|
||
t2 = re.sub(r'(:\s*)([^,\[\]{}]+)([,}\]])', quote_val, t)
|
||
return try_parse(t2)
|
||
|
||
|
||
def rough_fields(xlsx: Path, label_dict: dict[str, str]) -> dict[str, str]:
|
||
fields: dict[str, str] = {}
|
||
if load_workbook is None:
|
||
return fields
|
||
wb = load_workbook(xlsx, read_only=True, data_only=True)
|
||
try:
|
||
# 扫所有 sheet（种子映射常在 Sheet2）
|
||
for ws in wb.worksheets:
|
||
sheet = ws.title
|
||
for row in ws.iter_rows(max_row=80, max_col=24):
|
||
for cell in row:
|
||
val = cell.value
|
||
if val is None:
|
||
continue
|
||
text = str(val).strip()
|
||
if not text:
|
||
continue
|
||
norm = text.replace("：", ":")
|
||
code = None
|
||
for lab, c in label_dict.items():
|
||
lab_n = lab.replace("：", ":").replace(":", "").strip()
|
||
t_n = norm.replace(":", "").strip()
|
||
if lab_n and (t_n == lab_n or t_n.startswith(lab_n)):
|
||
code = c
|
||
break
|
||
if code and code not in fields:
|
||
col_i = int(getattr(cell, "column", 1) or 1)
|
||
fill = f"{sheet}!{get_column_letter(col_i + 1)}{cell.row}"
|
||
fields[code] = fill
|
||
finally:
|
||
wb.close()
|
||
return fields
|
||
|
||
|
||
def main() -> int:
|
||
env = load_env(Path(r"E:\wwwroot\.env"))
|
||
upload = env.get("JEECG_UPLOAD_PATH") or r"E:\wwwroot\data\upFiles"
|
||
user = env.get("MYSQL_USER") or "inquiry_robot"
|
||
password = env.get("MYSQL_PASSWORD") or ""
|
||
db = env.get("MYSQL_DB") or "inquiry_robot"
|
||
host = "127.0.0.1"
|
||
|
||
contract_paths = [
|
||
Path(r"E:\wwwroot\inquiry-new\config\contracts\quote-template-mapping-v1.json"),
|
||
Path(r"C:\Users\ytd_extdev_0903\incoming\quote-template-mapping-v1.json"),
|
||
]
|
||
label_dict: dict[str, str] = {}
|
||
for cp in contract_paths:
|
||
if cp.is_file():
|
||
label_dict = (json.loads(cp.read_text(encoding="utf-8")).get("labelDictionary") or {})
|
||
break
|
||
if not label_dict:
|
||
# 最小词典，避免完全空
|
||
label_dict = {
|
||
"客户名称": "customer_name",
|
||
"客户": "customer_name",
|
||
"报价单编号": "quote_no",
|
||
"报价日期": "quote_date",
|
||
"起运港": "pol",
|
||
"目的港": "pod",
|
||
"业务员": "sales_name",
|
||
}
|
||
|
||
if pymysql is None:
|
||
print("ERR no pymysql")
|
||
return 2
|
||
|
||
conn = pymysql.connect(
|
||
host=host, user=user, password=password, database=db, charset="utf8mb4", port=3306
|
||
)
|
||
fixed = 0
|
||
ok_already = 0
|
||
failed = 0
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT id, object_key, file_name, mapping_json FROM inq_quote_template_version"
|
||
)
|
||
rows = cur.fetchall()
|
||
for vid, object_key, file_name, mapping_json in rows:
|
||
parsed = try_parse(mapping_json or "")
|
||
if parsed is None:
|
||
parsed = repair_ps_style(mapping_json or "")
|
||
xlsx = Path(upload) / str(object_key).replace("/", os.sep)
|
||
if parsed is None or not isinstance(parsed.get("fields"), dict) or not parsed.get("fields"):
|
||
fields = {}
|
||
if xlsx.is_file():
|
||
fields = rough_fields(xlsx, label_dict)
|
||
# 若修复出的 fields 可用则保留其它键
|
||
if parsed and isinstance(parsed.get("fields"), dict) and parsed.get("fields"):
|
||
fields = parsed["fields"]
|
||
elif not fields and parsed and isinstance(parsed.get("fields"), dict):
|
||
fields = parsed["fields"]
|
||
mapping = {
|
||
"schemaVersion": "quote-template-mapping-v1",
|
||
"fields": fields,
|
||
"fixedYellowCells": [],
|
||
"unmappedRequired": [],
|
||
"note": "repaired_mapping_json",
|
||
}
|
||
else:
|
||
# 合法但规范化
|
||
mapping = {
|
||
"schemaVersion": parsed.get("schemaVersion") or "quote-template-mapping-v1",
|
||
"fields": parsed.get("fields") or {},
|
||
"fixedYellowCells": parsed.get("fixedYellowCells") or [],
|
||
"unmappedRequired": parsed.get("unmappedRequired") or [],
|
||
}
|
||
if try_parse(mapping_json or "") is not None and mapping_json.strip().startswith('{"'):
|
||
ok_already += 1
|
||
print(f"OK_ALREADY {vid} points={len(mapping['fields'])}")
|
||
continue
|
||
|
||
new_json = json.dumps(mapping, ensure_ascii=False, separators=(",", ":"))
|
||
points = len(mapping.get("fields") or {})
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE inq_quote_template_version SET mapping_json=%s, mapping_point_count=%s, "
|
||
"parse_status=%s, parse_message=%s WHERE id=%s",
|
||
(
|
||
new_json,
|
||
points,
|
||
"parsed",
|
||
f"repaired mapping, {points} points",
|
||
vid,
|
||
),
|
||
)
|
||
conn.commit()
|
||
fixed += 1
|
||
exists = xlsx.is_file()
|
||
print(f"FIXED {vid} points={points} file_exists={exists} name={file_name}")
|
||
except Exception as e:
|
||
failed += 1
|
||
print("ERR", e)
|
||
return 1
|
||
finally:
|
||
conn.close()
|
||
print(f"REPAIR_DONE fixed={fixed} ok_already={ok_already} failed={failed}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|