139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
导入 init/data.json 中的 appcodes / appcodes_kv 编码字典到 sage 数据库。
|
||
|
||
用法:
|
||
cd ~/repos/sage
|
||
py3/bin/python import_codes.py ../supplychain/init/data.json
|
||
py3/bin/python import_codes.py ../product_management/init/data.json ../supplychain/init/data.json
|
||
"""
|
||
import sys
|
||
import json
|
||
import os
|
||
import asyncio
|
||
import argparse
|
||
|
||
# 修正 venv site-packages 路径（venv 可能是旧版 Python 创建的）
|
||
_script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
_venv_lib = os.path.join(_script_dir, 'py3', 'lib')
|
||
if os.path.isdir(_venv_lib):
|
||
for _d in os.listdir(_venv_lib):
|
||
_sp = os.path.join(_venv_lib, _d, 'site-packages')
|
||
if os.path.isdir(_sp) and _sp not in sys.path:
|
||
sys.path.insert(0, _sp)
|
||
|
||
# 添加本地依赖路径（editable install 可能指向 Docker 路径）
|
||
_repos_dir = os.path.dirname(_script_dir)
|
||
for _dep in ['apppublic', 'sqlor']:
|
||
_dep_dir = os.path.join(_repos_dir, _dep)
|
||
if os.path.isdir(_dep_dir) and _dep_dir not in sys.path:
|
||
sys.path.insert(0, _dep_dir)
|
||
|
||
from appPublic.jsonConfig import getConfig
|
||
from sqlor.dbpools import DBPools
|
||
|
||
|
||
async def import_data_json(sor, filepath):
|
||
"""将单个 data.json 中的 appcodes 和 appcodes_kv 导入数据库"""
|
||
with open(filepath, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
# 导入 appcodes — 支持两种格式:
|
||
# 格式1: [{"id":..., "name":..., "hierarchy_flg":...}]
|
||
# 格式2: [{"parentid":..., "parentname":..., "items":[...]}]
|
||
appcodes = data.get('appcodes', [])
|
||
codes_count = 0
|
||
kv_count = 0
|
||
|
||
for item in appcodes:
|
||
# 格式2: parentid/parentname + 内嵌 items
|
||
if 'parentid' in item and 'items' in item:
|
||
parentid = item['parentid']
|
||
parentname = item.get('parentname', parentid)
|
||
await sor.sqlExe("""
|
||
INSERT INTO appcodes (id, name, hierarchy_flg)
|
||
VALUES (${id}$, ${name}$, ${hierarchy_flg}$)
|
||
ON DUPLICATE KEY UPDATE name=VALUES(name), hierarchy_flg=VALUES(hierarchy_flg)
|
||
""", {
|
||
'id': parentid,
|
||
'name': parentname,
|
||
'hierarchy_flg': item.get('hierarchy_flg', '0')
|
||
})
|
||
codes_count += 1
|
||
# 导入内嵌的 kv 条目
|
||
for kv_item in item.get('items', []):
|
||
kv_id = f"{parentid}_{kv_item['k']}"
|
||
await sor.sqlExe("""
|
||
INSERT INTO appcodes_kv (id, parentid, k, v)
|
||
VALUES (${id}$, ${parentid}$, ${k}$, ${v}$)
|
||
ON DUPLICATE KEY UPDATE id=VALUES(id), v=VALUES(v)
|
||
""", {
|
||
'id': kv_id,
|
||
'parentid': parentid,
|
||
'k': kv_item['k'],
|
||
'v': kv_item['v']
|
||
})
|
||
kv_count += 1
|
||
# 格式1: 标准 id/name
|
||
elif 'id' in item:
|
||
await sor.sqlExe("""
|
||
INSERT INTO appcodes (id, name, hierarchy_flg)
|
||
VALUES (${id}$, ${name}$, ${hierarchy_flg}$)
|
||
ON DUPLICATE KEY UPDATE name=VALUES(name), hierarchy_flg=VALUES(hierarchy_flg)
|
||
""", {
|
||
'id': item['id'],
|
||
'name': item['name'],
|
||
'hierarchy_flg': item.get('hierarchy_flg', '0')
|
||
})
|
||
codes_count += 1
|
||
|
||
print(f" appcodes: {codes_count} 条")
|
||
|
||
# 导入独立 appcodes_kv (格式1兼容)
|
||
appcodes_kv = data.get('appcodes_kv', [])
|
||
for item in appcodes_kv:
|
||
await sor.sqlExe("""
|
||
INSERT INTO appcodes_kv (id, parentid, k, v)
|
||
VALUES (${id}$, ${parentid}$, ${k}$, ${v}$)
|
||
ON DUPLICATE KEY UPDATE id=VALUES(id), v=VALUES(v)
|
||
""", {
|
||
'id': item['id'],
|
||
'parentid': item['parentid'],
|
||
'k': item['k'],
|
||
'v': item['v']
|
||
})
|
||
kv_count += 1
|
||
if appcodes_kv:
|
||
print(f" appcodes_kv: {len(appcodes_kv)} 条 (独立)")
|
||
|
||
print(f" appcodes_kv: {kv_count} 条 (合计)")
|
||
return codes_count, kv_count
|
||
|
||
|
||
async def main():
|
||
parser = argparse.ArgumentParser(description='导入编码字典到 sage 数据库')
|
||
parser.add_argument('files', nargs='+', help='init/data.json 文件路径')
|
||
args = parser.parse_args()
|
||
|
||
# 使用 getConfig() 获取配置，DBPools 初始化数据库池
|
||
config = getConfig('.')
|
||
db = DBPools(config.databases)
|
||
|
||
print(f"连接数据库: sage")
|
||
|
||
total_codes = 0
|
||
total_kv = 0
|
||
|
||
async with db.sqlorContext('sage') as sor:
|
||
for filepath in args.files:
|
||
print(f"导入: {filepath}")
|
||
c, kv = await import_data_json(sor, filepath)
|
||
total_codes += c
|
||
total_kv += kv
|
||
|
||
print(f"\n完成! 共导入 appcodes: {total_codes} 条, appcodes_kv: {total_kv} 条")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
asyncio.run(main())
|