Trích Xuất Dữ Liệu Hóa Đơn Bằng AI OCR Với Script Python Đơn Giản

import json
import os
import re
import time
import requests
import flowhunt
from flowhunt.rest import ApiException
from pprint import pprint
from pdf2image import convert_from_path

def convert_pdf_to_image(path: str) -> None:
    """
    Convert a PDF file to images, storing each page as a JPEG.
    """
    images = convert_from_path(path)
    for i in range(len(images)):
        images[i].save('data/images/' + 'page' + str(i) + '.jpg', 'JPEG')

def extract_attachment_url(data_string):
    pattern = r'```flowhunt\n({.*})\n```'
    match = re.search(pattern, data_string, re.DOTALL)
    if match:
        json_string = match.group(1)
        try:
            json_data = json.loads(json_string)
            return json_data.get('download_link', None)
        except json.JSONDecodeError:
            print("Error: Failed to decode JSON.")
            return None
    return None

convert_pdf_to_image("data/test.pdf")
FLOW_ID = "<FLOW_ID_HERE>"

configuration = flowhunt.Configuration(
    host="https://api.flowhunt.io",
    api_key={"APIKeyHeader": "<API_KEY_HERE>"}
)

with flowhunt.ApiClient(configuration) as api_client:
    auth_api = flowhunt.AuthApi(api_client)
    api_response = auth_api.get_user()
    workspace_id = api_response.api_key_workspace_id

flows_api = flowhunt.FlowsApi(api_client)
from_flow_create_session_req = flowhunt.FlowSessionCreateFromFlowRequest(flow_id=FLOW_ID)
create_session_rsp = flows_api.create_flow_session(workspace_id, from_flow_create_session_req)

for image in os.listdir("data/images"):
    image_name, image_extension = os.path.splitext(image)
    with open("data/images/" + image, "rb") as file:
        try:
            flow_sess_attachment = flows_api.upload_attachments(
                create_session_rsp.session_id,
                file.read()
            )

invoke_rsp = flows_api.invoke_flow_response(
    create_session_rsp.session_id, 
    flowhunt.FlowSessionInvokeRequest(message="")
)
while True:
    get_flow_rsp = flows_api.poll_flow_response(
        create_session_rsp.session_id, invoke_rsp.message_id
    )
    print("Flow response: ", get_flow_rsp)
    if get_flow_rsp.response_status == "S":
        print("done OCR")
        break
    time.sleep(3)

attachment_url = extract_attachment_url(get_flow_rsp.final_response[0])
if attachment_url:
    response = requests.get(attachment_url)
    with open("data/results/" + image_name + ".csv", "wb") as file:
        file.write(response.content)

import json
import os
import re
import time
import requests
import flowhunt
from flowhunt.rest import ApiException
from pprint import pprint
from pdf2image import convert_from_path

def convert_pdf_to_image(path: str) -> None:
    """
    Convert a pdf file to an image
    :return:
    """
    images = convert_from_path(path)
    for i in range(len(images)):
        images[i].save('data/images/' + 'page'+ str(i) +'.jpg', 'JPEG')

def extract_attachment_url(data_string):
    pattern = r'```flowhunt\n({.*})\n```'
    match = re.search(pattern, data_string, re.DOTALL)
    if match:
        json_string = match.group(1)
        try:
            json_data = json.loads(json_string)
            return json_data.get('download_link', None)
        except json.JSONDecodeError:
            print("Error: Failed to decode JSON.")
            return None
    return None

convert_pdf_to_image("data/test.pdf")
FLOW_ID = "<FLOW_ID_HERE>"

configuration = flowhunt.Configuration(host = "https://api.flowhunt.io",
                                       api_key = {"APIKeyHeader": "<API_KEY_HERE>"})

with flowhunt.ApiClient(configuration) as api_client:
    auth_api = flowhunt.AuthApi(api_client)
    api_response = auth_api.get_user()
    workspace_id = api_response.api_key_workspace_id

    flows_api = flowhunt.FlowsApi(api_client)
    from_flow_create_session_req = flowhunt.FlowSessionCreateFromFlowRequest(
        flow_id=FLOW_ID
    )
    create_session_rsp = flows_api.create_flow_session(workspace_id, from_flow_create_session_req)

    for image in os.listdir("data/images"):
        image_name, image_extension = os.path.splitext(image)
        with open("data/images/" + image, "rb") as file:
            try:
                flow_sess_attachment = flows_api.upload_attachments(
                    create_session_rsp.session_id,
                    file.read()
                )
                invoke_rsp = flows_api.invoke_flow_response(create_session_rsp.session_id, flowhunt.FlowSessionInvokeRequest(
                    message="",
                ))
                while True:
                    get_flow_rsp = flows_api.poll_flow_response(create_session_rsp.session_id, invoke_rsp.message_id)
                    print("Flow response: ", get_flow_rsp)
                    if get_flow_rsp.response_status == "S":
                        print("done OCR")
                        attachment_url = extract_attachment_url(get_flow_rsp.final_response[0])
                        if attachment_url:
                            print("Attachment URL: ", attachment_url, "\n Downloading the file...")
                            response = requests.get(attachment_url)
                            with open("data/results/" + image_name + ".csv", "wb") as file:
                                file.write(response.content)
                        break
                    time.sleep(3)
            except ApiException as e:
                print("error for file ", image)
                print(e)