55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
# coding=utf-8
|
||
"""手动测试管理系统基地接口同步。
|
||
|
||
用法：
|
||
python tools/test_mgmt_bases.py
|
||
python tools/test_mgmt_bases.py --sync
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
if ROOT not in sys.path:
|
||
sys.path.insert(0, ROOT)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='测试管理系统基地接口')
|
||
parser.add_argument('--sync', action='store_true', help='拉取后写入 MySQL')
|
||
parser.add_argument('--url', default=None)
|
||
parser.add_argument('--key', default=None)
|
||
args = parser.parse_args()
|
||
|
||
from service.mgmt_bases_api import MgmtApiError, fetch_bases, map_remote_base
|
||
|
||
try:
|
||
items = fetch_bases(url=args.url, api_key=args.key)
|
||
except MgmtApiError as exc:
|
||
print(f'[FAIL] fetch: {exc}')
|
||
print('请检查 DISH_MGMT_API_URL / DISH_MGMT_API_KEY，或用 --url/--key 传入。')
|
||
return 1
|
||
|
||
print(f'[OK] fetched {len(items)} bases')
|
||
mapped = [map_remote_base(x) for x in items]
|
||
mapped = [m for m in mapped if m]
|
||
print(f'[OK] mapped {len(mapped)}')
|
||
if mapped:
|
||
print(json.dumps(mapped[0], ensure_ascii=False, indent=2))
|
||
|
||
if args.sync:
|
||
from dao.database import init_local_engine, init_server_engine
|
||
from dao.base_sync import sync_bases_from_mgmt
|
||
init_local_engine()
|
||
init_server_engine()
|
||
result = sync_bases_from_mgmt(force=True, url=args.url, api_key=args.key)
|
||
print('[SYNC]', result)
|
||
return 0 if result.get('ok') else 2
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
raise SystemExit(main())
|