File size: 3,114 Bytes
494c9e4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""API 工具函数"""
import math
import os
import traceback


def round_to_sig_figs(x: float, n: int = 7) -> float:
    """将浮点数舍入为 n 位有效数字。0 或非有限值原样返回。"""
    if x == 0 or not math.isfinite(x):
        return x
    return float(f"{x:.{n}g}")
from functools import wraps
from pathlib import Path
from flask import request, jsonify


def get_demo_directory(create=False):
    """获取 demo 目录路径"""
    from backend.app_context import get_demo_directory as _get_demo_dir
    return _get_demo_dir(create=create)


def handle_api_error(operation_name: str, error: Exception) -> dict:
    """
    统一的 API 错误处理
    
    Args:
        operation_name: 操作名称(如 'Save failed'、'Delete failed')
        error: 异常对象
        
    Returns:
        标准错误响应字典
    """
    error_msg = f'{operation_name}: {str(error)}'
    print(f"❌ {error_msg}")
    traceback.print_exc()
    return {
        'success': False,
        'message': error_msg
    }


def handle_api_success(result: dict, operation_name: str = None) -> dict:
    """
    处理 API 成功响应,打印日志
    
    Args:
        result: 操作结果字典
        operation_name: 可选的操作名称,用于日志
        
    Returns:
        结果字典
    """
    if result.get('success'):
        if operation_name:
            print(f"✓ {operation_name}")
        elif result.get('message'):
            print(f"✓ {result.get('message')}")
    else:
        message = result.get('message', 'Operation failed')
        print(f"❌ {message}")
    return result


def get_admin_token() -> str:
    """
    获取管理员token(从环境变量读取)
    
    Returns:
        管理员token字符串,如果未设置则返回None
    """
    return os.environ.get('INFORADAR_ADMIN_TOKEN')


def validate_admin_token(request_token: str) -> tuple[bool, str]:
    """
    验证管理员token是否有效
    
    Args:
        request_token: 要验证的token
    
    Returns:
        (是否有效, 错误信息)
    """
    admin_token = get_admin_token()
    
    # 如果未配置INFORADAR_ADMIN_TOKEN,返回未启用状态
    if admin_token is None:
        return False, 'Admin features are not enabled'
    
    # 验证token
    if request_token == admin_token:
        return True, ''
    else:
        return False, 'Invalid admin token'


def require_admin(f):
    """
    装饰器:要求管理员权限才能访问的API
    
    检查请求头中的 X-Admin-Token 是否与配置的 INFORADAR_ADMIN_TOKEN 匹配
    如果未配置 INFORADAR_ADMIN_TOKEN,视为全是普通用户,拒绝所有写操作
    """
    @wraps(f)
    def wrapper(*args, **kwargs):
        request_token = request.headers.get('X-Admin-Token')
        is_valid, error_message = validate_admin_token(request_token)
        
        if not is_valid:
            return {
                'success': False,
                'message': 'Admin permission required'
            }, 403
        
        return f(*args, **kwargs)
    return wrapper