95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
测试配置保存功能
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
def check_config():
|
|||
|
|
"""检查当前配置"""
|
|||
|
|
config_path = Path("default.json")
|
|||
|
|
|
|||
|
|
if not config_path.exists():
|
|||
|
|
print("❌ default.json 不存在")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|||
|
|
config = json.load(f)
|
|||
|
|
|
|||
|
|
influx_config = config.get('influx', {})
|
|||
|
|
|
|||
|
|
print("📋 当前InfluxDB配置:")
|
|||
|
|
print(f" URL: {influx_config.get('url', 'N/A')}")
|
|||
|
|
print(f" Org: {influx_config.get('org', 'N/A')}")
|
|||
|
|
print(f" Token: {influx_config.get('token', 'N/A')[:20]}..." if influx_config.get('token') else " Token: N/A")
|
|||
|
|
print(f" Bucket: {influx_config.get('bucket', 'N/A')}")
|
|||
|
|
print(f" Measurement: {influx_config.get('measurement', 'N/A')}")
|
|||
|
|
|
|||
|
|
# 检查必要字段是否存在
|
|||
|
|
required_fields = ['bucket', 'measurement']
|
|||
|
|
missing_fields = []
|
|||
|
|
|
|||
|
|
for field in required_fields:
|
|||
|
|
if field not in influx_config:
|
|||
|
|
missing_fields.append(field)
|
|||
|
|
|
|||
|
|
if missing_fields:
|
|||
|
|
print(f"\n⚠️ 缺少字段: {missing_fields}")
|
|||
|
|
return False
|
|||
|
|
else:
|
|||
|
|
print("\n✅ 所有必要字段都存在")
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ 读取配置失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def add_missing_fields():
|
|||
|
|
"""添加缺少的字段"""
|
|||
|
|
config_path = Path("default.json")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|||
|
|
config = json.load(f)
|
|||
|
|
|
|||
|
|
influx_config = config.get('influx', {})
|
|||
|
|
|
|||
|
|
# 添加缺少的字段
|
|||
|
|
if 'bucket' not in influx_config:
|
|||
|
|
influx_config['bucket'] = 'PCM'
|
|||
|
|
print("➕ 添加 bucket: PCM")
|
|||
|
|
|
|||
|
|
if 'measurement' not in influx_config:
|
|||
|
|
influx_config['measurement'] = 'experiment_status'
|
|||
|
|
print("➕ 添加 measurement: experiment_status")
|
|||
|
|
|
|||
|
|
# 保存配置
|
|||
|
|
with open(config_path, 'w', encoding='utf-8') as f:
|
|||
|
|
json.dump(config, f, indent=2, ensure_ascii=False)
|
|||
|
|
|
|||
|
|
print("✅ 配置已更新并保存")
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ 更新配置失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
print("配置保存测试工具")
|
|||
|
|
print("=" * 30)
|
|||
|
|
|
|||
|
|
print("1️⃣ 检查当前配置...")
|
|||
|
|
if check_config():
|
|||
|
|
print("\n🎉 配置完整，无需修改")
|
|||
|
|
else:
|
|||
|
|
print("\n2️⃣ 添加缺少的字段...")
|
|||
|
|
if add_missing_fields():
|
|||
|
|
print("\n3️⃣ 重新检查配置...")
|
|||
|
|
check_config()
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|