本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 將資料表導出為 CSV 檔
這些 Python 範例示範如何將表格從文件圖像導出至以逗號分隔值 (CSV) 檔案。
同步文檔分析示例從調用[AnalyzeDocument](API_AnalyzeDocument.md)。異步文檔分析示例調用[StartDocumentAnalysis](API_StartDocumentAnalysis.md)，然後從[GetDocumentAnalysis](API_GetDocumentAnalysis.md)如`Block`物件。
表信息返回為[Block](API_Block.md)對象從調用[AnalyzeDocument](API_AnalyzeDocument.md)。如需詳細資訊，請參閱 [資料表](how-it-works-tables.md)。所以此`Block`對象存儲在用於將表數據導出到 CSV 文件的地圖結構中。
------
#### [ Synchronous ]
在此示例中，您將使用以下函數：
+ `get_table_csv_results`— 呼叫[AnalyzeDocument](API_AnalyzeDocument.md)，並構建在文檔中檢測到的表的映射。創建所有檢測到的表的 CSV 表示形式。
+ `generate_table_csv`— 為單個表生成 CSV 文件。
+ `get_rows_columns_map`— 從地圖中獲取行和列。
+ `get_text`— 從單元格獲取文本。
**將表導出到 CSV 文件**
1. 設定您的環境。如需詳細資訊，請參閱 [先決條件](examples-blocks.md#examples-prerequisites)。
1. 將下列範本程式碼複製到名為的*textract\$1python\$1table\$1parser.py*。
```
import webbrowser, os
import json
import boto3
import io
from io import BytesIO
import sys
from pprint import pprint
def get_rows_columns_map(table_result, blocks_map):
rows = {}
for relationship in table_result['Relationships']:
if relationship['Type'] == 'CHILD':
for child_id in relationship['Ids']:
cell = blocks_map[child_id]
if cell['BlockType'] == 'CELL':
row_index = cell['RowIndex']
col_index = cell['ColumnIndex']
if row_index not in rows:
# create new row
rows[row_index] = {}
# get the text value
rows[row_index][col_index] = get_text(cell, blocks_map)
return rows
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 get_table_csv_results(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
# get the results
client = boto3.client('textract')
response = client.analyze_document(Document={'Bytes': bytes_test}, FeatureTypes=['TABLES'])
# Get the text blocks
blocks=response['Blocks']
pprint(blocks)
blocks_map = {}
table_blocks = []
for block in blocks:
blocks_map[block['Id']] = block
if block['BlockType'] == "TABLE":
table_blocks.append(block)
if len(table_blocks) <= 0:
return " NO Table FOUND "
csv = ''
for index, table in enumerate(table_blocks):
csv += generate_table_csv(table, blocks_map, index +1)
csv += '\n\n'
return csv
def generate_table_csv(table_result, blocks_map, table_index):
rows = get_rows_columns_map(table_result, blocks_map)
table_id = 'Table_' + str(table_index)
# get cells.
csv = 'Table: {0}\n\n'.format(table_id)
for row_index, cols in rows.items():
for col_index, text in cols.items():
csv += '{}'.format(text) + ","
csv += '\n'
csv += '\n\n\n'
return csv
def main(file_name):
table_csv = get_table_csv_results(file_name)
output_file = 'output.csv'
# replace content
with open(output_file, "wt") as fout:
fout.write(table_csv)
# show the results
print('CSV OUTPUT FILE: ', output_file)
if __name__ == "__main__":
file_name = sys.argv[1]
main(file_name)
```
1. 在命令提示中，輸入以下命令：Replace`file`以您要分析的檔案圖像文件名稱取代。
```
python textract_python_table_parser.py file
```
運行該示例時，CSV 輸出將保存在名為`output.csv`。
------
#### [ Asynchronous ]
在本示例中，您將使用兩個不同的腳本。第一個腳本開始異步分析文檔的過程`StartDocumentAnalysis`並獲取`Block`返回的信息`GetDocumentAnalysis`。第二個腳本採用返回的`Block`信息，將數據格式化為表格，並將表保存到 CSV 文件中。
**將表導出到 CSV 文件**
1. 設定您的環境。如需詳細資訊，請參閱 [先決條件](examples-blocks.md#examples-prerequisites)。
1. 確保您已遵循[為異步操作配置 Amazon Textract](api-async-roles.md)。通過該頁上記錄的過程，您可以發送和接收有關異步作業完成狀態的消息。
1. 在下面的代碼示例中，替換`roleArn`並將 Arn 分配給您在步驟 2 中建立的角色。替換`bucket`以包含您的檔案的 S3 儲存體名稱取代。替換`document`取代為 S3 儲存貯體中的檔案名稱。替換`region_name`取代為儲存貯體的區域名稱。
將下列範本程式碼複製到名為的*start\$1doc\$1analysis\$1for\$1table\$1extraction.py。*。
```
import boto3
import time
class DocumentProcessor:
jobId = ''
region_name = ''
roleArn = ''
bucket = ''
document = ''
sqsQueueUrl = ''
snsTopicArn = ''
processType = ''
def __init__(self, role, bucket, document, region):
self.roleArn = role
self.bucket = bucket
self.document = document
self.region_name = region
self.textract = boto3.client('textract', region_name=self.region_name)
self.sqs = boto3.client('sqs')
self.sns = boto3.client('sns')
def ProcessDocument(self):
jobFound = False
response = self.textract.start_document_analysis(DocumentLocation={'S3Object': {'Bucket': self.bucket, 'Name': self.document}},
FeatureTypes=["TABLES", "FORMS"], NotificationChannel={'RoleArn': self.roleArn, 'SNSTopicArn': self.snsTopicArn})
print('Processing type: Analysis')
print('Start Job Id: ' + response['JobId'])
print('Done!')
def CreateTopicandQueue(self):
millis = str(int(round(time.time() * 1000)))
# Create SNS topic
snsTopicName = "AmazonTextractTopic" + millis
topicResponse = self.sns.create_topic(Name=snsTopicName)
self.snsTopicArn = topicResponse['TopicArn']
# create SQS queue
sqsQueueName = "AmazonTextractQueue" + millis
self.sqs.create_queue(QueueName=sqsQueueName)
self.sqsQueueUrl = self.sqs.get_queue_url(QueueName=sqsQueueName)['QueueUrl']
attribs = self.sqs.get_queue_attributes(QueueUrl=self.sqsQueueUrl,
AttributeNames=['QueueArn'])['Attributes']
sqsQueueArn = attribs['QueueArn']
# Subscribe SQS queue to SNS topic
self.sns.subscribe(TopicArn=self.snsTopicArn, Protocol='sqs', Endpoint=sqsQueueArn)
# Authorize SNS to write SQS queue
policy = """{{
"Version":"2012-10-17",
"Statement":[
{{
"Sid":"MyPolicy",
"Effect":"Allow",
"Principal" : {{"AWS" : "*"}},
"Action":"SQS:SendMessage",
"Resource": "{}",
"Condition":{{
"ArnEquals":{{
"aws:SourceArn": "{}"
}}
}}
}}
]
}}""".format(sqsQueueArn, self.snsTopicArn)
response = self.sqs.set_queue_attributes(
QueueUrl=self.sqsQueueUrl,
Attributes={
'Policy': policy
})
def main():
roleArn = 'role-arn'
bucket = 'bucket-name'
document = 'document-name'
region_name = 'region-name'
analyzer = DocumentProcessor(roleArn, bucket, document, region_name)
analyzer.CreateTopicandQueue()
analyzer.ProcessDocument()
if __name__ == "__main__":
main()
```
1. 執行程式碼。代碼將打印一個 JobId。將此 JobId 複製下來。
1. 等待您的作業完成處理，並在完成後，將以下代碼複製到名為*get\$1doc\$1analysis\$1for\$1table\$1extraction.py*。替換`jobId`使用您之前複製的 Job ID。替換`region_name`，其中包含與您的 Textract 角色關聯的區域的名稱。替換`file_name`取代為您想要指定輸出 CSV 的名稱。
```
import boto3
from pprint import pprint
jobId = 'job-id'
region_name = 'region-name'
file_name = "output-file-name.csv"
textract = boto3.client('textract', region_name=region_name)
# Display information about a block
def DisplayBlockInfo(block):
print("Block Id: " + block['Id'])
print("Type: " + block['BlockType'])
if 'EntityTypes' in block:
print('EntityTypes: {}'.format(block['EntityTypes']))
if 'Text' in block:
print("Text: " + block['Text'])
if block['BlockType'] != 'PAGE':
print("Confidence: " + "{:.2f}".format(block['Confidence']) + "%")
def GetResults(jobId, file_name):
maxResults = 1000
paginationToken = None
finished = False
while finished == False:
response = None
if paginationToken == None:
response = textract.get_document_analysis(JobId=jobId, MaxResults=maxResults)
else:
response = textract.get_document_analysis(JobId=jobId, MaxResults=maxResults,
NextToken=paginationToken)
blocks = response['Blocks']
table_csv = get_table_csv_results(blocks)
output_file = file_name
# replace content
with open(output_file, "at") as fout:
fout.write(table_csv)
# show the results
print('Detected Document Text')
print('Pages: {}'.format(response['DocumentMetadata']['Pages']))
print('OUTPUT TO CSV FILE: ', output_file)
# Display block information
for block in blocks:
DisplayBlockInfo(block)
print()
print()
if 'NextToken' in response:
paginationToken = response['NextToken']
else:
finished = True
def get_rows_columns_map(table_result, blocks_map):
rows = {}
for relationship in table_result['Relationships']:
if relationship['Type'] == 'CHILD':
for child_id in relationship['Ids']:
try:
cell = blocks_map[child_id]
if cell['BlockType'] == 'CELL':
row_index = cell['RowIndex']
col_index = cell['ColumnIndex']
if row_index not in rows:
# create new row
rows[row_index] = {}
# get the text value
rows[row_index][col_index] = get_text(cell, blocks_map)
except KeyError:
print("Error extracting Table data - {}:".format(KeyError))
pass
return rows
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']:
try:
word = blocks_map[child_id]
if word['BlockType'] == 'WORD':
text += word['Text'] + ' '
if word['BlockType'] == 'SELECTION_ELEMENT':
if word['SelectionStatus'] == 'SELECTED':
text += 'X '
except KeyError:
print("Error extracting Table data - {}:".format(KeyError))
return text
def get_table_csv_results(blocks):
pprint(blocks)
blocks_map = {}
table_blocks = []
for block in blocks:
blocks_map[block['Id']] = block
if block['BlockType'] == "TABLE":
table_blocks.append(block)
if len(table_blocks) <= 0:
return " NO Table FOUND "
csv = ''
for index, table in enumerate(table_blocks):
csv += generate_table_csv(table, blocks_map, index + 1)
csv += '\n\n'
return csv
def generate_table_csv(table_result, blocks_map, table_index):
rows = get_rows_columns_map(table_result, blocks_map)
table_id = 'Table_' + str(table_index)
# get cells.
csv = 'Table: {0}\n\n'.format(table_id)
for row_index, cols in rows.items():
for col_index, text in cols.items():
csv += '{}'.format(text) + ","
csv += '\n'
csv += '\n\n\n'
return csv
response_blocks = GetResults(jobId, file_name)
```
1. 執行程式碼。
獲得結果後，請務必刪除相關的 SNS 和 SQS 資源，否則您可能會為這些資源產生費用。
------