CSV 和文档相互转换处理-夜雨聆风

CSV 和文档相互转换处理

"""任务：处理CSV数据文件场景：从CSV导入文档数据"""import csvfrom typing import List, Dictdef csv_to_documents(filepath, text_column="content", metadata_columns=None):    """    将CSV转换为文档列表    返回: [        {"content": "...", "metadata": {"id": "1", "source": "..."}},        ...    ]    """    documents = []    with open(filepath, 'r', encoding='utf-8') as f:        reader = csv.DictReader(f)        for row in reader:            doc = {                'content': row.get(text_column, ''),                'metadata': {}            }            if metadata_columns:                for col in metadata_columns:                    if col in row:                        doc['metadata'][col] = row[col]            documents.append(doc)    return documentsdef documents_to_csv(documents, filepath):    """    将文档列表导出为CSV    """    if not documents:        return    fieldnames = ['content']    if documents[0].get('metadata'):        fieldnames.extend(documents[0]['metadata'].keys())    with open(filepath, 'w', newline='', encoding='utf-8') as f:        writer = csv.DictWriter(f, fieldnames=fieldnames)        writer.writeheader()        for doc in documents:            row = {'content': doc['content']}            row.update(doc.get('metadata', {}))            writer.writerow(row)# 测试# 创建测试CSVwith open("documents.csv", "w", newline='') as f:    writer = csv.writer(f)    writer.writerow(["id", "content", "source"])    writer.writerow(["1", "First document", "web"])    writer.writerow(["2", "Second document", "pdf"])docs = csv_to_documents(    "documents.csv",    text_column="content",    metadata_columns=["id", "source"])assert len(docs) == 2assert docs[0]['metadata']['id'] == "1"

一、场景

需要用 CSV 批量导入文本数据做 AI 应用的场景，例如从 CSV 导入知识库、处理结构化文本数据、模型处理完一批文档再统一导出成 CSV 保存、CSV → 文档 → 向量化 → 存入向量数据库等。

上述代码中分两部分：

csv_to_documents：把 CSV 文件转成程序能直接用的文档对象列表，documents_to_csv：把文档对象列表存回 CSV 文件。

二、个人理解/最初卡在哪里

1、csv.DictReader(f)

创建一个字典读取器对象，输入文件对象f，输出一个可迭代对象，每次迭代返回一个字典。

假设有一个 CSV 文件 data.csv：

name,age,city张三,25,北京李四,30,上海

使用 csv.DictReader（返回字典）：

import csvwith open('data.csv', 'r', encoding='utf-8') as f:    reader = csv.DictReader(f)    for row in reader:        print(row)

输出：

{'name': '张三', 'age': '25', 'city': '北京'}{'name': '李四', 'age': '30', 'city': '上海'}

第一行自动作为键名

name,age,city    # 这一行自动成为字典的键张三,25,北京

自定义列名（如果文件没有表头）

fieldnames = ['姓名', '年龄', '城市']reader = csv.DictReader(f, fieldnames=fieldnames)

指定分隔符（如 TSV 文件）

reader = csv.DictReader(f, delimiter='\t')

常见陷阱

⚠️ 所有值都是字符串，即使 CSV 中是数字，DictReader 也返回字符串：

row = {'age': '25'}  # 是字符串，不是整数age = int(row['age'])  # 需要手动转换

⚠️ BOM 头问题（Excel 保存的 UTF-8 文件）

with open(filepath, 'r', encoding='utf-8-sig') as f:  # 使用 utf-8-sig    reader = csv.DictReader(f)

普通 csv.reader（）（返回列表）

reader = csv.reader(f)for row in reader:    print(row)  # ['张三', '25', '北京']# 需要记住列的顺序：row[0] 是 name，row[1] 是 age

2、writer = csv.writer(f)

是Python 内置的 CSV 表格专用写入工具，会自动处理逗号分隔、引号包裹、换行、特殊字符转义等，只需要传列表，就可以写成标准 CSV 行。

f.write() 原始写入只能写纯字符串，要自己处理逗号、引号、换行，容易写错格式，导致 CSV 打不开，例如：

f.write("id,content,source\n")f.write("1,First document,web\n")

CSV 文件本质是一行一行存的文本表格，所以CSV 里只有 “按行写”，没有 “按列写”。csv 模块最常用的 4 个方法：

① writerow (列表) → 写一行

writer.writerow(["id", "name", "age"])  # 写表头writer.writerow([1, "小明", 20])        # 写一行数据

② writerows (二维列表) → 一次性写多行

rows = [    ["id", "name"],    [1, "小红"],    [2, "小刚"]]writer.writerows(rows)  # 批量写

③ writeheader () → 只写表头（DictWriter 专用）

writer = csv.DictWriter(f, fieldnames=["id", "name"])writer.writeheader()  # 自动写第一行列名

④ writerow (字典) → 按字典写行（DictWriter 专用）

writer.writerow({"id": 1, "name": "小李"})

写 CSV 标准模板

import csvwith open("test.csv", "w", encoding="utf-8", newline="") as f:    writer = csv.writer(f)    writer.writerow(["列1", "列2", "列3"])  # 表头    writer.writerow([1, 2, 3])              # 行1    writer.writerow([4, 5, 6])              # 行2

字典写法（最常用）

withopen("test.csv", "w", encoding="utf-8", newline="") as f:    writer = csv.DictWriter(f, fieldnames=["id", "content"])    writer.writeheader()    writer.writerow({"id": 1, "content": "hello"})

3、row.update(字典)

把另一个字典的所有键值对，批量添加 / 合并到当前字典里，例如：

row 一开始只有 content：

row = {"content": "First document"}

执行：

row.update({"id": "1", "source": "web"})

执行后，row 被合并、更新了，变成：

row = {    "content": "First document",    "id": "1",    "source": "web"}

三、代码中的关键洞察

csv.DictReader 把每行变成 “列名：值” 字典，是上述读取函数的关键。

通过 metadata_columns，不用改函数内部代码，函数非常灵活，想保留哪些列作为元数据，完全由自己控制。

字典结构 {content, metadata} 是 AI/RAG 标准格式：

{  "content": "正文内容",  "metadata": { "id": "1", "source": "web" }}

content 给大模型阅读，metadata 给程序做溯源、过滤、管理，这是知识库、向量库、RAG 系统的标准数据结构。

四、关键代码/可复用片段

CSV 转标准文档

def csv_to_documents(filepath, text_column="content", metadata_columns=None):    documents = []    with open(filepath, 'r', encoding='utf-8') as f:        reader = csv.DictReader(f)        for row in reader:            doc = {                'content': row.get(text_column, ''),                'metadata': {}            }            if metadata_columns:                for col in metadata_columns:                    if col in row:                        doc['metadata'][col] = row[col]            documents.append(doc)    return documents

文档写回 CSV

def documents_to_csv(documents, filepath):    if not documents:        return    fieldnames = ['content']    if documents[0].get('metadata'):        fieldnames.extend(documents[0]['metadata'].keys())    withopen(filepath, 'w', newline='', encoding='utf-8') as f:        writer = csv.DictWriter(f, fieldnames=fieldnames)        writer.writeheader()        for doc in documents:            row = {'content': doc['content']}            row.update(doc.get('metadata', {}))            writer.writerow(row)

一行调用导入数据

docs = csv_to_documents(    "documents.csv",    text_column="content",    metadata_columns=["id", "source", "date"])