Python如何处理多分隔符的字符串拆分

# 典型复杂字符串示例
log_line = "2023-05-01t08:15:23 | info | server-01 | user login | id:101, name:'zhang san'"
csv_line = "productid:1024; name:python cookbook; price:45.99; categories:books,programming"

# 基本split方法
parts = log_line.split('|')
# 结果: ['2023-05-01t08:15:23 ', ' info ', ' server-01 ', ' user login ', " id:101, name:'zhang san'"]

# 链式处理多个分隔符
def multi_split(text, delimiters):
    for delim in delimiters:
        text = text.replace(delim, delimiters[0])
    return text.split(delimiters[0])

# 使用示例
csv_data = "name:zhang san; age:30; location:new york"
delims = [':', ';']
result = multi_split(csv_data, delims)
# 结果: ['name', 'zhang san', ' age', '30', ' location', 'new york']

import re

# 使用正则表达式拆分
address = "123 main st, suite 100, new york, ny 10001"
parts = re.split(r',|\s', address)  # 按逗号或空格拆分
# 结果: ['123', 'main', 'st', '', 'suite', '100', '', 'new', 'york', '', 'ny', '10001']

# 只拆分特定模式
text = "apple, banana; cherry: date"
parts = re.split(r'[,;:]', text)  # 匹配逗号、分号或冒号
# 结果: ['apple', ' banana', ' cherry', ' date']

# 使用捕获分组保留分隔符
text = "hello! how are you? i'm fine."
parts = re.split(r'([!?.])', text)  # 保留标点符号
# 结果: ['hello', '!', ' how are you', '?', " i'm fine", '.']

# 处理多种空格变体
text = "python\tis\na great\r\nprogramming language"
parts = re.split(r'\s+', text)  # 匹配任意空白字符序列
# 结果: ['python', 'is', 'a', 'great', 'programming', 'language']

def stateful_split(text, delimiters):
    """带状态的分割引擎，处理引号内的分隔符"""
    tokens = []
    current = []
    in_quote = false
    quote_char = none
    
    for char in text:
        if char in ('"', "'") and not in_quote:
            in_quote = true
            quote_char = char
            current.append(char)
        elif char == quote_char and in_quote:
            in_quote = false
            quote_char = none
            current.append(char)
        elif char in delimiters and not in_quote:
            if current:
                tokens.append(''.join(current))
                current = []
        else:
            current.append(char)
    
    if current:
        tokens.append(''.join(current))
    
    return tokens

# 测试包含引号的字符串
text = 'name="zhang, san" age=30 city="new, york"'
result = stateful_split(text, [' ', '=', ','])
# 结果: ['name', '"zhang, san"', 'age', '30', 'city', '"new, york"']

def recursive_split(text, delimiters):
    """递归处理分层分隔符"""
    if not delimiters:
        return [text]
    
    current_delim = delimiters[0]
    remaining_delims = delimiters[1:]
    
    parts = []
    for part in text.split(current_delim):
        if remaining_delims:
            parts.extend(recursive_split(part, remaining_delims))
        else:
            parts.append(part)
    
    return parts

# 分层拆分示例
text = "a:b;c,d|e;f"
result = recursive_split(text, [';', ',', ':', '|'])
# 结果: ['a', 'b', 'c', 'd', 'e', 'f']

def stream_split(text, delimiters):
    """生成器实现流式分割，节省内存"""
    current = []
    for char in text:
        if char in delimiters:
            if current:
                yield ''.join(current)
                current = []
        else:
            current.append(char)
    if current:
        yield ''.join(current)

# 处理大文件
with open('huge_file.txt') as f:
    for line in f:
        for token in stream_split(line, [',', ';', '|']):
            process_token(token)  # 流式处理每个token

def parse_log_line(line):
    """解析复杂日志格式"""
    # 定义日志格式: [时间] [级别] [服务器] [消息] [额外数据]
    pattern = r'\[(.*?)\] \[(.*?)\] \[(.*?)\] - (.*?) \| (.*)'
    match = re.match(pattern, line)
    
    if match:
        timestamp, level, server, message, extra = match.groups()
        
        # 解析额外数据
        extra_data = {}
        for item in re.split(r',\s*', extra):
            if ':' in item:
                key, value = re.split(r':\s*', item, 1)
                extra_data[key] = value.strip("'\"")
        
        return {
            'timestamp': timestamp,
            'level': level,
            'server': server,
            'message': message,
            'extra': extra_data
        }
    return none

# 示例日志
log_line = '[2023-05-01t08:15:23] [info] [server-01] - user login | id:101, name:"zhang san", role:admin'
parsed = parse_log_line(log_line)

def clean_csv_line(line, delimiters=[',', ';', '|']):
    """处理混合分隔符的csv行"""
    # 第一步：统一分隔符
    normalized = line
    for delim in delimiters[1:]:
        normalized = normalized.replace(delim, delimiters[0])
    
    # 第二步：处理引号内的分隔符
    tokens = []
    current = []
    in_quote = false
    
    for char in normalized:
        if char == '"':
            in_quote = not in_quote
            current.append(char)
        elif char == delimiters[0] and not in_quote:
            tokens.append(''.join(current))
            current = []
        else:
            current.append(char)
    
    tokens.append(''.join(current))
    
    # 第三步：去除多余空格
    return [token.strip() for token in tokens]

# 测试混合分隔符csv
csv_line = '101; "zhang, san", 30; "new, york" | "software engineer"'
cleaned = clean_csv_line(csv_line)
# 结果: ['101', '"zhang, san"', '30', '"new, york"', '"software engineer"']

def advanced_tokenizer(text):
    """高级文本分词器"""
    # 处理缩写和特殊符号
    text = re.sub(r"(\w+)'(\w+)", r"\1'\2", text)  # 保留i'm中的撇号
    text = re.sub(r"(\w+)\.(\w+)", r"\1.\2", text)  # 保留e.g.中的点
    
    # 定义分词模式
    pattern = r'''
        \w+(?:-\w+)*        # 带连字符的单词
        | \d+\.\d+          # 浮点数
        | \d+               # 整数
        | \.\.\.            # 省略号
        | [^\w\s]           # 其他符号
    '''
    
    return re.findall(pattern, text, re.verbose)

# 测试复杂文本
text = "i'm 99.9% sure that a.i. will change the world... don't you think?"
tokens = advanced_tokenizer(text)
# 结果: ["i'm", '99.9', '%', 'sure', 'that', 'a.i.', 'will', 'change', 'the', 'world', '...', "don't", 'you', 'think', '?']

# 预编译常用模式
delimiter_pattern = re.compile(r'[,;:|]')
whitespace_pattern = re.compile(r'\s+')

def optimized_split(text):
    """使用预编译正则提高性能"""
    return delimiter_pattern.split(text)

# 性能对比（100万次调用）：
# 未编译: 2.8秒
# 预编译: 1.2秒

# 使用cython编写高性能分割函数
# splitter.pyx
def cython_split(text, delimiters):
    cdef list tokens = []
    cdef list current = []
    cdef char c
    cdef set delim_set = set(delimiters)
    
    for c in text:
        if c in delim_set:
            if current:
                tokens.append(''.join(current))
                current = []
        else:
            current.append(c)
    
    if current:
        tokens.append(''.join(current))
    
    return tokens

# 编译后调用
from splitter import cython_split
result = cython_split("a,b;c:d", [',', ';', ':'])

from concurrent.futures import processpoolexecutor
import os

def parallel_file_split(file_path, delimiters, workers=4):
    """并行处理大文件分割"""
    results = []
    chunk_size = os.path.getsize(file_path) // workers
    
    with open(file_path, 'r') as f:
        with processpoolexecutor(max_workers=workers) as executor:
            futures = []
            start = 0
            
            for i in range(workers):
                end = start + chunk_size
                if i == workers - 1:
                    end = none  # 最后一块包含剩余内容
                
                # 提交任务
                futures.append(executor.submit(
                    process_chunk, file_path, start, end, delimiters
                ))
                start += chunk_size
            
            # 收集结果
            for future in futures:
                results.extend(future.result())
    
    return results

def process_chunk(file_path, start, end, delimiters):
    """处理文件块"""
    tokens = []
    with open(file_path, 'r') as f:
        if start > 0:
            f.seek(start)
            # 找到下一个完整行开始
            while f.read(1) not in ('\n', '\r'):
                start -= 1
                f.seek(start)
        
        # 读取直到结束位置
        while true:
            pos = f.tell()
            if end is not none and pos >= end:
                break
                
            line = f.readline()
            if not line:
                break
                
            tokens.extend(advanced_split(line, delimiters))
    
    return tokens

# 空字符串处理
text = ",a,b,,c,"
# 错误: ['', 'a', 'b', '', 'c', '']
# 正确: [x for x in text.split(',') if x] → ['a', 'b', 'c']

# 可读性优先
def parse_config_line(line):
    # 注释处理
    if line.startswith('#') or not line.strip():
        return none
    
    # 键值分割
    if '=' in line:
        key, value = line.split('=', 1)
        return key.strip(), value.strip()
    
    return line.strip()

# 错误：处理非ascii分隔符
text = "日本＄東京＄中国＄北京"
parts = text.split('＄')  # 全角美元符号

# 解决方案：明确指定分隔符
delim = '＄'  # 直接使用实际字符

# 错误：未转义特殊字符
text = "a.b|c"
parts = re.split(r'.|', text)  # .和|在正则中有特殊含义

# 解决方案：正确转义
parts = re.split(r'\.|\|', text)  # 结果: ['a', 'b', 'c']

# 危险：一次性读取大文件
with open('huge.log') as f:
    lines = f.readlines()  # 可能耗尽内存
    for line in lines:
        parts = line.split('|')

# 解决方案：流式处理
with open('huge.log') as f:
    for line in f:
        parts = line.split('|')