验证并修复 CSV 内容(目标为 3 列:title, page_number, level)。 1. 若列数少于 3,尝试从右向左将全角逗号替换为半角逗号(最多替换两次)。 2. 若解析为 4 列以上,说明标题中包含了半角逗号,将前 n-2 列合并为 title。
(content: str)
| 126 | |
| 127 | # 1. 如果列数少于3,尝试将全角逗号替换为半角逗号 |
| 128 | # 因为目标是3列,最多需要2个分隔符,所以尝试从右向左替换两次 |
| 129 | for _ in range(2): |
| 130 | if len(row) < 3: |
| 131 | last_fullwidth_idx = line.rfind(',') |
| 132 | if last_fullwidth_idx != -1: |
| 133 | line = line[:last_fullwidth_idx] + ',' + line[last_fullwidth_idx+1:] |
| 134 | try: |
| 135 | row = next(csv.reader(StringIO(line))) |
| 136 | except Exception: |
| 137 | pass |
| 138 | else: |
| 139 | break # 没有全角逗号可替换了 |
| 140 | |
| 141 | # 2. 根据最终的 row 长度进行处理并重新生成标准 CSV 行 |
| 142 | if len(row) > 3: |
| 143 | # 超过3列:合并前 n-2 列作为 title |
| 144 | title = ",".join(row[:-2]) |
| 145 | page = row[-2] |
| 146 | level = row[-1] |
| 147 | buffer = StringIO() |
| 148 | writer = csv.writer(buffer) |
| 149 | writer.writerow([title, page, level]) |
| 150 | fixed_lines.append(buffer.getvalue().strip()) |
| 151 | elif len(row) == 3: |
| 152 | # 正常3列:使用 csv.writer 重新写入,确保格式绝对标准 |
| 153 | buffer = StringIO() |
| 154 | writer = csv.writer(buffer) |
| 155 | writer.writerow(row) |
| 156 | fixed_lines.append(buffer.getvalue().strip()) |
| 157 | else: |
| 158 | # 仍然无法修复为至少3列,原样保留,后续校验会失败或跳过 |
| 159 | fixed_lines.append(line) |
| 160 | |
| 161 | return '\n'.join(fixed_lines) |
| 162 | |
| 163 | def parse_csv_response(csv_text: str, source_file: str) -> list: |
| 164 | """ |
| 165 | 将模型返回的 CSV 文本解析为目标 JSON 结构。 |
| 166 | 目标结构:[{"text": "...", "number": int, "level": int}, ...] |
| 167 | """ |
| 168 | cleaned_text = csv_text.strip() |
| 169 | # 移除可能的 markdown 代码块标记 |
| 170 | if cleaned_text.startswith("```csv"): |
| 171 | cleaned_text = cleaned_text[6:] |
| 172 | if cleaned_text.startswith("```"): |
| 173 | cleaned_text = cleaned_text[3:] |
| 174 | if cleaned_text.endswith("```"): |
| 175 | cleaned_text = cleaned_text[:-3] |
| 176 | cleaned_text = cleaned_text.strip() |
| 177 | |
| 178 | # 注入防全角逗号与列数修复逻辑 |
| 179 | cleaned_text = validate_and_fix_csv_content(cleaned_text) |
| 180 | |
| 181 | result_data = [] |
| 182 | try: |
| 183 | reader = csv.DictReader(StringIO(cleaned_text)) |
| 184 | # 检查必要的列 |
| 185 | if not {'title', 'page_number', 'level'}.issubset(set(reader.fieldnames or [])): |
no outgoing calls
no test coverage detected