180 lines
5.9 KiB
Python
180 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
InfluxDB Token权限检查和生成助手
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
from influxdb_client import InfluxDBClient
|
||
|
||
def load_config():
|
||
"""从default.json加载配置"""
|
||
config_path = os.path.join(os.path.dirname(__file__), "default.json")
|
||
try:
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
return config.get('influx', {})
|
||
except Exception as e:
|
||
print(f"❌ 加载配置文件失败: {e}")
|
||
return {}
|
||
|
||
def check_token_permissions():
|
||
"""检查当前Token权限"""
|
||
config = load_config()
|
||
url = config.get('url', '')
|
||
token = config.get('token', '')
|
||
|
||
if not url or not token:
|
||
print("❌ URL或Token配置缺失")
|
||
return
|
||
|
||
print("🔍 检查Token权限...")
|
||
print(f"URL: {url}")
|
||
print(f"Token: {token[:20]}...")
|
||
|
||
try:
|
||
client = InfluxDBClient(url=url, token=token)
|
||
|
||
# 检查基本连接
|
||
health = client.health()
|
||
print(f"✅ 基本连接: {health.status}")
|
||
|
||
# 检查各种权限
|
||
permissions = {}
|
||
|
||
# 1. 检查组织权限
|
||
try:
|
||
orgs = client.organizations_api().find_organizations()
|
||
permissions['organizations'] = True
|
||
print(f"✅ 组织权限: 可访问 {len(orgs)} 个组织")
|
||
for org in orgs:
|
||
print(f" - {org.name} (ID: {org.id})")
|
||
except Exception as e:
|
||
permissions['organizations'] = False
|
||
print(f"❌ 组织权限: {e}")
|
||
|
||
# 2. 检查bucket权限
|
||
try:
|
||
buckets = client.buckets_api().find_buckets()
|
||
permissions['buckets'] = True
|
||
bucket_count = len(buckets.buckets) if buckets.buckets else 0
|
||
print(f"✅ Bucket权限: 可访问 {bucket_count} 个bucket")
|
||
if buckets.buckets:
|
||
for bucket in buckets.buckets[:5]: # 只显示前5个
|
||
print(f" - {bucket.name} (ID: {bucket.id})")
|
||
except Exception as e:
|
||
permissions['buckets'] = False
|
||
print(f"❌ Bucket权限: {e}")
|
||
|
||
# 3. 检查写入权限（尝试写入一个测试点）
|
||
try:
|
||
write_api = client.write_api()
|
||
# 这里不实际写入，只是检查API是否可用
|
||
permissions['write'] = True
|
||
print("✅ 写入API: 可访问")
|
||
except Exception as e:
|
||
permissions['write'] = False
|
||
print(f"❌ 写入权限: {e}")
|
||
|
||
# 4. 检查查询权限
|
||
try:
|
||
query_api = client.query_api()
|
||
permissions['query'] = True
|
||
print("✅ 查询API: 可访问")
|
||
except Exception as e:
|
||
permissions['query'] = False
|
||
print(f"❌ 查询权限: {e}")
|
||
|
||
client.close()
|
||
|
||
# 总结
|
||
print("\n📊 权限总结:")
|
||
all_good = all(permissions.values())
|
||
if all_good:
|
||
print("🎉 所有权限正常!")
|
||
else:
|
||
print("⚠️ 存在权限问题:")
|
||
for perm, status in permissions.items():
|
||
status_icon = "✅" if status else "❌"
|
||
print(f" {status_icon} {perm}")
|
||
|
||
return permissions
|
||
|
||
except Exception as e:
|
||
print(f"❌ 连接失败: {e}")
|
||
return None
|
||
|
||
def generate_token_instructions():
|
||
"""生成Token的说明"""
|
||
print("\n📝 如何生成新的Token:")
|
||
print("1. 打开InfluxDB Web界面: http://127.0.0.1:8086")
|
||
print("2. 登录到InfluxDB")
|
||
print("3. 点击左侧菜单 'Data' -> 'API Tokens'")
|
||
print("4. 点击 'Generate API Token' -> 'All Access API Token'")
|
||
print("5. 输入描述，如 'DocxCreator Token'")
|
||
print("6. 点击 'Save'")
|
||
print("7. 复制生成的Token")
|
||
print("8. 更新default.json中的token字段")
|
||
|
||
print("\n🔧 或者使用CLI生成Token:")
|
||
print("influx auth create --all-access --description 'DocxCreator Token'")
|
||
|
||
def suggest_minimal_permissions():
|
||
"""建议最小权限设置"""
|
||
config = load_config()
|
||
org = config.get('org', 'MEASCON')
|
||
bucket = config.get('bucket', 'test_bucket')
|
||
|
||
print(f"\n🎯 最小权限建议 (组织: {org}, Bucket: {bucket}):")
|
||
print("需要以下权限:")
|
||
print("1. 读取权限:")
|
||
print(f" - buckets: read (bucket: {bucket})")
|
||
print(f" - orgs: read (org: {org})")
|
||
print("2. 写入权限:")
|
||
print(f" - buckets: write (bucket: {bucket})")
|
||
print("3. 查询权限:")
|
||
print(f" - buckets: read (bucket: {bucket})")
|
||
|
||
def main():
|
||
print("InfluxDB Token权限检查助手")
|
||
print("=" * 50)
|
||
|
||
config = load_config()
|
||
if not config:
|
||
print("❌ 无法加载配置")
|
||
return
|
||
|
||
# 检查当前Token权限
|
||
permissions = check_token_permissions()
|
||
|
||
if permissions is None:
|
||
print("\n❌ 无法连接到InfluxDB，请检查:")
|
||
print("1. InfluxDB服务是否运行")
|
||
print("2. URL是否正确")
|
||
print("3. Token是否有效")
|
||
generate_token_instructions()
|
||
return
|
||
|
||
# 如果权限不足，提供解决方案
|
||
if not all(permissions.values()):
|
||
print("\n🔧 解决方案:")
|
||
|
||
if not permissions.get('organizations', True) or not permissions.get('buckets', True):
|
||
print("1. Token权限不足，需要生成新Token:")
|
||
generate_token_instructions()
|
||
|
||
suggest_minimal_permissions()
|
||
|
||
print("\n💡 快速解决方案:")
|
||
print("1. 生成 'All Access API Token' (最简单)")
|
||
print("2. 或者创建具有特定权限的Token (更安全)")
|
||
|
||
else:
|
||
print("\n🎉 Token权限正常，可以正常使用!")
|
||
print("\n下一步可以:")
|
||
print("1. 运行 python influx_test_tool.py 测试数据写入")
|
||
print("2. 在程序中测试实验监控功能")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|