本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 將資料表匯出至 CSV 檔案
<a name="examples-export-table-csv"></a>

這些 Python 範例示範如何將資料表從文件的映像匯出至逗號分隔值 (CSV) 檔案。

同步文件分析的範例會從對 [AnalyzeDocument](https://docs.aws.amazon.com/textract/latest/APIReference/API_AnalyzeDocument.html) 的呼叫收集資料表資訊。非同步文件分析的範例會呼叫 [StartDocumentAnalysis](https://docs.aws.amazon.com/textract/latest/APIReference/API_StartDocumentAnalysis.html)，然後從 [GetDocumentAnalysis](https://docs.aws.amazon.com/textract/latest/APIReference/API_GetDocumentAnalysis.html) 擷取結果做為`Block`物件。

資料表資訊會以[封鎖](https://docs.aws.amazon.com/textract/latest/APIReference/API_Block.html)物件的形式傳回給 [AnalyzeDocument](https://docs.aws.amazon.com/textract/latest/APIReference/API_AnalyzeDocument.html)。如需詳細資訊，請參閱[表格](how-it-works-tables.md)。`Block` 物件存放在映射結構中，用於將資料表資料匯出至 CSV 檔案。

------
#### [ Synchronous ]

在此範例中，您將使用 函數：
+ `get_table_csv_results` – 呼叫 [AnalyzeDocument](https://docs.aws.amazon.com/textract/latest/APIReference/API_AnalyzeDocument.html)，並建置文件中偵測到的資料表映射。建立所有偵測到資料表的 CSV 表示法。
+ `generate_table_csv` – 產生個別資料表的 CSV 檔案。
+ `get_rows_columns_map` – 從地圖取得資料列和資料欄。
+ `get_text` – 從儲存格取得文字。

**將資料表匯出至 CSV 檔案**

1. 設定您的環境。如需詳細資訊，請參閱[先決條件](examples-blocks.md#examples-prerequisites)。

1. 將下列範例程式碼儲存至名為 *textract\_python\_table\_parser.py *的檔案。在函數 中`get_table_csv_results`，`profile-name`將 取代為可擔任角色的設定檔名稱，並將 `region`取代為您要執行程式碼的區域。

   ```
   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 = {}
       scores = []
       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 confidence score
                       scores.append(str(cell['Confidence']))
                           
                       # get the text value
                       rows[row_index][col_index] = get_text(cell, blocks_map)
       return rows, scores
   
   
   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':
                           if "," in word['Text'] and word['Text'].replace(",", "").isnumeric():
                               text += '"' + word['Text'] + '"' + ' '
                           else:
                               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
       session = boto3.Session(profile_name='profile-name')
       client = session.client('textract', region_name='region')
       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 "<b> NO Table FOUND </b>"
   
       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, scores = 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():
               col_indices = len(cols.items())
               csv += '{}'.format(text) + ","
           csv += '\n'
           
       csv += '\n\n Confidence Scores % (Table Cell) \n'
       cols_count = 0
       for score in scores:
           cols_count += 1
           csv += score + ","
           if cols_count == col_indices:
               csv += '\n'
               cols_count = 0
   
       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. 在命令提示字元中，輸入下列命令。`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`為您在步驟 2 中建立的角色指派的 Arn。將 的值取代`bucket`為包含 文件的 S3 儲存貯體名稱。將 的值取代`document`為 S3 儲存貯體中的文件名稱。將 的值取代`region_name`為您儲存貯體區域的名稱。

   將下列範例程式碼儲存至名為 *start\_doc\_analysis\_for\_table\_extraction.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\_doc\_analysis\_for\_table\_extraction.py *的檔案。將 的值取代`jobId`為您先前複製的任務 ID。將 的值取代`region_name`為與您的 Textract 角色相關聯的區域名稱。將 的值取代`file_name`為您要提供輸出 CSV 的名稱。

   ```
   import boto3
   from pprint import pprint
   
   jobId = ''
   region_name = ''
   file_name = ''
   
   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 + ".csv"
           # 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 "<b> NO Table FOUND </b>"
   
       csv = ''
       for index, table in enumerate(table_blocks):
           csv += generate_table_csv(table, blocks_map, index + 1)
           csv += '\n\n'
           # In order to generate separate CSV file for every table, uncomment code below
           #inner_csv = ''
           #inner_csv += generate_table_csv(table, blocks_map, index + 1)
           #inner_csv += '\n\n'
           #output_file = file_name + "___" + str(index) + ".csv"
           # replace content
           #with open(output_file, "at") as fout:
           #    fout.write(inner_csv)
   
       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 資源，否則可能會產生這些資源的費用。

------