Python中嵌套序列扁平化的多种实现方法详解

# 简单嵌套列表扁平化
nested_list = [[1, 2, 3], [4, 5], [6, 7, 8, 9]]
flattened = [item for sublist in nested_list for item in sublist]
print("列表推导式结果:", flattened)  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# 使用itertools.chain
import itertools
flattened_chain = list(itertools.chain.from_iterable(nested_list))
print("itertools.chain结果:", flattened_chain)  # 同上

def flatten_dict(d, parent_key='', sep='_'):
    """字典扁平化"""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

# 使用示例
nested_dict = {
    'name': 'Alice',
    'address': {
        'city': 'New York',
        'zip': {
            'main': 10001,
            'secondary': 10002
        }
    },
    'scores': [90, 85, 95]
}

print("字典扁平化结果:")
print(flatten_dict(nested_dict))
# {'name': 'Alice', 'address_city': 'New York', 
#  'address_zip_main': 10001, 'address_zip_secondary': 10002,
#  'scores': [90, 85, 95]}

def recursive_flatten(nested_list):
    """递归列表扁平化"""
    result = []
    for item in nested_list:
        if isinstance(item, list):
            result.extend(recursive_flatten(item))
        else:
            result.append(item)
    return result

# 使用示例
deep_nested = [1, [2, [3, 4], 5], [6, 7, [8, 9]]]
print("递归扁平化结果:", recursive_flatten(deep_nested))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

def flatten_generator(nested):
    """递归生成器扁平化"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten_generator(item)
        else:
            yield item

# 使用示例
nested_data = [1, [2, (3, 4), [5, [6, 7]]], 8]
print("生成器扁平化:", list(flatten_generator(nested_data)))  # [1, 2, 3, 4, 5, 6, 7, 8]

def iterative_flatten(nested):
    """栈实现迭代扁平化"""
    stack = list(nested)[::-1]
    result = []
    
    while stack:
        item = stack.pop()
        if isinstance(item, (list, tuple)):
            stack.extend(item[::-1])
        else:
            result.append(item)
    
    return result

# 使用示例
complex_nested = [1, [2, [3, [4, 5], 6], 7], 8]
print("迭代扁平化结果:", iterative_flatten(complex_nested))  # [1, 2, 3, 4, 5, 6, 7, 8]

from collections import deque

def bfs_flatten(nested):
    """广度优先扁平化"""
    queue = deque(nested)
    result = []
    
    while queue:
        item = queue.popleft()
        if isinstance(item, (list, tuple)):
            queue.extend(item)
        else:
            result.append(item)
    
    return result

# 使用示例
tree_structure = [1, [2, [3, 4], [5, 6]], [7, 8]]
print("广度优先扁平化:", bfs_flatten(tree_structure))  # [1, 2, 7, 8, 3, 4, 5, 6]

def mixed_flatten(nested):
    """混合类型扁平化"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from mixed_flatten(item)
        elif isinstance(item, dict):
            yield from mixed_flatten(list(item.values()))
        else:
            yield item

# 使用示例
mixed_data = [1, {'a': 2, 'b': [3, 4]}, (5, [6, 7])]
print("混合类型扁平化:", list(mixed_flatten(mixed_data)))  # [1, 2, 3, 4, 5, 6, 7]

def conditional_flatten(nested, condition):
    """带条件扁平化"""
    for item in nested:
        if isinstance(item, (list, tuple)) and condition(item):
            yield from conditional_flatten(item, condition)
        else:
            yield item

# 使用示例
data = [1, [2, [3, 4], [5, 6]], [7, 8]]
# 只扁平化长度大于2的列表
result = list(conditional_flatten(data, lambda x: len(x) > 2))
print("条件扁平化:", result)  # [1, 2, 3, 4, 5, 6, 7, 8]

def json_flatten(data, parent_key='', sep='_'):
    """JSON数据扁平化"""
    if isinstance(data, dict):
        items = []
        for k, v in data.items():
            new_key = f"{parent_key}{sep}{k}" if parent_key else k
            if isinstance(v, (dict, list)):
                items.extend(json_flatten(v, new_key, sep=sep).items())
            else:
                items.append((new_key, v))
        return dict(items)
    elif isinstance(data, list):
        items = {}
        for i, item in enumerate(data):
            new_key = f"{parent_key}{sep}{i}" if parent_key else str(i)
            if isinstance(item, (dict, list)):
                items.update(json_flatten(item, new_key, sep=sep))
            else:
                items[new_key] = item
        return items
    else:
        return {parent_key: data} if parent_key else data

# 使用示例
complex_json = {
    "user": {
        "name": "Alice",
        "age": 30,
        "addresses": [
            {"city": "New York", "zip": 10001},
            {"city": "Boston", "zip": 20001}
        ]
    },
    "orders": [
        {"id": 1, "products": ["A", "B"]},
        {"id": 2, "products": ["C"]}
    ]
}

print("JSON扁平化结果:")
flattened_json = json_flatten(complex_json)
for k, v in flattened_json.items():
    print(f"{k}: {v}")

import json
import ijson

def stream_json_flatten(file_path):
    """流式JSON扁平化"""
    with open(file_path, 'r') as f:
        # 使用ijson解析大型JSON
        parser = ijson.parse(f)
        
        # 当前路径
        current_path = []
        
        for prefix, event, value in parser:
            if event == 'map_key':
                current_path.append(value)
            elif event == 'end_map':
                current_path.pop()
            elif event in ['string', 'number', 'boolean']:
                # 生成扁平键值对
                full_key = '_'.join(current_path)
                yield full_key, value

# 使用示例
# 假设有large_data.json文件
# for key, value in stream_json_flatten('large_data.json'):
#     process(key, value)

class TreeNode:
    """树节点"""
    def __init__(self, value):
        self.value = value
        self.children = []
    
    def add_child(self, node):
        self.children.append(node)

def tree_flatten(root, order='preorder'):
    """树结构扁平化"""
    if order == 'preorder':
        yield root.value
        for child in root.children:
            yield from tree_flatten(child, order)
    elif order == 'postorder':
        for child in root.children:
            yield from tree_flatten(child, order)
        yield root.value
    elif order == 'level':
        queue = [root]
        while queue:
            node = queue.pop(0)
            yield node.value
            queue.extend(node.children)

# 使用示例
root = TreeNode('A')
b = TreeNode('B')
c = TreeNode('C')
d = TreeNode('D')
e = TreeNode('E')
root.add_child(b)
root.add_child(c)
b.add_child(d)
b.add_child(e)

print("树结构前序扁平化:", list(tree_flatten(root)))  # ['A', 'B', 'D', 'E', 'C']
print("树结构后序扁平化:", list(tree_flatten(root, 'postorder')))  # ['D', 'E', 'B', 'C', 'A']
print("树结构层级扁平化:", list(tree_flatten(root, 'level')))  # ['A', 'B', 'C', 'D', 'E']

def nary_tree_to_list(root):
    """多叉树转嵌套列表"""
    if not root.children:
        return root.value
    return [root.value] + [nary_tree_to_list(child) for child in root.children]

# 使用示例
tree_list = nary_tree_to_list(root)
print("树转嵌套列表:", tree_list)  # ['A', [['B', ['D'], ['E']], ['C']]]

# 扁平化嵌套列表
flattened_tree = list(flatten_generator(tree_list))
print("扁平化树结构:", flattened_tree)  # ['A', 'B', 'D', 'E', 'C']

def large_data_flatten(nested_iter):
    """大型数据集扁平化生成器"""
    stack = [iter(nested_iter)]
    while stack:
        try:
            item = next(stack[-1])
            if isinstance(item, (list, tuple)):
                stack.append(iter(item))
            else:
                yield item
        except StopIteration:
            stack.pop()

# 使用示例
def generate_large_data():
    """生成大型嵌套数据集"""
    for i in range(1000):
        yield [i, [i*2, [i*3, i*4]]]

print("大型数据集扁平化:")
count = 0
for item in large_data_flatten(generate_large_data()):
    print(item, end=' ')
    count += 1
    if count >= 10:  # 只显示前10个
        break
# 0 0 0 0 1 2 3 4 2 4 ...

from concurrent.futures import ThreadPoolExecutor

def parallel_flatten(nested_list, max_workers=4):
    """并行扁平化"""
    # 第一层分块
    chunks = [nested_list[i::max_workers] for i in range(max_workers)]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 并行处理每个块
        results = executor.map(recursive_flatten, chunks)
    
    # 合并结果
    return list(itertools.chain.from_iterable(results))

# 使用示例
large_nested = [[i, [i*2, i*3]] for i in range(10000)]
flattened = parallel_flatten(large_nested)
print("\n并行扁平化长度:", len(flattened))  # 30000

def flatten_features(data):
    """特征工程嵌套特征扁平化"""
    flattened = []
    for sample in data:
        flat_sample = {}
        for feature, value in sample.items():
            if isinstance(value, list):
                # 列表特征: 展开为多个特征
                for i, v in enumerate(value):
                    flat_sample[f"{feature}_{i}"] = v
            elif isinstance(value, dict):
                # 字典特征: 扁平化键
                for k, v in value.items():
                    flat_sample[f"{feature}_{k}"] = v
            else:
                flat_sample[feature] = value
        flattened.append(flat_sample)
    return flattened

# 使用示例
features = [
    {'user': 'Alice', 'scores': [90, 85, 95], 'metadata': {'age': 30, 'city': 'NY'}},
    {'user': 'Bob', 'scores': [80, 75], 'metadata': {'age': 25, 'city': 'LA'}}
]

print("特征工程扁平化:")
for flat in flatten_features(features):
    print(flat)
# {'user': 'Alice', 'scores_0': 90, 'scores_1': 85, 'scores_2': 95, 'metadata_age': 30, 'metadata_city': 'NY'}
# {'user': 'Bob', 'scores_0': 80, 'scores_1': 75, 'metadata_age': 25, 'metadata_city': 'LA'}

def config_flatten(config, parent_key='', sep='.'):
    """配置文件扁平化"""
    items = {}
    for k, v in config.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.update(config_flatten(v, new_key, sep))
        else:
            items[new_key] = v
    return items

# 使用示例
app_config = {
    'database': {
        'host': 'localhost',
        'port': 5432,
        'credentials': {
            'user': 'admin',
            'password': 'secret'
        }
    },
    'logging': {
        'level': 'INFO',
        'file': '/var/log/app.log'
    }
}

print("配置文件扁平化:")
flat_config = config_flatten(app_config)
for key, value in flat_config.items():
    print(f"{key}: {value}")
# database.host: localhost
# database.port: 5432
# database.credentials.user: admin
# database.credentials.password: secret
# logging.level: INFO
# logging.file: /var/log/app.log

# 小数据: 递归
def recursive_flatten(data):
    if not isinstance(data, (list, tuple)):
        return [data]
    return [item for sublist in data for item in recursive_flatten(sublist)]

# 大数据: 生成器
def generator_flatten(data):
    for item in data:
        if isinstance(item, (list, tuple)):
            yield from generator_flatten(item)
        else:
            yield item

def safe_flatten(data, visited=None):
    """安全扁平化(防止循环引用)"""
    if visited is None:
        visited = set()

    if id(data) in visited:
        return []
    visited.add(id(data))

    if isinstance(data, (list, tuple)):
        result = []
        for item in data:
            result.extend(safe_flatten(item, visited))
        return result
    else:
        return [data]

from collections.abc import Iterable

def is_nested(item):
    """检查是否可迭代(排除字符串)"""
    return isinstance(item, Iterable) and not isinstance(item, (str, bytes))

def optimized_flatten(data):
    """优化类型检查的扁平化"""
    if is_nested(data):
        return [item for sublist in data for item in optimized_flatten(sublist)]
    return [data]

def iterative_flatten_perf(data):
    """高性能迭代扁平化"""
    stack = [iter(data)]
    result = []
    while stack:
        try:
            item = next(stack[-1])
            if is_nested(item):
                stack.append(iter(item))
            else:
                result.append(item)
        except StopIteration:
            stack.pop()
    return result

def robust_flatten(data):
    """健壮的扁平化函数"""
    try:
        if is_nested(data):
            return [item for sublist in data for item in robust_flatten(sublist)]
        return [data]
    except RecursionError:
        # 递归深度过大转迭代
        return iterative_flatten_perf(data)
    except Exception as e:
        print(f"扁平化错误: {e}")
        return []

def flatten(nested, max_depth=None):
    """
    扁平化嵌套序列

    参数:
        nested: 嵌套序列(列表/元组/字典)
        max_depth: 最大扁平深度(可选)

    返回:
        扁平化后的列表

    注意:
        默认处理所有嵌套层级
        字符串和字节不被视为嵌套结构
    """
    # 实现代码