在Django中用AWS S3与Textract提取PDF键值对的异常问题求助

问题现状 #

通过Django接口上传PDF到S3，调用Textract的analyze_document接口提取表单键值对时，输出结果大量空键，仅能提取零散值，无法正确关联键与对应的值。

问题根源分析 #

原代码存在以下核心逻辑错误：

1. EntityTypes遍历逻辑错误：每个KEY_VALUE_SET的EntityTypes数组仅包含一个元素（要么是KEY，要么是VALUE），循环遍历会导致逻辑分支混乱。
2. Relationships处理不严谨：未按关联类型（CHILD/VALUE）筛选关系，盲目取第一个关联项，导致无法正确获取键/值的文本块。
3. 文本块匹配逻辑错误：处理VALUE时硬编码取第2个ID，无业务依据，且未考虑多块拼接的情况。
4. 键值对关联错误：每个KEY_VALUE_SET单独生成一个条目，未将KEY与对应的VALUE配对，导致大量空键空值。

修复后的代码 #

def extract_data_from_pdf(request):
    if "file" not in request.FILES:
        return Response({"error": "No file found."}, status=400)

    file_obj = request.FILES["file"]
    filename = file_obj.name

    s3_client.upload_fileobj(file_obj, "bucket-name", filename)

    response = textract_client.analyze_document(
        Document={
            "S3Object": {
                "Bucket": "bucket-name",
                "Name": filename,
            }
        },
        FeatureTypes=["FORMS"],
    )

    # 构建块ID到块对象的映射，避免重复遍历提升效率
    block_map = {block["Id"]: block for block in response["Blocks"]}
    extracted_data = {}

    for item in response["Blocks"]:
        if item["BlockType"] == "KEY_VALUE_SET":
            # 每个KEY_VALUE_SET仅对应一种实体类型，直接取第一个元素
            entity_type = item["EntityTypes"][0]
            
            if entity_type == "KEY":
                key_text = ""
                # 遍历KEY的子块，拼接键文本
                if "Relationships" in item:
                    for rel in item["Relationships"]:
                        if rel["Type"] == "CHILD":
                            for child_id in rel["Ids"]:
                                child_block = block_map.get(child_id)
                                if child_block and "Text" in child_block:
                                    key_text += child_block["Text"] + " "
                    key_text = key_text.strip()
                
                value_text = ""
                # 遍历KEY关联的VALUE块，获取对应值文本
                if "Relationships" in item:
                    for rel in item["Relationships"]:
                        if rel["Type"] == "VALUE":
                            for value_id in rel["Ids"]:
                                value_item = block_map.get(value_id)
                                if value_item and "Relationships" in value_item:
                                    # VALUE的文本存储在其子块中
                                    for val_rel in value_item["Relationships"]:
                                        if val_rel["Type"] == "CHILD":
                                            for val_child_id in val_rel["Ids"]:
                                                val_child_block = block_map.get(val_child_id)
                                                if val_child_block and "Text" in val_child_block:
                                                    value_text += val_child_block["Text"] + " "
                    value_text = value_text.strip()
            
                # 仅当键文本非空时存入结果
                if key_text:
                    extracted_data[key_text] = value_text

    # 转换为接口要求的列表格式
    result = [{"key": k, "value": v} for k, v in extracted_data.items()]
    return Response(result)

修复要点说明 #

• 块映射优化：用block_map缓存所有文本块，避免反复遍历response["Blocks"]，提升处理效率。
• 实体类型判断：直接取EntityTypes[0]判断当前KEY_VALUE_SET是键还是值，逻辑更清晰。
• 关联类型筛选：处理KEY时，通过CHILD类型关联获取键的文本，通过VALUE类型关联找到对应的值块；值的文本同样从其子块中拼接。
• 多块文本拼接：支持键或值由多个文本块组成的场景，避免丢失部分内容。
• 有效条目过滤：仅保留有键的条目，彻底解决空键问题，确保键值对正确关联。

内容的提问来源于stack exchange，提问作者ninad muranjan