Python JSON终极指南：从基础到高级实战

import json

# Python对象转JSON字符串
data = {
    "name": "Alice",
    "age": 30,
    "is_employee": True,
    "skills": ["Python", "Data Analysis"],
    "projects": None
}

json_str = json.dumps(data, indent=2)
print("序列化结果:")
print(json_str)

# JSON字符串转Python对象
restored_data = json.loads(json_str)
print("\n反序列化结果:")
print(restored_data)

# 写入JSON文件
def write_json(data, filename, encoding='utf-8'):
    with open(filename, 'w', encoding=encoding) as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

# 读取JSON文件
def read_json(filename, encoding='utf-8'):
    with open(filename, 'r', encoding=encoding) as f:
        return json.load(f)

# 使用示例
user_data = {
    "id": 12345,
    "preferences": {"theme": "dark", "language": "zh-CN"}
}

write_json(user_data, 'user_settings.json')
loaded_data = read_json('user_settings.json')
print(f"加载的数据: {loaded_data}")

class User:
    def __init__(self, name, email, join_date):
        self.name = name
        self.email = email
        self.join_date = join_date
        self._password = "secure_hash"  # 敏感信息不应序列化
    
    def to_dict(self):
        return {
            "name": self.name,
            "email": self.email,
            "join_date": self.join_date.isoformat()
        }

# 自定义编码器
class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, User):
            return obj.to_dict()
        elif isinstance(obj, datetime):
            return obj.isoformat()
        elif isinstance(obj, Decimal):
            return float(obj)
        return super().default(obj)

# 使用自定义编码器
user = User("Bob", "bob@example.com", datetime.now())
json_data = json.dumps(user, cls=CustomEncoder)
print(f"自定义对象序列化: {json_data}")

def serialize_complex(obj):
    """处理复杂对象图的序列化函数"""
    visited = set()
    
    def _serialize(item):
        if id(item) in visited:
            return "<循环引用>"
        visited.add(id(item))
        
        if isinstance(item, (list, tuple, set)):
            return [_serialize(i) for i in item]
        elif isinstance(item, dict):
            return {k: _serialize(v) for k, v in item.items()}
        elif hasattr(item, '__dict__'):
            return _serialize(item.__dict__)
        else:
            return item
    
    return _serialize(obj)

# 测试循环引用
class Node:
    def __init__(self, value):
        self.value = value
        self.children = []

root = Node("root")
child1 = Node("child1")
child2 = Node("child2")
root.children = [child1, child2]
child1.children = [root]  # 创建循环引用

safe_data = serialize_complex(root)
print(f"处理循环引用后的JSON: {json.dumps(safe_data)}")

import ijson

def process_large_json(filename):
    """流式处理大型JSON数组"""
    count = 0
    total_amount = 0
    
    with open(filename, 'rb') as f:
        # 使用items获取数组中的对象
        for record in ijson.items(f, 'item'):
            # 处理每个记录
            count += 1
            total_amount += record.get('amount', 0)
            
            if count % 10000 == 0:
                print(f"已处理 {count} 条记录...")
    
    print(f"总计记录: {count}, 总金额: {total_amount}")

# 生成模拟大型JSON文件
def generate_large_json(filename, num_records=1000000):
    import random
    with open(filename, 'w') as f:
        f.write('[')
        for i in range(num_records):
            record = {
                "id": i,
                "name": f"Item_{i}",
                "amount": random.uniform(10, 1000),
                "timestamp": datetime.now().isoformat()
            }
            json.dump(record, f)
            if i < num_records - 1:
                f.write(',')
        f.write(']')

# 使用示例
generate_large_json('big_data.json', 100000)
process_large_json('big_data.json')

import timeit
import json
import ujson
import orjson
import numpy as np

# 创建大型数据集
data = {
    "users": [
        {
            "id": i,
            "name": f"User_{i}",
            "email": f"user_{i}@example.com",
            "balance": np.random.uniform(100, 10000),
            "is_active": np.random.choice([True, False])
        } 
        for i in range(10000)
    ]
}

# 测试函数
def test_serialization(lib_name):
    lib = globals()[lib_name]
    if lib_name == "orjson":
        return lambda: lib.dumps(data)
    return lambda: lib.dumps(data)

def test_deserialization(lib_name, json_str):
    lib = globals()[lib_name]
    if lib_name == "orjson":
        return lambda: lib.loads(json_str)
    return lambda: lib.loads(json_str)

# 基准测试
libs = ["json", "ujson", "orjson"]
results = {}

for lib in libs:
    # 序列化测试
    ser_time = timeit.timeit(
        test_serialization(lib), 
        number=100
    )
    
    # 反序列化测试
    json_str = test_serialization(lib)()
    deser_time = timeit.timeit(
        test_deserialization(lib, json_str), 
        number=100
    )
    
    results[lib] = {
        "serialize": ser_time,
        "deserialize": deser_time
    }

# 打印结果
print("库\t\t序列化时间\t反序列化时间")
for lib, times in results.items():
    print(f"{lib:8}\t{times['serialize']:.6f}\t{times['deserialize']:.6f}")

import json

def safe_json_loads(json_str):
    """安全的JSON解析函数"""
    try:
        # 限制最大嵌套深度
        return json.loads(json_str, max_depth=20)
    except RecursionError:
        raise ValueError("JSON结构嵌套过深")
    except json.JSONDecodeError as e:
        raise ValueError(f"无效的JSON格式: {str(e)}")

def sanitize_json_input(data):
    """清理用户提供的JSON数据"""
    if isinstance(data, dict):
        return {k: sanitize_json_input(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [sanitize_json_input(item) for item in data]
    elif isinstance(data, str):
        # 转义HTML特殊字符
        return data.replace("<", "&lt;").replace(">", "&gt;")
    else:
        return data

# 使用安全函数处理用户输入
user_input = '{"name": "<script>alert(1)</script>", "admin": true}'
try:
    parsed = safe_json_loads(user_input)
    safe_data = sanitize_json_input(parsed)
    print(f"安全处理后的数据: {safe_data}")
except ValueError as e:
    print(f"安全错误: {str(e)}")

import json
from functools import wraps

def mask_sensitive_fields(fields):
    """敏感字段掩码装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            if isinstance(result, dict):
                for field in fields:
                    if field in result:
                        value = str(result[field])
                        result[field] = value[:2] + '*' * (len(value) - 4) + value[-2:]
            return result
        return wrapper
    return decorator

class SecureJSONEncoder(json.JSONEncoder):
    """安全JSON编码器，自动过滤敏感字段"""
    sensitive_keys = {'password', 'api_key', 'token', 'credit_card'}
    
    def encode(self, obj):
        if isinstance(obj, dict):
            obj = {k: v for k, v in obj.items() if k not in self.sensitive_keys}
        return super().encode(obj)

# 使用示例
@mask_sensitive_fields(['email', 'phone'])
def get_user_profile(user_id):
    """模拟获取用户信息"""
    return {
        "id": user_id,
        "name": "Alice",
        "email": "alice@example.com",
        "phone": "13800138000",
        "password": "hashed_value"
    }

profile = get_user_profile(123)
print(f"掩码后的数据: {json.dumps(profile, cls=SecureJSONEncoder)}")

from datetime import datetime, date, time
import json

class DateTimeEncoder(json.JSONEncoder):
    """日期时间编码器"""
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        elif isinstance(obj, date):
            return obj.isoformat()
        elif isinstance(obj, time):
            return obj.isoformat()
        return super().default(obj)

def datetime_decoder(dct):
    """自动转换ISO格式的日期时间字符串"""
    for k, v in dct.items():
        if isinstance(v, str):
            try:
                # 尝试解析日期时间
                dct[k] = datetime.fromisoformat(v)
            except (ValueError, TypeError):
                pass
    return dct

# 使用示例
data = {
    "event": "Python Conference",
    "start_time": datetime.now(),
    "end_date": date.today()
}

json_str = json.dumps(data, cls=DateTimeEncoder)
print(f"序列化结果: {json_str}")

restored = json.loads(json_str, object_hook=datetime_decoder)
print(f"反序列化后类型: {type(restored['start_time'])}")

import json
import base64

class BinaryEncoder(json.JSONEncoder):
    """二进制数据编码器"""
    def default(self, obj):
        if isinstance(obj, bytes):
            return {"_binary": base64.b64encode(obj).decode('utf-8')}
        return super().default(obj)

def binary_decoder(dct):
    """二进制数据解码器"""
    if "_binary" in dct:
        return base64.b64decode(dct["_binary"])
    return dct

# 使用示例
data = {
    "filename": "image.png",
    "thumbnail": b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR..."  # 二进制数据
}

json_str = json.dumps(data, cls=BinaryEncoder)
print(f"序列化结果: {json_str}")

restored = json.loads(json_str, object_hook=binary_decoder)
print(f"反序列化后类型: {type(restored['thumbnail'])}")

from jsonschema import validate, ValidationError
import json

# 定义用户配置的JSON Schema
user_schema = {
    "type": "object",
    "properties": {
        "username": {
            "type": "string",
            "minLength": 3,
            "maxLength": 20
        },
        "email": {
            "type": "string",
            "format": "email"
        },
        "age": {
            "type": "integer",
            "minimum": 18,
            "maximum": 100
        },
        "preferences": {
            "type": "object",
            "properties": {
                "theme": {"enum": ["light", "dark", "system"]},
                "notifications": {"type": "boolean"}
            },
            "required": ["theme"],
            "additionalProperties": False
        }
    },
    "required": ["username", "email"],
    "additionalProperties": False
}

def validate_user_config(config):
    """验证用户配置是否符合Schema"""
    try:
        validate(instance=config, schema=user_schema)
        return True, "配置有效"
    except ValidationError as e:
        return False, f"配置错误: {e.message}"

# 测试用例
valid_config = {
    "username": "alice123",
    "email": "alice@example.com",
    "age": 30,
    "preferences": {
        "theme": "dark",
        "notifications": True
    }
}

invalid_config = {
    "username": "ab",  # 太短
    "email": "invalid-email",
    "age": 15,         # 太小
    "preferences": {
        "theme": "blue"  # 无效值
    },
    "extra_field": True  # 不允许的额外字段
}

print(validate_user_config(valid_config))
print(validate_user_config(invalid_config))

import json
from jsonschema import Draft7Validator
import json_schema_generator

def generate_schema_from_json(filename):
    """从JSON文件生成Schema"""
    with open(filename) as f:
        data = json.load(f)
    
    # 使用第三方库生成Schema
    generator = json_schema_generator.SchemaGenerator()
    generator.add_object(data)
    
    # 获取并优化Schema
    schema = generator.to_schema()
    Draft7Validator.check_schema(schema)  # 验证Schema有效性
    
    # 简化Schema
    if "definitions" in schema and len(schema["definitions"]) == 1:
        # 如果只有一个定义，直接使用它
        main_def = next(iter(schema["definitions"].values()))
        schema.update(main_def)
        del schema["definitions"]
    
    return schema

# 使用示例
config = {
    "app_name": "JSON Tool",
    "version": "1.0.0",
    "features": ["validation", "generation"],
    "settings": {
        "auto_save": True,
        "interval": 300
    }
}

with open('app_config.json', 'w') as f:
    json.dump(config, f)

schema = generate_schema_from_json('app_config.json')
print("生成的JSON Schema:")
print(json.dumps(schema, indent=2))

import json
import os
from collections import ChainMap

class ConfigManager:
    """分层配置管理系统"""
    def __init__(self, base_config='config.json', env_config='env.json'):
        self.base_config = self._load_config(base_config)
        self.env_config = self._load_config(env_config) if env_config else {}
        self.runtime_config = {}
        
        # 合并配置：运行时 > 环境 > 基础
        self.config = ChainMap(
            self.runtime_config,
            self.env_config,
            self.base_config
        )
    
    def _load_config(self, filename):
        """加载JSON配置文件"""
        if not os.path.exists(filename):
            return {}
        
        with open(filename, 'r') as f:
            try:
                return json.load(f)
            except json.JSONDecodeError:
                print(f"警告: {filename} 包含无效JSON，将使用空配置")
                return {}
    
    def get(self, key, default=None):
        """获取配置值"""
        return self.config.get(key, default)
    
    def set(self, key, value):
        """设置运行时配置"""
        self.runtime_config[key] = value
    
    def save_env_config(self):
        """保存环境配置到文件"""
        with open('env.json', 'w') as f:
            json.dump(self.env_config, f, indent=2)

# 使用示例
if __name__ == "__main__":
    # 创建基础配置文件
    with open('config.json', 'w') as f:
        json.dump({
            "app_name": "MyApp",
            "debug": False,
            "log_level": "INFO"
        }, f)
    
    # 初始化配置管理器
    config = ConfigManager()
    
    # 使用环境变量覆盖配置
    config.set("log_level", "DEBUG")
    
    print(f"应用名称: {config.get('app_name')}")
    print(f"日志级别: {config.get('log_level')}")
    print(f"调试模式: {config.get('debug')}")

import json
import requests
from requests.exceptions import RequestException

class APIClient:
    """基于JSON的API客户端"""
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json"
        })
        
        if api_key:
            self.session.headers["Authorization"] = f"Bearer {api_key}"
    
    def _handle_response(self, response):
        """处理API响应"""
        try:
            response.raise_for_status()
            return response.json()
        except RequestException as e:
            error_msg = f"API请求失败: {str(e)}"
            try:
                error_data = response.json()
                error_msg += f" | 错误信息: {error_data.get('error', '未知错误')}"
            except json.JSONDecodeError:
                error_msg += f" | 响应内容: {response.text[:200]}"
            raise APIException(error_msg) from e
    
    def get(self, endpoint, params=None):
        """发送GET请求"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.get(url, params=params)
        return self._handle_response(response)
    
    def post(self, endpoint, data=None):
        """发送POST请求"""
        url = f"{self.base_url}/{endpoint}"
        json_data = json.dumps(data) if data else None
        response = self.session.post(url, data=json_data)
        return self._handle_response(response)
    
    def patch(self, endpoint, data):
        """发送PATCH请求"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.patch(url, data=json.dumps(data))
        return self._handle_response(response)
    
    def delete(self, endpoint):
        """发送DELETE请求"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.delete(url)
        return self._handle_response(response)

class APIException(Exception):
    """自定义API异常"""
    pass

# 使用示例
if __name__ == "__main__":
    # 创建API客户端
    client = APIClient("https://api.example.com/v1", "your_api_key")
    
    try:
        # 获取用户列表
        users = client.get("users")
        print(f"获取到 {len(users)} 个用户")
        
        # 创建新用户
        new_user = {
            "name": "Alice",
            "email": "alice@example.com",
            "role": "developer"
        }
        created_user = client.post("users", new_user)
        print(f"创建的用户ID: {created_user['id']}")
        
        # 更新用户
        update_data = {"role": "admin"}
        updated_user = client.patch(f"users/{created_user['id']}", update_data)
        print(f"更新后的角色: {updated_user['role']}")
        
        # 删除用户
        client.delete(f"users/{created_user['id']}")
        print("用户已删除")
        
    except APIException as e:
        print(f"API错误: {str(e)}")