c86d02552f
- Add filter_xdp_device_info function to query device by PSN or passcode - Display device info in formatted table, excluding RSA keys and internal fields - Format timestamps using dataintel utility - Add interactive menu option for device info query 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
355 lines
11 KiB
Python
355 lines
11 KiB
Python
import os
|
|
import sys
|
|
import django
|
|
|
|
# --- 关键的 Django 设置 ---
|
|
project_path = '/srv/sso'
|
|
if project_path not in sys.path:
|
|
sys.path.append(project_path)
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'sso.settings')
|
|
django.setup()
|
|
|
|
import canton.models
|
|
from canton.models import *
|
|
import bson
|
|
import uuid
|
|
from utils import const
|
|
import hydrogen.flex_engine
|
|
import requests
|
|
from django.utils.timezone import get_current_timezone, localtime, now
|
|
import datetime
|
|
import utils.const
|
|
import django.conf
|
|
import utils.dataintel
|
|
|
|
|
|
# --- 功能函数定义 ---
|
|
def get_engine(table_name):
|
|
return hydrogen.flex_engine.FlexEngine.get_engine_by_name(
|
|
miniapp_id=44328,
|
|
schema_name=table_name,
|
|
enable_acl_check=False)
|
|
|
|
|
|
def gen_code(info_id, count, print_only=False):
|
|
info_engine = get_engine('license_info')
|
|
code_engine = get_engine('license_redeem_code')
|
|
|
|
info = info_engine.get_by_id(info_id)
|
|
|
|
uuids = []
|
|
for x in range(count):
|
|
uuids.append(str(uuid.uuid4()))
|
|
|
|
for uid in uuids:
|
|
code_engine.create(
|
|
code=uid,
|
|
status='valid',
|
|
valid_from=info['valid_from'],
|
|
valid_until=info['valid_until'],
|
|
license_info=bson.dbref.DBRef(
|
|
info_engine.collection.name,
|
|
info['_id']),
|
|
created_by=27068886,
|
|
_read_perm=[const.HYDROGEN_ACL_CREATED_BY_TPL],
|
|
_write_perm=[],
|
|
)
|
|
if print_only:
|
|
print(uid)
|
|
if print_only:
|
|
return
|
|
# save uuids to csv
|
|
fn = '/tmp/%s-%s.csv' % (info['name'].replace(' ', ''), localtime().strftime('%Y%m%d'))
|
|
with open(fn, 'a') as f:
|
|
for uid in uuids:
|
|
f.write(uid + '\n')
|
|
|
|
print(fn)
|
|
|
|
|
|
def issue_license(code, psn):
|
|
code_engine = get_engine('license_redeem_code')
|
|
code = code_engine.filter(code=code, status=utils.const.TANGZHI_LICENSE_REDEEM_CODE_STATUS_VALID)[0]
|
|
if not code:
|
|
return False
|
|
code_engine = get_engine('license_redeem_code')
|
|
license_info = code_engine.dereference(code['license_info'])
|
|
feature_engine = get_engine('license_feature')
|
|
license_data = {
|
|
'license': {
|
|
'psn': psn,
|
|
'features': [],
|
|
'signing_key': 'IFANR_IOT_KEY',
|
|
},
|
|
'enable_cloud_push': True}
|
|
for feature_id in license_info['license_feature']:
|
|
feature = feature_engine.get(feature_id=feature_id)
|
|
if feature and 'CP02' in feature['applicable_devices']:
|
|
license_data['license']['features'].append({
|
|
'feature_id': feature_id,
|
|
'applicable_devices': feature['applicable_devices'],
|
|
})
|
|
if not license_data['license']['features']:
|
|
return
|
|
if license_info['license_type'] == utils.const.TANGZHI_LICENSE_TYPE_TIMED:
|
|
if license_info.get('expiry_days'):
|
|
license_data['license']['not_before'] = utils.timestamp_of(now())
|
|
license_data['license']['not_after'] = utils.timestamp_of(
|
|
now() + datetime.timedelta(
|
|
days=license_info['expiry_days']))
|
|
else:
|
|
license_data['license']['not_before'] = license_info['not_before']
|
|
license_data['license']['not_after'] = license_info['not_after']
|
|
elif license_info['license_type'] == utils.const.TANGZHI_LICENSE_TYPE_COUNTER:
|
|
license_data['license']['max_usage'] = license_info['max_usage']
|
|
elif license_info['license_type'] == utils.const.TANGZHI_LICENSE_TYPE_RELATIVE_TIME:
|
|
license_data['license']['validity_hours'] = license_info['validity_hours']
|
|
|
|
resp = requests.post(
|
|
django.conf.settings.TANGZHI_LICENSE_ISSUE_URL,
|
|
json=license_data)
|
|
if resp.status_code != utils.const.HTTP_OK:
|
|
print('Failed to issue license: %s, data: %s' % (resp.text, license_data))
|
|
return
|
|
device_license_engine = get_engine(
|
|
django.conf.settings.IOT_DEVICE_LICENSE_SCHEMA_NAME)
|
|
snapshot = {
|
|
'license_redeem_code': code_engine.document_2_dict(code, dereference=True)
|
|
}
|
|
code_engine.update_by_id(
|
|
code['_id'],
|
|
psn=psn)
|
|
device_license_engine.update_by_id(
|
|
resp.json()['license']['id'],
|
|
snapshot=snapshot)
|
|
|
|
resp_data = device_license_engine.get_by_id(
|
|
resp.json()['license']['id'])
|
|
for feature in resp_data['license']['features']:
|
|
feature['tags'] = feature_engine.get(feature_id=feature['feature_id'])['tags']
|
|
|
|
return resp_data
|
|
|
|
|
|
def bind_product(product_id, batch_id, start, stop):
|
|
"""绑定商品到序列号批次"""
|
|
# 绑定商品
|
|
null_product = canton.models.ProductInfo.objects.filter(name='').first()
|
|
new_product = canton.models.ProductInfo.objects.filter(id=product_id).first()
|
|
|
|
psn__in = [_data['psn'] for _data in canton.models.SerialNumber.generate_serial_number(**{
|
|
'batch_id': batch_id,
|
|
'start': start,
|
|
'stop': stop,
|
|
'sign': False,
|
|
})]
|
|
|
|
sns = SerialNumber.objects.filter(psn__in=psn__in)
|
|
|
|
fail_sns = sns.filter(~Q(product=null_product) & ~Q(product=new_product))
|
|
if fail_sns:
|
|
raise ValueError("部分设备已绑定 %s" % fail_sns.values_list('psn', flat=True))
|
|
|
|
sns.update(product=new_product)
|
|
return sns.last().serialize()
|
|
|
|
|
|
def update_product(new_product_id, batch_id, start, stop):
|
|
"""更新商品绑定"""
|
|
# 更新商品
|
|
new_product = canton.models.ProductInfo.objects.filter(id=new_product_id).first()
|
|
|
|
psn__in = [_data['psn'] for _data in canton.models.SerialNumber.generate_serial_number(**{
|
|
'batch_id': batch_id,
|
|
'start': start,
|
|
'stop': stop,
|
|
'sign': False,
|
|
})]
|
|
|
|
sns = SerialNumber.objects.filter(psn__in=psn__in)
|
|
|
|
fail_sns = sns.filter(product=new_product)
|
|
if fail_sns:
|
|
raise ValueError("部分设备已绑定为当前商品 %s" % fail_sns.values_list('psn', flat=True))
|
|
|
|
sns.update(product=new_product)
|
|
return sns.last().serialize()
|
|
|
|
|
|
def filter_xdp_device_info(keyword):
|
|
engine = get_engine('device_info')
|
|
try:
|
|
if len(keyword) == 4:
|
|
result = engine.filter(passcode=keyword)[0]
|
|
else:
|
|
result = engine.filter(psn=keyword)[0]
|
|
except IndexError:
|
|
result = None
|
|
print("未找到设备信息")
|
|
return
|
|
|
|
# 定义 RSA 相关的字段
|
|
rsa_keys = ['rsa_public_key', 'rsa_private_key']
|
|
# 定义不需要在表格中展示的字段
|
|
exclude_fields = ['_read_perm', '_write_perm', 'created_by', 'mcu_uid']
|
|
|
|
# 分离 RSA 字段和其他字段
|
|
rsa_data = {}
|
|
table_data = {}
|
|
for key, value in result.items():
|
|
if key in rsa_keys:
|
|
rsa_data[key] = value
|
|
elif key in ['created_at', 'updated_at']:
|
|
table_data[key] = utils.dataintel.strf_time(utils.timestamp_to_local_datetime(value))
|
|
elif key not in exclude_fields:
|
|
table_data[key] = value
|
|
|
|
# 计算表格宽度
|
|
max_key_len = max(len(str(k)) for k in table_data.keys())
|
|
max_val_len = min(80, max(len(str(v)[:80]) for v in table_data.values()))
|
|
|
|
# 打印表格
|
|
print("\n" + "=" * (max_key_len + max_val_len + 7))
|
|
print(f"| {'field'.ljust(max_key_len)} | {'value'.ljust(max_val_len)} |")
|
|
print("|" + "-" * (max_key_len + 2) + "|" + "-" * (max_val_len + 2) + "|")
|
|
|
|
for key, value in table_data.items():
|
|
val_str = str(value)
|
|
if len(val_str) > 80:
|
|
val_str = val_str[:77] + "..."
|
|
print(f"| {str(key).ljust(max_key_len)} | {val_str.ljust(max_val_len)} |")
|
|
|
|
print("=" * (max_key_len + max_val_len + 7))
|
|
|
|
# 打印 RSA keys
|
|
for key, value in rsa_data.items():
|
|
print(f"\n--- {key} ---")
|
|
print(value)
|
|
|
|
|
|
|
|
# --- 交互式菜单功能 ---
|
|
def create_xdp_ultra_redeem_code():
|
|
"""创建 xdp ultra redeem code"""
|
|
print("\n=== 创建 XDP Ultra Redeem Code ===")
|
|
count = input("请输入要生成的数量: ")
|
|
try:
|
|
count = int(count)
|
|
if count <= 0:
|
|
print("数量必须大于 0")
|
|
return
|
|
print(f"\n开始生成 {count} 个 redeem code...")
|
|
# ultra 购买
|
|
info_id = '677758f12c450457fad77408'
|
|
gen_code(info_id, count, True)
|
|
print(f"\n成功生成 {count} 个 redeem code")
|
|
except ValueError:
|
|
print("错误: 请输入有效的数字")
|
|
except Exception as e:
|
|
print(f"执行出错: {e}")
|
|
|
|
|
|
def batch_bind_product():
|
|
"""批次绑定商品"""
|
|
print("\n=== 批次绑定商品 ===")
|
|
try:
|
|
product_id = int(input("请输入商品ID (product_id): "))
|
|
batch_id = int(input("请输入批次ID (batch_id): "))
|
|
start = int(input("请输入开始流水号 (start): "))
|
|
stop = int(input("请输入结束流水号 (stop): "))
|
|
|
|
print(f"\n开始绑定商品...")
|
|
print(f"商品ID: {product_id}, 批次ID: {batch_id}, Start: {start}, Stop: {stop}")
|
|
|
|
result = bind_product(product_id, batch_id, start, stop)
|
|
print(f"\n绑定成功!")
|
|
print(f"结果: {result}")
|
|
except ValueError as e:
|
|
print(f"输入错误: {e}")
|
|
except Exception as e:
|
|
print(f"执行出错: {e}")
|
|
|
|
|
|
def batch_update_product():
|
|
"""批次更新商品"""
|
|
print("\n=== 批次更新商品 ===")
|
|
try:
|
|
new_product_id = int(input("请输入新商品ID (new_product_id): "))
|
|
batch_id = int(input("请输入批次ID (batch_id): "))
|
|
start = int(input("请输入开始流水号 (start): "))
|
|
stop = int(input("请输入结束流水号 (stop): "))
|
|
|
|
print(f"\n开始更新商品...")
|
|
print(f"新商品ID: {new_product_id}, 批次ID: {batch_id}, Start: {start}, Stop: {stop}")
|
|
|
|
result = update_product(new_product_id, batch_id, start, stop)
|
|
print(f"\n更新成功!")
|
|
print(f"结果: {result}")
|
|
except ValueError as e:
|
|
print(f"输入错误: {e}")
|
|
except Exception as e:
|
|
print(f"执行出错: {e}")
|
|
|
|
|
|
def query_xdp_device_info():
|
|
"""查询 XDP 设备信息"""
|
|
print("\n=== 查询 XDP 设备信息 ===")
|
|
keyword = input("请输入 PSN 或 Passcode (4位): ").strip()
|
|
if not keyword:
|
|
print("错误: 请输入有效的关键字")
|
|
return
|
|
filter_xdp_device_info(keyword)
|
|
|
|
|
|
def show_menu():
|
|
"""显示功能菜单"""
|
|
print("\n" + "="*50)
|
|
print("SSO 交互式脚本")
|
|
print("="*50)
|
|
print("请选择要执行的功能:")
|
|
print("1. 创建 XDP Ultra Redeem Code")
|
|
print("2. 批次绑定商品")
|
|
print("3. 批次更新商品")
|
|
print("4. 查询 XDP 设备信息")
|
|
print("0. 退出")
|
|
print("="*50)
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("--- 开始执行交互式 Django 脚本 ---\n")
|
|
|
|
while True:
|
|
show_menu()
|
|
choice = input("\n请输入功能编号: ").strip()
|
|
|
|
if choice == '1':
|
|
create_xdp_ultra_redeem_code()
|
|
elif choice == '2':
|
|
batch_bind_product()
|
|
elif choice == '3':
|
|
batch_update_product()
|
|
elif choice == '4':
|
|
query_xdp_device_info()
|
|
elif choice == '0':
|
|
print("\n退出脚本...")
|
|
break
|
|
else:
|
|
print("\n无效的选择，请重新输入")
|
|
|
|
# 询问是否继续
|
|
if choice != '0':
|
|
continue_choice = input("\n是否继续执行其他功能? (y/n): ").strip().lower()
|
|
if continue_choice != 'y':
|
|
print("\n退出脚本...")
|
|
break
|
|
|
|
print("\n--- 脚本执行完毕 ---")
|
|
|
|
|
|
# 执行主函数
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
|