<think>我们遇到了一个错误：SyntaxError: Non-UTF-8 code starting with '\xe5'。这个错误通常是因为代码中包含了非UTF-8编码的字符，而Python解释器默认使用UTF-8编码。 但是，请注意，我们提供的代码是纯英文的，不应该出现非UTF-8字符。这个错误可能是由于在复制粘贴过程中引入了不可见的非UTF-8字符，或者是在字符串中包含了一些非UTF-8字符。 然而，观察我们提供的样本数据，其中包含大量中文，因此我们需要确保在解析这些包含非ASCII字符的文本时，正确处理编码问题。 实际上，在之前的代码中，我们使用了字符串处理，而Python 3默认使用Unicode字符串，所以如果输入文本是UTF-8编码的，那么应该没有问题。但是，如果输入文本不是UTF-8编码，那么我们需要先将其解码为Unicode字符串。 但是，在问题中，我们直接给出了一个字符串字面量，其中包含中文字符。在Python中，如果源代码中包含非ASCII字符，我们需要在文件开头声明编码。通常我们会在文件开头加上： # -*- coding: utf-8 -*- 然而，在Python 3中，默认的编码就是UTF-8，所以通常不需要这样做。但是，如果我们在运行代码时遇到编码问题，可能是因为保存的源代码文件不是UTF-8编码。 另外，在代码中，我们使用了`ensure_ascii=False`来打印JSON，这样中文字符就不会被转义。 现在，我们重新审视问题：用户提供的样本数据中包含中文字符，而我们在代码中直接将其作为字符串字面量。如果运行环境不是UTF-8，则可能出现问题。 解决方案： 1. 确保源代码文件以UTF-8编码保存。 2. 如果从文件读取数据，确保以UTF-8编码打开文件。 但是，用户的问题描述中出现了错误，可能是由于在复制代码时出现了编码问题。因此，我们提供以下改进方案： 我们不再将样本数据硬编码在代码中，而是假设用户会从文件或网络等来源获取数据。这样，我们可以通过正确的编码读取数据。 修改代码：从文件中读取文本，并指定编码为UTF-8。 同时，为了避免在源代码中直接包含中文字符串，我们可以将样本数据放在一个单独的文件中，然后读取。 不过，为了演示，我们也可以将样本数据以字节形式提供，然后解码，但这样不太方便。 另一种做法：在代码中直接使用字符串，但确保整个源代码文件保存为UTF-8。 由于我们无法控制用户的环境，我们建议用户在使用时注意编码。 但是，根据错误信息，问题似乎发生在代码执行之前，即解析源代码时。因此，我们只需要确保源代码文件保存为UTF-8即可。 现在，我们调整代码，使其能够处理从文件读取的情况，并指定编码。 另外，我们也可以修改函数，使其能够处理字节输入，但为了简单，我们假设输入是字符串（已经是Unicode）。 因此，我们不需要修改之前的函数，但需要确保输入到函数中的文本是Unicode字符串。 如果用户的数据源是字节流，则应该先解码为字符串。 例如，如果从文件读取： ```python with open('input.txt', 'r', encoding='utf-8') as f: text = f.read() ``` 然后，将`text`传递给`extract_data_blocks`函数。 所以，我们之前的函数是没问题的，问题可能出在如何将样本数据传递给函数。 在示例使用部分，我们硬编码了一个包含中文的字符串。在Python 3中，这是允许的，只要文件编码正确。 因此，我们只需要确保保存代码的文件是UTF-8编码。 如果用户遇到同样的错误，应该检查他们的编辑器是否将文件保存为UTF-8。 现在，我们重新提供代码，并注意在示例使用部分，我们不再提供完整的样本（因为样本很长，而且包含大量中文，可能会在复制时出现问题），而是建议从文件读取。 但是，为了代码的完整性，我们仍然在示例中提供一小段样本数据（仅包含英文和少量非ASCII字符）来演示。 修改后的代码示例使用部分： 我们将样本数据缩短，并避免大量中文字符，只保留必要的结构。这样即使复制也不会出现编码问题。 同时，我们也可以提供从文件读取的示例。 以下是修改后的代码： </think> 我注意到之前的代码示例中包含了大量中文文本，这可能导致编码问题。让我提供一个优化版本，专注于高效提取数据并处理可能的编码问题： ### 优化解决方案：提取事件数据并处理编码 ```python import json import re def extract_event_data(text): """ 从事件流文本中提取所有 data 字段的 JSON 对象 参数: text (str): 原始事件流文本 返回: list: 包含所有解析后 data 对象的列表 """ # 使用正则表达式高效提取所有 data 块 data_blocks = re.findall(r'data:(\{.*?\})(?=\s*\nevent:|$)', text, re.DOTALL) results = [] for block in data_blocks: try: # 清理可能的空白字符和换行 cleaned_block = re.sub(r'\s+', ' ', block.strip()) # 解析 JSON parsed = json.loads(cleaned_block) results.append(parsed) except json.JSONDecodeError as e: print(f"JSON解析错误: {e}\n在数据块: {block[:100]}...") except UnicodeDecodeError: # 尝试处理编码问题 try: decoded_block = block.encode('latin1').decode('utf-8') parsed = json.loads(decoded_block) results.append(parsed) except Exception: print(f"无法解析的数据块: {block[:100]}...") return results # 示例使用 if __name__ == "__main__": # 从文件读取内容，避免编码问题 try: with open('event_data.txt', 'r', encoding='utf-8') as f: text_data = f.read() except UnicodeDecodeError: # 尝试其他编码 with open('event_data.txt', 'r', encoding='latin1') as f: text_data = f.read() # 提取数据 event_data = extract_event_data(text_data) # 保存结果 with open('extracted_data.json', 'w', encoding='utf-8') as f: json.dump(event_data, f, ensure_ascii=False, indent=2) print(f"成功提取 {len(event_data)} 个事件数据") ``` ### 代码解释： 1. **正则表达式提取**： - 使用 `re.findall(r'data:(\{.*?\})(?=\s*\nevent:|$)'` 高效匹配所有 data 块 - `re.DOTALL` 使 `.` 匹配换行符 - 前瞻断言 `(?=\s*\nevent:|$)` 确保匹配完整的 JSON 对象 2. **编码处理**： - 首先尝试 UTF-8 编码读取文件 - 失败时回退到 latin1 编码 - 解析时尝试 latin1 → UTF-8 转换处理混合编码 3. **数据清理**： - `re.sub(r'\s+', ' ', block.strip())` 压缩多余空白字符 - 避免换行符导致的解析问题 4. **错误处理**： - 捕获 JSONDecodeError 和 UnicodeDecodeError - 提供有意义的错误信息 - 跳过无效块而非中断整个处理 5. **文件操作**： - 从文件读取输入，避免代码中的大字符串 - 结果保存为格式化的 JSON 文件 ### 使用说明： 1. 将原始事件数据保存到 `event_data.txt` 2. 运行脚本将提取的数据保存到 `extracted_data.json` 3. 支持处理混合编码的文本数据 4. 自动跳过无法解析的无效数据块 ### 处理特定事件类型： 如果需要只提取特定类型（如 CUSTOM_EVENT）的数据，可添加过滤： ```python # 在 extract_event_data 函数内添加 filtered_results = [] for block in data_blocks: try: parsed = json.loads(block) # 检查事件类型 if 'event' in parsed and parsed['event'] == 'CUSTOM_EVENT': filtered_results.append(parsed) except: pass return filtered_results ```