基于AWS Textract表单KV输出创建单行DataFrame并入库

我用AWS Textract分析表单文档，得到了键值对结果。现在需要把这些键作为DataFrame的列标题，对应的值放到单行里，之后提取部分值插入数据库。我是编程新手，求指导。

import boto3
import sys
import re
import json
from collections import defaultdict


def get_kv_map(file_name):
    with open(file_name, 'rb') as file:
        img_test = file.read()
        bytes_test = bytearray(img_test)
        print('Image loaded', file_name)

    # process using image bytes
    client = boto3.client('textract')
    response = client.analyze_document(Document={'Bytes': bytes_test}, FeatureTypes=['FORMS'])

    # Get the text blocks
    blocks = response['Blocks']

    # get key and value maps
    key_map = {}
    value_map = {}
    block_map = {}
    for block in blocks:
        block_id = block['Id']
        block_map[block_id] = block
        if block['BlockType'] == "KEY_VALUE_SET":
            if 'KEY' in block['EntityTypes']:
                key_map[block_id] = block
            else:
                value_map[block_id] = block

    return key_map, value_map, block_map


def get_kv_relationship(key_map, value_map, block_map):
    kvs = defaultdict(list)
    for block_id, key_block in key_map.items():
        value_block = find_value_block(key_block, value_map)
        key = get_text(key_block, block_map)
        val = get_text(value_block, block_map)
        kvs[key].append(val)
    return kvs


def find_value_block(key_block, value_map):
    for relationship in key_block['Relationships']:
        if relationship['Type'] == 'VALUE':
            for value_id in relationship['Ids']:
                value_block = value_map[value_id]
    return value_block


def get_text(result, blocks_map):
    text = ''
    if 'Relationships' in result:
        for relationship in result['Relationships']:
            if relationship['Type'] == 'CHILD':
                for child_id in relationship['Ids']:
                    word = blocks_map[child_id]
                    if word['BlockType'] == 'WORD':
                        text += word['Text'] + ' '
                    if word['BlockType'] == 'SELECTION_ELEMENT':
                        if word['SelectionStatus'] == 'SELECTED':
                            text += 'X '

    return text


def print_kvs(kvs):
    for key, value in kvs.items():
        print(key, ":", value)


def search_value(kvs, search_key):
    for key, value in kvs.items():
        if re.search(search_key, key, re.IGNORECASE):
            return value


def main(file_name):
    key_map, value_map, block_map = get_kv_map(file_name)

    # Get Key Value relationship
    kvs = get_kv_relationship(key_map, value_map, block_map)
    print("\n\n== FOUND KEY : VALUE pairs ===\n")
    print_kvs(kvs)



if __name__ == "__main__":
    file_name = sys.argv[1]
    main(file_name)

== FOUND KEY : VALUE pairs ===

    Social#:  : ['XXXX ']
    Landlord Name:  : ['XXXX Trust ']
    Date:  : ['']
    Last Name:  : ['XXX ']
    Date of Birth:  : ['04/30/1959 ', '05/31/1955 ']
    State:  : ['CA ', 'CA ', 'CA ']
    Zip Code:  : ['92860 ', '92860 ', '92860 ']
    City:  : ['Norco ', 'Norco ', 'Norco ']
    Phone #:  : ['XXXX ', 'XXXX ', 'XXX ']
    Landlord Contact#:  : ['XXXX']
    Social Number/SIN#:  : ['XXXX ']
    Signature:  : ['']
    Primary Contact Number:  : ['XXXX ']
    Street Address:  : ['684 XXXX ']
    Physical Location Phone #:  : ['XXXXXXX ']
    Business Physical Street Address:  : ['684 XXXX ']
    Business Start Date of Current Ownership:  : ['05/01/2022 ']
    Amount Requested:  : ['550000.00 ']
    Home Address:  : ['1400 EXXXXX, 2201, Long Beach, CA 90802 ']
    Yes  : ['', 'X ', '', '']
    Industry Type: (Describe)  : ['manufacturing ']
    Business Legal Name:  : ['XXXXXX, LLC ']
    Business Trade Reference #3:  : ['XXXX ']
    Owner First Name  : ['Jim ']
    Billing Location Phone #:  : ['XXXX ']
    Est. Credit Score:  : ['700 ']
    LLC  : ['X ']
    Billing Street Address:  : ['684 XXXXX ']
    No  : ['', 'X ', 'X ', 'X ']
    Rented  : ['X ']
    Use of Proceeds:  : ['XXXXX ']
    Business Tax ID:  : ['XXXX ']
    Agent Name:  : ['']
    Date (M/D/YY)  : ['Sep 01 2022 ']
    Fax Number #:  : ['']
    Current Estimated MCA/ LINE OF CREDIT Balance  : ['350000.00 ']
    Credit Card Processing?  : ['X Yes No ']
    Business Trade Reference #1  : ['XXXXX']
    Business Website Address:  : ['']
    Mortgaged  : ['']
    Gross Annual Sales (from previous year's Tax return):  : ['2900000.00 ']
    Owner / Officer's Signature:  : ['']
    Business DBA Name:  : ['same ']
    Business Trade Reference #2:  : ['UPS ']
    State of Incorporation  : ['XXXX ']
    Ownership Percentage %  : ['37 ']
    Monthly Payment:  : ['7745.00 ']
    Name of Credit Card Processor  : ['XXXX']
    Owner / Officer's Name: (Print)  : ['XXXX ']
    LLP  : ['']
    Second Owner:  : ['XXXX ']
    Merchant Email Address  : ['XXXXX ']
    Fax:  : ['XXXXX ']

1. 清理并格式化KV数据 #

原输出的键存在多余空格、冒号，值多为列表类型（部分包含多个元素），先做清洗：

• 去除键的首尾空格和末尾多余冒号
• 将列表值合并为字符串（多值用逗号分隔）

在原代码基础上，修改main函数添加清洗逻辑：

import pandas as pd  # 新增导入pandas

def main(file_name):
    key_map, value_map, block_map = get_kv_map(file_name)

    # Get Key Value relationship
    kvs = get_kv_relationship(key_map, value_map, block_map)
    print("\n\n== FOUND KEY : VALUE pairs ===\n")
    print_kvs(kvs)
    
    # 清洗KV数据
    cleaned_kvs = {}
    for key, values in kvs.items():
        # 清理键格式
        cleaned_key = key.strip().rstrip(':')
        # 合并列表值为字符串，空值保留
        cleaned_value = ', '.join([v.strip() for v in values if v.strip()]) if any(v.strip() for v in values) else ''
        cleaned_kvs[cleaned_key] = cleaned_value
    
    print("\n\n== 清洗后的KEY : VALUE pairs ===\n")
    for key, value in cleaned_kvs.items():
        print(f"{key}: {value}")

2. 转换为Pandas DataFrame #

将清洗后的字典转换为单行DataFrame，直接在main函数中添加：

# 转换为DataFrame
    df = pd.DataFrame([cleaned_kvs])
    print("\n\n== 生成的DataFrame ===")
    print(df)

3. 提取部分数据插入数据库 #

以SQLite为例，演示插入指定字段的逻辑，新增插入函数并调用：

import sqlite3  # 新增导入

def insert_to_db(df, db_name='form_data.db', table_name='form_records'):
    # 连接数据库（不存在则自动创建）
    conn = sqlite3.connect(db_name)
    # 选择需要插入的字段，可按需调整
    selected_cols = ['Business Legal Name', 'Amount Requested', 'Gross Annual Sales (from previous year\'s Tax return)', 'Owner First Name']
    df_selected = df[selected_cols]
    # 插入数据（表不存在则自动创建）
    df_selected.to_sql(table_name, conn, if_exists='append', index=False)
    # 关闭连接
    conn.close()
    print(f"\n数据已成功插入数据库 {db_name} 的 {table_name} 表")

# 在main函数末尾调用
insert_to_db(df)

import boto3
import sys
import re
import json
from collections import defaultdict
import pandas as pd
import sqlite3


def get_kv_map(file_name):
    with open(file_name, 'rb') as file:
        img_test = file.read()
        bytes_test = bytearray(img_test)
        print('Image loaded', file_name)

    # process using image bytes
    client = boto3.client('textract')
    response = client.analyze_document(Document={'Bytes': bytes_test}, FeatureTypes=['FORMS'])

    # Get the text blocks
    blocks = response['Blocks']

    # get key and value maps
    key_map = {}
    value_map = {}
    block_map = {}
    for block in blocks:
        block_id = block['Id']
        block_map[block_id] = block
        if block['BlockType'] == "KEY_VALUE_SET":
            if 'KEY' in block['EntityTypes']:
                key_map[block_id] = block
            else:
                value_map[block_id] = block

    return key_map, value_map, block_map


def get_kv_relationship(key_map, value_map, block_map):
    kvs = defaultdict(list)
    for block_id, key_block in key_map.items():
        value_block = find_value_block(key_block, value_map)
        key = get_text(key_block, block_map)
        val = get_text(value_block, block_map)
        kvs[key].append(val)
    return kvs


def find_value_block(key_block, value_map):
    for relationship in key_block['Relationships']:
        if relationship['Type'] == 'VALUE':
            for value_id in relationship['Ids']:
                value_block = value_map[value_id]
    return value_block


def get_text(result, blocks_map):
    text = ''
    if 'Relationships' in result:
        for relationship in result['Relationships']:
            if relationship['Type'] == 'CHILD':
                for child_id in relationship['Ids']:
                    word = blocks_map[child_id]
                    if word['BlockType'] == 'WORD':
                        text += word['Text'] + ' '
                    if word['BlockType'] == 'SELECTION_ELEMENT':
                        if word['SelectionStatus'] == 'SELECTED':
                            text += 'X '

    return text


def print_kvs(kvs):
    for key, value in kvs.items():
        print(key, ":", value)


def search_value(kvs, search_key):
    for key, value in kvs.items():
        if re.search(search_key, key, re.IGNORECASE):
            return value

def insert_to_db(df, db_name='form_data.db', table_name='form_records'):
    # 连接数据库（不存在则自动创建）
    conn = sqlite3.connect(db_name)
    # 选择需要插入的字段，可按需调整
    selected_cols = ['Business Legal Name', 'Amount Requested', 'Gross Annual Sales (from previous year\'s Tax return)', 'Owner First Name']
    df_selected = df[selected_cols]
    # 插入数据（表不存在则自动创建）
    df_selected.to_sql(table_name, conn, if_exists='append', index=False)
    # 关闭连接
    conn.close()
    print(f"\n数据已成功插入数据库 {db_name} 的 {table_name} 表")


def main(file_name):
    key_map, value_map, block_map = get_kv_map(file_name)

    # Get Key Value relationship
    kvs = get_kv_relationship(key_map, value_map, block_map)
    print("\n\n== FOUND KEY : VALUE pairs ===\n")
    print_kvs(kvs)
    
    # 清洗KV数据
    cleaned_kvs = {}
    for key, values in kvs.items():
        # 清理键格式
        cleaned_key = key.strip().rstrip(':')
        # 合并列表值为字符串，空值保留
        cleaned_value = ', '.join([v.strip() for v in values if v.strip()]) if any(v.strip() for v in values) else ''
        cleaned_kvs[cleaned_key] = cleaned_value
    
    print("\n\n== 清洗后的KEY : VALUE pairs ===\n")
    for key, value in cleaned_kvs.items():
        print(f"{key}: {value}")
    
    # 转换为DataFrame
    df = pd.DataFrame([cleaned_kvs])
    print("\n\n== 生成的DataFrame ===")
    print(df)
    
    # 插入数据库
    insert_to_db(df)


if __name__ == "__main__":
    file_name = sys.argv[1]
    main(file_name)

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