if isinstance(json_obj, dict): new_json_obj = {k: remove_keys(v, keys) for k, v in json_obj.items() if k not in keys} # 将新的json_obj转换为map对象new_map_obj new_map_obj = dict(new_json_obj.items()) # map对象按照key的ASCLL码小到大排序生成k=v键值对，k=v键值对之间以"&"连接输出result sorted_m = sorted(new_json_obj.items(), key=lambda x: x[0]) result = "".join([f"{k}={v}&" for k, v in sorted_m])[:-2] print("复制去加签字典：" + result)

import json from urllib.parse import quote_plus def json_to_url_params(json_obj): """将复杂的 JSON 对象转换为 URL 参数字符串""" if isinstance(json_obj, dict): # 对字典类型的值进行处理 items = json_obj.items() elif isinstance(json_obj, list): # 对列表类型的值进行处理 items = enumerate(json_obj) else: # 对其他类型的值直接返回字符串形式 return str(json_obj) # 递归处理每个键值对，并按 ASCII 码排序 params = [] for key, value in sorted(items, key=lambda x: str(x[0])): if isinstance(value, (dict, list)): # 对嵌套的字典或列表类型的值进行递归处理 sub_params = json_to_url_params(value) params.extend([(f"{key}[{sub_key}]", sub_value) for sub_key, sub_value in sub_params]) else: # 对其他类型的值进行 URL 编码后拼接成键值对 params.append((str(key), quote_plus(str(value)))) return params # 示例 JSON 字符串 json_str = '{"a": {"b": {"c": 1}, "d": [2, 3]}, "e": "f"}' # 将 JSON 字符串转换为 Python 对象 json_obj = json.loads(json_str) # 将 Python 对象转换为 URL 参数字符串 params = json_to_url_params(json_obj) # 将 URL 参数字符串拼接成完整的 URL url = "&".join([f"{key}={value}" for key, value in params]) print(url) 上面代码如果json中value存在中文字符串要特殊处理吗

在这个代码中，如果 JSON 中存在中文字符串，会自动被转换为 Unicode 编码，并使用 quote_plus() 函数进行 URL 编码。因此，不需要对中文字符串进行特殊处理。 quote_plus() 函数可以将字符串中的特殊字符编码...

以下是Superset的配置文件，请检查是否启用了Guest Token 功能： #!/usr/bin/env python # -- coding: utf-8 -- """ 优化的Superset混合认证配置 """ import os import sys import logging import pymysql import ssl from urllib.parse import quote_plus from flask_sqlalchemy import SQLAlchemy # ============================================================================ # 基础配置 # ============================================================================ BASE_DIR = os.path.dirname(file) sys.path.append(BASE_DIR) # 环境变量 ENVIRONMENT = os.getenv("SUPERSET_ENV", "development") # production/development IS_PRODUCTION = ENVIRONMENT == "production" IS_DEVELOPMENT = ENVIRONMENT == "development" # 版本和部署信息 SUPERSET_VERSION = os.getenv("SUPERSET_VERSION", "4.1.2") DEPLOYMENT_ID = os.getenv("DEPLOYMENT_ID", "default") print(f"🔧 加载Superset配置 - 环境: {ENVIRONMENT}") # ============================================================================ # 数据库配置（优化版） # ============================================================================ pymysql.install_as_MySQLdb() db = SQLAlchemy() DB_USER = os.getenv("SUPERSET_DB_USER", "superset") DB_PASSWORD = os.getenv("SUPERSET_DB_PASSWORD", "SupersetPassword2025!") DB_HOST = os.getenv("SUPERSET_DB_HOST", "10.18.6.120") DB_PORT = int(os.getenv("SUPERSET_DB_PORT", "3306")) DB_NAME = os.getenv("SUPERSET_DB_NAME", "superset") SQLALCHEMY_DATABASE_URI = ( f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}" "?charset=utf8mb4" ) # 优化的数据库引擎配置 SQLALCHEMY_ENGINE_OPTIONS = { "pool_pre_ping": True, "pool_recycle": 3600, "pool_timeout": 30, "max_overflow": 20, "pool_size": 10, "echo": False, "isolation_level": "READ_COMMITTED", "connect_args": { "connect_timeout": 10, "read_timeout": 60, "write_timeout": 60, "charset": "utf8mb4", # 移除了collation参数 "autocommit": False, "sql_mode": "STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO", } } # SQLAlchemy配置优化 SQLALCHEMY_TRACK_MODIFICATIONS = False SQLALCHEMY_RECORD_QUERIES = False SQLALCHEMY_ECHO = False # 数据库健康检查 DATABASE_HEALTH_CHECK_ENABLED = True DATABASE_HEALTH_CHECK_INTERVAL = 30 # ============================================================================ # SQLLab专用配置（修复前端错误） # ============================================================================ # SQLLab基础配置 ENABLE_SQLLAB = True SQLLAB_BACKEND_PERSISTENCE = True SQLLAB_TIMEOUT = 300 SQL_MAX_ROW = 10000 SQLLAB_DEFAULT_DBID = None SQLLAB_CTAS_NO_LIMIT = True SQLLAB_QUERY_COST_ESTIMATES_ENABLED = False # 关闭查询成本估算 SQLLAB_ASYNC_TIME_LIMIT_SEC = 600 # 防止SQLLab错误的配置 PREVENT_UNSAFE_DB_CONNECTIONS = False SQLLAB_VALIDATION_TIMEOUT = 10 ENABLE_TEMPLATE_PROCESSING = True # 查询结果配置 SUPERSET_WEBSERVER_TIMEOUT = 300 SUPERSET_WORKERS = 1 if IS_DEVELOPMENT else 4 # ============================================================================ # 安全配置（多层防护） # ============================================================================ SECRET_KEY = os.environ.get( "SUPERSET_SECRET_KEY", "FPwbFnYKL6wQTD0vtQfGBw7Y530FUfufsnHwXQTLlrrn8koVcctkMwiK", ) # 安全检查 if len(SECRET_KEY) < 32: raise ValueError("SECRET_KEY长度必须至少32位") if IS_DEVELOPMENT: logging.debug(f"SECRET_KEY已设置: {'' len(SECRET_KEY)}") # 禁用安全警告弹框 SECURITY_WARNING_BANNER = False # 或者设置为空字符串 SECURITY_WARNING_MESSAGE = "" # ============================================================================ # Redis和异步查询检查 # ============================================================================ # 检查Redis可用性 REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = int(os.getenv("REDIS_PORT", "6379")) REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "minth@888") # Redis密码 REDIS_DB = int(os.getenv("REDIS_DB", "0")) # URL编码密码以处理特殊字符 REDIS_PASSWORD_ENCODED = quote_plus(REDIS_PASSWORD) if REDIS_PASSWORD else "" # 构建Redis连接URL if REDIS_PASSWORD: REDIS_URL = f"redis://:{REDIS_PASSWORD_ENCODED}@{REDIS_HOST}:{REDIS_PORT}" else: REDIS_URL = f"redis://{REDIS_HOST}:{REDIS_PORT}" # 检查Redis可用性 REDIS_AVAILABLE = False try: import redis # 创建Redis连接（带密码） r = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD if REDIS_PASSWORD else None, socket_timeout=5, socket_connect_timeout=5 ) # 测试连接 r.ping() REDIS_AVAILABLE = True print(f"✅ Redis连接成功: {REDIS_HOST}:{REDIS_PORT} (已认证)") # 测试基本操作 test_key = "superset_test" r.set(test_key, "test_value", ex=10) # 10秒过期 if r.get(test_key): print("✅ Redis读写测试成功") r.delete(test_key) except Exception as e: print(f"⚠️ Redis连接失败: {e}") REDIS_AVAILABLE = False # ============================================================================ # 特性标志 # ============================================================================ FEATURE_FLAGS = { # SQLLab核心功能 "ENABLE_SQLLAB": True, "SQLLAB_BACKEND_PERSISTENCE": True, "ESTIMATE_QUERY_COST": False, # 关闭可能导致ROLLBACK的功能 "QUERY_COST_FORMATTERS_BY_ENGINE": {}, "RESULTS_BACKEND_USE_MSGPACK": True, # 🔧 强制启用msgpack # 权限和安全 "DASHBOARD_RBAC": True, "ENABLE_EXPLORE_JSON_CSRF_PROTECTION": True, "ENABLE_TEMPLATE_PROCESSING": True, "ROW_LEVEL_SECURITY": True, # 行级安全 # 嵌入功能 "EMBEDDED_SUPERSET": True, "ALLOW_DASHBOARD_EMBEDDING": True, "EMBEDDED_IN_FRAME": True, # 过滤器和交互 "DASHBOARD_NATIVE_FILTERS": True, "DASHBOARD_CROSS_FILTERS": True, "ENABLE_FILTER_BOX_MIGRATION": True, "DASHBOARD_FILTERS": True, # 异步查询禁用 "GLOBAL_ASYNC_QUERIES": False , "ASYNC_QUERIES": False, "SCHEDULED_QUERIES": False, # 高级功能暂时不设置 "ENABLE_JAVASCRIPT_CONTROLS": True, "DYNAMIC_PLUGINS": False, "THUMBNAILS": REDIS_AVAILABLE, "SCREENSHOTS": REDIS_AVAILABLE, # 开发和调试（仅开发环境） "ENABLE_REACT_CRUD_VIEWS": IS_DEVELOPMENT, "ENABLE_BROAD_ACTIVITY_ACCESS": IS_DEVELOPMENT, } # ============================================================================ # 禁用异步查询配置·RESULTS_BACKEND # ============================================================================ RESULTS_BACKEND = None CELERY_CONFIG = None # ============================================================================ # 结果处理修复配置 # ============================================================================ RESULTS_BACKEND_USE_MSGPACK = True import json class SupersetJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, set): return list(obj) if hasattr(obj, 'isoformat'): return obj.isoformat() if hasattr(obj, 'dict'): return obj.dict return super().default(obj) JSON_DEFAULT = SupersetJSONEncoder().encode SQLLAB_RESULTS_BACKEND_PERSISTENCE = True SQLLAB_TIMEOUT = 300 SUPERSET_SQLLAB_TIMEOUT = 300 print("🔧 结果处理配置已修复") # ============================================================================ # LDAP认证配置（增强版） # ============================================================================ from flask_appbuilder.security.manager import AUTH_LDAP AUTH_TYPE = AUTH_LDAP # LDAP服务器配置 AUTH_LDAP_SERVER = "ldap://10.18.2.50:389" AUTH_LDAP_USE_TLS = False # 可根据服务器能力调整 # 用户名和搜索配置 AUTH_LDAP_USERNAME_FORMAT = "%(username)[email protected]" AUTH_LDAP_SEARCH = "dc=minth,dc=intra" AUTH_LDAP_SEARCH_FILTER = "(sAMAccountName={username})" # 用户属性映射 AUTH_LDAP_UID_FIELD = "sAMAccountName" AUTH_LDAP_FIRSTNAME_FIELD = "givenName" AUTH_LDAP_LASTNAME_FIELD = "sn" AUTH_LDAP_EMAIL_FIELD = "mail" # LDAP行为配置 AUTH_LDAP_ALLOW_SELF_SIGNED = True AUTH_LDAP_ALWAYS_SEARCH = True # 用户管理 AUTH_USER_REGISTRATION = False AUTH_USER_REGISTRATION_ROLE = "Public" AUTH_ROLES_SYNC_AT_LOGIN = True # 角色映射 AUTH_ROLES_MAPPING = { "cn=SupersetAdmins,ou=Groups,dc=minth,dc=intra": ["Admin"], "cn=DataAnalysts,ou=Groups,dc=minth,dc=intra": ["Alpha"], "cn=DataViewers,ou=Groups,dc=minth,dc=intra": ["Gamma"], "cn=ReportViewers,ou=Groups,dc=minth,dc=intra": ["Public"], } AUTH_LDAP_GROUP_FIELD = "memberOf" # ============================================================================ # 混合安全管理器 # ============================================================================ try: from security.hybrid_security_manager import HybridSecurityManager CUSTOM_SECURITY_MANAGER = HybridSecurityManager # 指定数据库认证用户（与HybridSecurityManager中的DB_AUTH_USERS对应） DB_AUTH_USERS = ['admin','superset'] # LDAP连接池和缓存配置 # AUTH_LDAP_POOL_SIZE = 10 # AUTH_LDAP_POOL_RETRY_MAX = 3 # AUTH_LDAP_POOL_RETRY_DELAY = 30 # AUTH_LDAP_CACHE_ENABLED = True # AUTH_LDAP_CACHE_TIMEOUT = 300 # 认证策略配置 AUTH_STRATEGY = { 'db_first_users': DB_AUTH_USERS, 'ldap_fallback_enabled': True, 'cache_enabled': REDIS_AVAILABLE, # 只有Redis可用时才启用缓存 'cache_timeout': 300, 'max_login_attempts': 5, 'lockout_duration': 900, # 15分钟 } print("✅ 混合安全管理器已启用") except ImportError as e: print(f"⚠️ 混合安全管理器导入失败，使用标准LDAP认证: {e}") CUSTOM_SECURITY_MANAGER = None # ============================================================================ # Guest Token配置（安全优化版） # ============================================================================ GUEST_TOKEN_JWT_SECRET = SECRET_KEY GUEST_TOKEN_JWT_ALGO = "HS256" GUEST_TOKEN_JWT_EXP_DELTA_SECONDS = 3600 GUEST_TOKEN_HEADER_NAME = "X-GuestToken" # 强制设置AUD验证 SUPERSET_BASE_URL = os.getenv("SUPERSET_BASE_URL", None) GUEST_TOKEN_JWT_AUD = SUPERSET_BASE_URL # Guest Token增强配置 GUEST_TOKEN_AUTO_REFRESH = True GUEST_TOKEN_REFRESH_THRESHOLD = 300 # 5分钟内过期自动刷新 GUEST_TOKEN_MAX_CONCURRENT = 100 # 最大并发Token数 # ============================================================================ # 缓存配置（修复版） # ============================================================================ if REDIS_AVAILABLE: # 生产环境Redis缓存 CACHE_CONFIG = { 'CACHE_TYPE': 'redis', 'CACHE_REDIS_HOST': REDIS_HOST, 'CACHE_REDIS_PORT': REDIS_PORT, 'CACHE_REDIS_DB': REDIS_DB, # 使用DB 0作为缓存 'CACHE_DEFAULT_TIMEOUT': 300, 'CACHE_KEY_PREFIX': f'superset_{DEPLOYMENT_ID}_', } # 如果有密码，添加密码配置 if REDIS_PASSWORD: CACHE_CONFIG['CACHE_REDIS_PASSWORD'] = REDIS_PASSWORD CACHE_CONFIG['CACHE_REDIS_URL'] = f"{REDIS_URL}/{REDIS_DB}" print(f"✅ Redis缓存配置: {REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}") else: # 开发环境或Redis不可用时使用简单缓存 CACHE_CONFIG = { 'CACHE_TYPE': 'simple', 'CACHE_DEFAULT_TIMEOUT': 60, } print("ℹ️ 使用简单缓存") # ============================================================================ # 安全头配置（生产环境优化） # ============================================================================ # 消除CSP警告 CONTENT_SECURITY_POLICY_WARNING = False if IS_PRODUCTION: # 生产环境：最严格安全配置 TRUSTED_DOMAINS = [d.strip() for d in os.getenv("TRUSTED_DOMAINS", "").split(",") if d.strip()] HTTP_HEADERS = { "X-Frame-Options": "SAMEORIGIN", # 更安全的默认值 "X-XSS-Protection": "1; mode=block", "X-Content-Type-Options": "nosniff", "Strict-Transport-Security": "max-age=31536000; includeSubDomains; preload", "Referrer-Policy": "strict-origin-when-cross-origin", "Permissions-Policy": "geolocation=(), microphone=(), camera=()", } # 如果有信任域名，则允许特定域名 if TRUSTED_DOMAINS: HTTP_HEADERS["X-Frame-Options"] = f"ALLOW-FROM {TRUSTED_DOMAINS[0]}" TALISMAN_ENABLED = True TALISMAN_CONFIG = { 'content_security_policy': { 'default-src': "'self'", 'script-src': "'self' 'unsafe-inline' 'unsafe-eval'", 'style-src': "'self' 'unsafe-inline'", 'img-src': "'self' data: blob: https:", 'font-src': "'self' data:", 'frame-ancestors': TRUSTED_DOMAINS or ["'self'"], 'connect-src': "'self'", }, 'force_https': os.getenv("FORCE_HTTPS", "false").lower() == "true", 'strict_transport_security': True, 'strict_transport_security_max_age': 31536000, } # CORS严格配置 CORS_ORIGIN_ALLOW_ALL = False CORS_ORIGIN_WHITELIST = TRUSTED_DOMAINS ENABLE_CORS = len(TRUSTED_DOMAINS) > 0 else: # 开发环境：宽松配置 HTTP_HEADERS = { "X-Frame-Options": "", "X-XSS-Protection": "1; mode=block", } TALISMAN_ENABLED = False CORS_ORIGIN_ALLOW_ALL = True ENABLE_CORS = True # ============================================================================ # 日志配置（生产级） # ============================================================================ LOG_DIR = os.getenv("SUPERSET_LOG_DIR", "/app/superset/logs") os.makedirs(LOG_DIR, exist_ok=True) LOG_LEVEL = "INFO" if IS_PRODUCTION else "DEBUG" LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(funcName)s() - %(message)s" LOGGING_CONFIG = { "version": 1, "disable_existing_loggers": False, "formatters": { "detailed": { "format": LOG_FORMAT, "datefmt": "%Y-%m-%d %H:%M:%S", }, "simple": { "format": "%(levelname)s - %(message)s", }, "json": { "format": "%(asctime)s %(name)s %(levelname)s %(message)s", }, }, "handlers": { "console": { "level": LOG_LEVEL, "class": "logging.StreamHandler", "formatter": "simple" if IS_DEVELOPMENT else "json", }, "app_file": { "level": "INFO", "class": "logging.handlers.RotatingFileHandler", "filename": f"{LOG_DIR}/superset.log", "maxBytes": 10485760, # 20MB "backupCount": 5, "formatter": "detailed", }, "auth_file": { "level": "INFO", "class": "logging.handlers.RotatingFileHandler", "filename": f"{LOG_DIR}/auth.log", "maxBytes": 10485760, # 10MB "backupCount": 5, "formatter": "detailed", }, "error_file": { "level": "ERROR", "class": "logging.handlers.RotatingFileHandler", "filename": f"{LOG_DIR}/error.log", "maxBytes": 10485760, "backupCount": 10, "formatter": "detailed", }, # "security_file": { # "level": "WARNING", # "class": "logging.handlers.RotatingFileHandler", # "filename": f"{LOG_DIR}/security.log", # "maxBytes": 10485760, # "backupCount": 10, # "formatter": "detailed", # }, }, "loggers": { "superset": { "level": LOG_LEVEL, "handlers": ["app_file", "console"], "propagate": False, }, "flask_appbuilder.security": { "level": "INFO", "handlers": ["auth_file", "console"], "propagate": False, }, "security_manager": { "level": LOG_LEVEL, "handlers": ["auth_file", "console"], "propagate": False, }, "werkzeug": { "level": "WARNING", "handlers": ["console"], "propagate": False, }, }, "root": { "level": LOG_LEVEL, "handlers": ["console"], }, } # ============================================================================ # 业务配置 # ============================================================================ # 角色设置 GUEST_ROLE_NAME = "Public" PUBLIC_ROLE_LIKE = "Gamma" PUBLIC_ROLE_LIKE_GAMMA = True # 国际化 BABEL_DEFAULT_LOCALE = "zh" BABEL_DEFAULT_FOLDER = "babel/translations" LANGUAGES = { "en": {"flag": "us", "name": "English"}, "zh": {"flag": "cn", "name": "Chinese"}, } # 系统设置 SUPERSET_DEFAULT_TIMEZONE = "Asia/Shanghai" SUPERSET_WEBSERVER_TIMEOUT = 120 if IS_PRODUCTION else 60 # 基础功能启用 ENABLE_CSRF = True ENABLE_SWAGGER = not IS_PRODUCTION # 生产环境关闭Swagger ENABLE_GUEST_TOKEN = True EMBEDDED_SUPERSET = True ALLOW_DASHBOARD_EMBEDDING = True # ============================================================================ # 环境特定配置 # ============================================================================ if IS_PRODUCTION: # 生产环境优化 WTF_CSRF_TIME_LIMIT = 7200 SEND_FILE_MAX_AGE_DEFAULT = 31536000 # 数据源连接超时 SQL_MAX_ROW = 100000 SQLLAB_TIMEOUT = 300 SUPERSET_WORKERS = int(os.getenv("SUPERSET_WORKERS", "4")) else: # 开发环境配置 WTF_CSRF_TIME_LIMIT = None SQL_MAX_ROW = 10000 SQLLAB_TIMEOUT = 60 # ============================================================================ # 启动验证 # ============================================================================ def validate_config(): """配置验证""" errors = [] # 必需的环境变量检查 required_env_vars = ["SUPERSET_SECRET_KEY"] if IS_PRODUCTION: required_env_vars.extend(["SUPERSET_BASE_URL", "TRUSTED_DOMAINS"]) for var in required_env_vars: if not os.getenv(var): errors.append(f"缺少必需的环境变量: {var}") # 安全配置检查 if IS_PRODUCTION and CORS_ORIGIN_ALLOW_ALL: errors.append("生产环境不应允许所有来源的CORS请求") if errors: raise ValueError(f"配置验证失败:\n" + "\n".join(f" - {error}" for error in errors)) # 执行配置验证 try: validate_config() print(f"🚀 Superset配置加载完成 - 环境: {ENVIRONMENT}") print(f" 数据库: {DB_HOST}:{DB_PORT}/{DB_NAME}") print(f" 认证: LDAP混合模式") print(f" 缓存: {'Redis' if IS_PRODUCTION else 'Simple'}") print(f" 日志级别: {LOG_LEVEL}") except ValueError as e: print(f"❌ 配置验证失败: {e}") if IS_PRODUCTION: raise # 生产环境严格要求 else: print("⚠️ 开发环境忽略配置错误")

### 检查 Superset 配置文件中的 Guest Token 相关设置在 Superset 中启用 Guest Token 功能需要确保配置文件中包含特定的参数，并且这些参数被正确设置。以下是关于如何确认 GUEST_TOKEN_JWT_SECRET 和 GUEST_...

def save_vehicle_parquet_partitioned(self, result_dict_dataset, max_rows_per_file=500000): """ 优化后的保存方法： 1. 使用分块写入避免内存溢出 2. 自动拆分大文件为多个part 3. 使用二进制流式写入 """ print('------------------开始保存车辆数据------------------') output_root = os.path.join(self.output_folder, 'partitioned_parquet') for vehicle_key, vehicle_data in result_dict_dataset.items(): # ================== 准备数据 ================== # 序列化元数据 meta_bytes = json.dumps( vehicle_data['vehicle_parameters'], ensure_ascii=False, default=self._json_serializer ).encode('utf-8') # 准备时间序列数据 df_data = pd.DataFrame(vehicle_data['data']) # ================== 拆分数据块 ================== total_rows = len(df_data) num_files = max(1, total_rows // max_rows_per_file) # 按时间排序确保分块连续性 df_data.sort_values('数据采集时间', inplace=True) # ================== 分块写入 ================== schema = pa.schema([ ('metadata', pa.large_binary()), ('timestamp', pa.timestamp('ms')), ('data', pa.large_binary()) ]) partition_path = os.path.join(output_root, f"vehicle_key={vehicle_key}") os.makedirs(partition_path, exist_ok=True) for i in range(num_files): start_idx = i * max_rows_per_file end_idx = (i + 1) * max_rows_per_file chunk = df_data.iloc[start_idx:end_idx] # 转换为PyArrow格式 timestamps = pd.to_datetime(chunk['数据采集时间']).view('int64') // 10 ** 6 # 转毫秒时间戳 data_bytes = [ json.dumps(row.to_dict(), ensure_ascii=False).encode('utf-8') for _, row in chunk.drop(columns=['数据采集时间']).iterrows() ] # 构建RecordBatch batch = pa.RecordBatch.from_arrays([ pa.array([meta_bytes] * len(chunk), type=pa.large_binary()), pa.array(timestamps, type=pa.timestamp('ms')), pa.array(data_bytes, type=pa.large_binary()) ], schema=schema) # 写入文件（每个分块单独文件） file_path = os.path.join(partition_path, f"part-{i:04d}.parquet") with pq.ParquetWriter( file_path, schema, compression='zstd', version='2.6', data_page_size=1024 * 1024 # 1MB per page ) as writer: writer.write_batch(batch) print( f"车辆 {vehicle_key} 第{i + 1}/{num_files}个分块已保存，大小: {os.path.getsize(file_path) / 1024 / 1024:.2f}MB") print(f"车辆 {vehicle_key} 的所有分块数据已保存到 {partition_path}") @staticmethod def _json_serializer(obj): """增强型序列化处理器""" if isinstance(obj, (np.generic, np.ndarray)): return obj.tolist() if isinstance(obj, pd.Timestamp): return obj.isoformat() if isinstance(obj, bytes): return obj.decode('utf-8', errors='replace') raise TypeError(f"无法序列化类型 {type(obj)}") def run(self): # 逐个处理车辆数据，每处理完一辆车就保存一次 for i, folder in enumerate(self.vehicle_folder_list): if i > 0: continue dict_data = self.process_data(folder) vehicle_key = self.vehicle_keys[i] # 构造一个包含单辆车数据的字典，符合 save_vehicle_parquet_partitioned() 的输入格式 result_dict_dataset = {vehicle_key: dict_data} # 这里单独保存每辆车的数据 print("车辆参数：", dict_data.get("vehicle_parameters")) self.save_vehicle_parquet_partitioned(result_dict_dataset)需求：对上面代码保存的车辆数据，给出代码将车辆数据加载为dict

sample_key = list(loaded_data.keys())[0] print("车辆参数示例：", loaded_data[sample_key]['vehicle_parameters']) print("数据记录示例：", loaded_data[sample_key]['data'][0]) 方法特点说明： 1. **...

Python内置库builtin秘籍全解：从函数到自定义异常的实战手册（详解）

# 1. Python内置库__builtin__概述 Python是一种编程语言，它包含了大量的内置库，而__builtin__就是其中之一。__builtin__库提供了一系列内置函数和异常类型，是Python解释器本身的一部分。...

【JSON_XML 数据处理秘籍】：DAHUA HTTP API 高级技巧解析

![【JSON_XML 数据处理秘籍】：DAHUA HTTP API 高级技巧解析]... JSON与XML数据格式概述数据在现代信息技术领域中占有举足轻重的地位，而数据格式的选

揭秘Python内置库builtin：提升代码效率与对象管理的20个技巧

!...# 1. Python内置库__builtin__概述 Python的__builtin__模块是一个特殊的内置库，它包含了Python解释器中可以直接使用的所有内置函数、类型、异常和变量。它是Python动态语言特性的根基，允许我们在不导入任何外部...

''' Hints: Flag在环境变量中 ''' from typing import Optional import pydash import bottle __forbidden_path='annotations', 'call', 'class', 'closure', 'code', 'defaults', 'delattr', 'dict', 'dir', 'doc', 'eq', 'format', 'ge', 'get', 'getattribute', 'gt', 'hash', 'init', 'init_subclass', 'kwdefaults', 'le', 'lt', 'module', 'name', 'ne', 'new', 'qualname', 'reduce', 'reduce_ex', 'repr', 'setattr', 'sizeof', 'str', 'subclasshook', 'wrapped', "Optional","func","render", forbidden_name= "bottle" forbidden_name.extend(dir(globals()"builtins")) def setval(name:str, path:str, value:str)-> Optionalbool: if name.find("")>=0: return False for word in __forbidden_name: if name==word: return False for word in forbidden_path__: if path.find(word)>=0: return False obj=globals()name try: pydash.set_(obj,path,value) except: return False return True @bottle.post('/setValue') def set_value(): name = bottle.request.query.get('name') path=bottle.request.json.get('path') if not isinstance(path,str): return "no" if len(name)>6 or len(path)>32: return "no" value=bottle.request.json.get('value') return "yes" if setval(name, path, value) else "no" @bottle.get('/render') def render_template(): path=bottle.request.query.get('path') if path.find("{")>=0 or path.find("}")>=0 or path.find(".")>=0: return "Hacker" return bottle.template(path) bottle.run(host='0.0.0.0', port=8000)

dir(globals()["__builtins__"])会返回所有内置函数和异常的名称，比如'str', 'int', 'len', 'getattr', 'setattr', 'dict'等等。所以，如果setval函数的name是'setval'，长度为6，但可能在__forbidden_name__中吗？...

def chat_default(request): if request.method in ["POST", "GET"]: msg = {"code": normal_code,"msg": mes.normal_code, "data": {}} req_dict = request.session.get("req_dict") req_dict.update({"isdefault":"是"}) data=chat.getbyparams(chat, chat, req_dict) if len(data)>0: msg['data'] = data[0] else: msg['data'] = {} return JsonResponse(msg) def chat_page(request): ''' ''' if request.method in ["POST", "GET"]: msg = {"code": normal_code, "msg": mes.normal_code, "data":{"currPage":1,"totalPage":1,"total":1,"pageSize":10,"list":[]}} req_dict = request.session.get("req_dict") #获取全部列名 columns= chat.getallcolumn( chat, chat) #当前登录用户所在表 tablename = request.session.get("tablename") #authColumn=list(authTables.keys())[0] #authTable=authTables.get(authColumn) # if authTable==tablename: #params = request.session.get("params") #req_dict[authColumn]=params.get(authColumn) '''authSeparate此属性为真，params添加userid，后台只查询个人数据''' try: authSeparate=chat.authSeparate except: authSeparate=None if authSeparate=="是": tablename=request.session.get("tablename") if tablename!="users" and 'userid' in columns: try: req_dict['userid']=request.session.get("params").get("id") except: pass #当项目属性hasMessage为”是”，生成系统自动生成留言板的表messages，同时该表的表属性hasMessage也被设置为”是”,字段包括userid（用户id），username(用户名)，content（留言内容），reply（回复） #接口page需要区分权限，普通用户查看自己的留言和回复记录，管理员查看所有的留言和回复记录 try: hasMessage=chat.hasMessage except: hasMessage=None if hasMessage=="是": tablename=request.session.get("tablename") if tablename!="users": req_dict["userid"]=request.session.get("params").get("id") # 判断当前表的表属性isAdmin,为真则是管理员表 # 当表属性isAdmin=”是”,刷出来的用户表也是管理员，即page和list可以查看所有人的考试记录(同时应用于其他表) isAdmin = None allModels = apps.get_app_config('main').get_models() for m in allModels: if m.tablename==tablename: try: isAdmin = m.isAdmin except: isAdmin = None break # 当前表也是有管理员权限的表 if isAdmin == "是" and 'chat' != 'forum': if req_dict.get("userid") and 'chat' != 'chat': del req_dict["userid"] else: #非管理员权限的表,判断当前表字段名是否有userid if tablename!="users" and 'chat'[:7]!='discuss'and "userid" in chat.getallcolumn(chat,chat): req_dict["userid"] = request.session.get("params").get("id") #当列属性authTable有值(某个用户表)[该列的列名必须和该用户表的登陆字段名一致]，则对应的表有个隐藏属性authTable为”是”，那么该用户查看该表信息时，只能查看自己的 try: authTables=chat.authTables except: authTables=None if authTables!=None and authTables!={}: try: del req_dict['userid'] # tablename=request.session.get("tablename") # if tablename=="users": # del req_dict['userid'] except: pass for authColumn,authTable in authTables.items(): if authTable==tablename: params = request.session.get("params") req_dict[authColumn]=params.get(authColumn) username=params.get(authColumn) break q = Q() msg['data']['list'], msg['data']['currPage'], msg['data']['totalPage'], msg['data']['total'], \ msg['data']['pageSize'] =chat.page(chat, chat, req_dict, request, q) return JsonResponse(msg)优化代码

if auth_table_map is None or not isinstance(auth_table_map, dict): return current_tablename = session.get("tablename") if current_tablename in auth_table_map.values(): user_params = session....

顯示結果是 { "match_id": "1904138", "home_team": {}, "away_team": {}, "source": "https://live.titan007.com/detail/1904138.htm" }

return {k: remove_empty_fields(v) for k, v in obj.items() if v not in [None, ""]} return obj clean_data = remove_empty_fields(data) print(json.dumps(clean_data)) # 输出: {"name": "Alice"} ...

UnauthorizedError Traceback (most recent call last) Cell In[162], line 11 9 # 获取上个月的天气情况 10 mgr = owm.weather_manager() ---> 11 observation = mgr.weather_at_place('上海') # 这里以北京为例 12 date_obj = datetime.datetime(last_month.year, last_month.month, 1) 13 one_call = mgr.one_call(lat=observation.weather.location.lat, lon=observation.weather.location.lon, dt=date_obj.timestamp(), exclude='current,minutely,hourly,alerts') File ~/opt/anaconda3/lib/python3.9/site-packages/pyowm/weatherapi25/weather_manager.py:53, in WeatherManager.weather_at_place(self, name) 51 assert isinstance(name, str), "Value must be a string" 52 params = {'q': name} ---> 53 _, json_data = self.http_client.get_json(OBSERVATION_URI, params=params) 54 return observation.Observation.from_dict(json_data) File ~/opt/anaconda3/lib/python3.9/site-packages/pyowm/commons/http_client.py:158, in HttpClient.get_json(self, path, params, headers) 156 except requests.exceptions.Timeout: 157 raise exceptions.TimeoutError('API call timeouted') --> 158 HttpClient.check_status_code(resp.status_code, resp.text) 159 try: 160 return resp.status_code, resp.json() File ~/opt/anaconda3/lib/python3.9/site-packages/pyowm/commons/http_client.py:313, in HttpClient.check_status_code(cls, status_code, payload) 311 raise exceptions.APIRequestError(payload) 312 elif status_code == 401: --> 313 raise exceptions.UnauthorizedError('Invalid API Key provided') 314 elif status_code == 404: 315 raise exceptions.NotFoundError('Unable to find the resource')解释下是什么原因的报错

这个报错是因为你提供的OpenWeatherMap API Key无效或者过期了。在OpenWeatherMap的API访问中，API Key是必须的，如果API Key无效或者过期，将无法访问OpenWeatherMap API。请检查你的API Key是否正确并且有效。...

def show_bboxes(axes, bboxes, labels=None, colors=None): def _make_list(obj, default_values=None): if obj is None: obj = default_values elif not isinstance(obj, (list, tuple)): obj = [obj] return obj labels = _make_list(labels) colors = _make_list(colors, ['b', 'g', 'r', 'm', 'c']) for i, bbox in enumerate(bboxes): color = colors[i % len(colors)] rect = d2l.bbox_to_rect(bbox.detach().numpy(), color) axes.add_patch(rect) if labels and len(labels) > 1: text_color = 'k' if color == 'w' else 'w' axes.text(rect.xy[0], rect.xy[1], labels[i], va='center', ha='center', fontsize=9, color=text_color, bbox=dict(facecolor=color, lw=0))

这是一个名为show_bboxes的函数，它用于在图形界面上显示边界框（bboxes）。函数有几个参数： - axes：图形界面的坐标轴对象。 - bboxes：边界框的列表，每个边界框由四个坐标值表示。 - labels：边界框...

data: {"id":"cmpl-ebd0805b48ee4ea3bf0a09a907c6fef7","object":"text_completion","created":1744333528,"model":"/host/models/NLM_2B","choices":[{"index":0,"text":"!","logprobs":null,"finish_reason":null,"stop_reason":null}],"usage":null} data: {"id":"cmpl-ebd0805b48ee4ea3bf0a09a907c6fef7","object":"text_completion","created":1744333528,"model":"/host/models/NLM_2B","choices":[{"index":0,"text":"!","logprobs":null,"finish_reason":null,"stop_reason":null}],"usage":null} 写一个Python脚本，将a.json文件中的text的值提取出来并且拼接一句话生成b.json

for item in obj: results.extend(extract_texts(item)) return results # 调用函数 process_json('a.json', 'b.json') --- #### 关键点分析 1. **读取 JSON 文件** 使用 with open() 打开文件，并...

[openai_api_compatible] Error: PluginInvokeError: {"args":{"description":"[models] Error: Object of type ImagePromptMessageContent is not JSON serializable"},"error_type":"InvokeError","message":"[models] Error: Object of type ImagePromptMessageContent is not JSON serializable"}

if isinstance(obj, datetime): return obj.isoformat() # 处理datetime类型[^2] elif isinstance(obj, Image.Image): # 处理PIL图像 return self.image_to_base64(obj) else: return super().default(obj) ...

{"model_details":{"multimodal$Normal":{"name":"Normal","score":"0.99000001"}}}对这样的json怎么取name下的值

if isinstance(obj, dict): # 如果当前项是一个字典 for key, value in obj.items(): if key == 'name': result.append(value) elif isinstance(value, (dict, list)): find_all_names(value, result) el

obj_from_dict

setattr(self, key, DictToObject(value) if isinstance(value, dict) else value) data_dict = {'name': 'Alice', 'details': {'age': 25, 'city': 'New York'}} obj = DictToObject(data_dict) print(obj.name)...

相关推荐

JSON 的正确用法探讨：Pyhong、MongoDB、JavaScript与Ajax

XML与JSON文件代码例子

python json 递归打印所有json子节点信息的例子

Python内置库__builtin__秘籍全解：从函数到自定义异常的实战手册（详解）

【JSON_XML 数据处理秘籍】：DAHUA HTTP API 高级技巧解析

揭秘Python内置库__builtin__：提升代码效率与对象管理的20个技巧

顯示結果是 { "match_id": "1904138", "home_team": {}, "away_team": {}, "source": "https://live.titan007.com/detail/1904138.htm" }

[openai_api_compatible] Error: PluginInvokeError: {"args":{"description":"[models] Error: Object of type ImagePromptMessageContent is not JSON serializable"},"error_type":"InvokeError","message":"[models] Error: Object of type ImagePromptMessageContent is not JSON serializable"}

{"model_details":{"multimodal$Normal":{"name":"Normal","score":"0.99000001"}}}对这样的json怎么取name下的值

obj_from_dict

大家在看

linux项目开发资源-firefox-esr-78.6流览器arm64安装包

VMware-VMRC (VMRC) 11.0.0-15201582 for Windows

高频双调谐谐振放大电路设计3MHz+电压200倍放大.zip

ffmpeg官方4.2源码编译出来的动态库

Delphi编写的SQL查询分析器.rar

最新推荐

python json 递归打印所有json子节点信息的例子

实现Struts2+IBatis+Spring集成的快速教程

【数据融合技术】：甘肃土壤类型空间分析中的专业性应用

Waymo使用稀疏图卷积处理LiDAR点云，目标检测精度提升15%

Dwr实现无刷新分页功能的代码与数据库实例

【空间分布规律】：甘肃土壤类型与农业生产的关联性研究

缓存延迟双删的实际解决方案通常怎么实现

企业内部文档管理平台使用Asp.net技术构建

【制图技术】：甘肃高质量土壤分布TIF图件的成图策略

化学结构式手写识别的第三方 API

Python内置库builtin秘籍全解：从函数到自定义异常的实战手册（详解）

揭秘Python内置库builtin：提升代码效率与对象管理的20个技巧