MCPcopy Create free account
hub / github.com/NatsUIJM/autoContents / validate_and_fix_csv_content

Function validate_and_fix_csv_content

mainprogress/determine_toc_levels.py:128–182  ·  view source on GitHub ↗

验证并修复 CSV 内容(目标为 3 列:title, page_number, level)。 1. 若列数少于 3,尝试从右向左将全角逗号替换为半角逗号(最多替换两次)。 2. 若解析为 4 列以上,说明标题中包含了半角逗号,将前 n-2 列合并为 title。

(content: str)

Source from the content-addressed store, hash-verified

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
163def 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 [])):

Callers 1

parse_csv_responseFunction · 0.70

Calls

no outgoing calls

Tested by

no test coverage detected