| import openai | |
| import gradio as gr | |
| import json | |
| import time | |
| from tqdm import tqdm | |
| import logging | |
| import requests | |
| import google | |
| from ocr_functions import detect_document, detect_image | |
| from ai_functions import chat_gpt_document, chat_gpt_image | |
| from helpers import save_json | |
| logging.basicConfig(filename='app.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| def retry_unprocessed_documents(): | |
| global global_document_type | |
| global global_context | |
| global unprocessed_documents | |
| if unprocessed_documents: | |
| output = batch_document(unprocessed_documents, global_document_type, global_context, "None") | |
| unprocessed_documents = [] | |
| return output | |
| else: | |
| unprocessed_documents = [] | |
| return save_json("No Unprocessed Documents", "No Unprocessed Documents") | |
| def combine_json_files(json_files, progress=gr.Progress()): | |
| combined_data = [] | |
| progress(0, desc="Starting") | |
| for file in progress.tqdm(json_files, desc="Combining JSON Files"): | |
| with open(file.name, 'r') as json_file: | |
| data = json.load(json_file) | |
| combined_data.extend(data) | |
| logging.info("Combined JSON File: ", combined_data) | |
| return save_json(combined_data, "Combined Json") | |
| unprocessed_documents = [] | |
| global_document_type = None | |
| global_context = None | |
| def batch_document(content, document_type, context, progress = gr.Progress()): | |
| logging.info(content) | |
| combined_data = [] | |
| global global_document_type | |
| global global_context | |
| global_document_type = document_type | |
| global_context = context | |
| if progress == "None": | |
| for x in content: | |
| retries = 1 | |
| timeout = 1 | |
| i = 0 | |
| while True: | |
| try: | |
| data = json.loads(chat_gpt_document(detect_document(x),document_type,context)) | |
| combined_data.append(data) | |
| break | |
| except (openai.error.APIConnectionError, openai.error.AuthenticationError, openai.error.RateLimitError, google.api_core.exceptions.RetryError, requests.exceptions.RequestException) as e: | |
| logging.error(f'Retry {i+1} failed: {e}') | |
| if i < retries - 1: | |
| logging.error(f'Retrying in {timeout} seconds...') | |
| time.sleep(timeout) | |
| i += 1 | |
| else: | |
| unprocessed_documents.append(x) | |
| break | |
| else: | |
| progress(0, desc="Starting") | |
| for x in progress.tqdm(content, desc="Processing"): | |
| retries = 1 | |
| timeout = 1 | |
| i = 0 | |
| while True: | |
| try: | |
| data = json.loads(chat_gpt_document(detect_document(x),document_type,context)) | |
| combined_data.append(data) | |
| break | |
| except (openai.error.APIConnectionError, openai.error.AuthenticationError, openai.error.RateLimitError, google.api_core.exceptions.RetryError, requests.exceptions.RequestException) as e: | |
| logging.error(f'Retry {i+1} failed: {e}') | |
| if i < retries - 1: | |
| logging.error(f'Retrying in {timeout} seconds...') | |
| time.sleep(timeout) | |
| i += 1 | |
| else: | |
| unprocessed_documents.append(x) | |
| break | |
| logging.info(combined_data) | |
| logging.info(unprocessed_documents) | |
| if document_type == "": | |
| document_type = "error" | |
| return save_json(combined_data, document_type) | |
| def image(content, context): | |
| return chat_gpt_image(detect_image(content), context) | |
| def document(content, document_type, context): | |
| return chat_gpt_document(detect_document(content),document_type,context) | |
| with gr.Blocks(title="Axon OCR", css=".markdown {text-align: center;}") as app: | |
| gr.Markdown("""# Axon OCR | |
| Attach Images or Files below and convert them to Text.""", elem_classes="markdown") | |
| with gr.Tab("Scan Image"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| image_input = [gr.Image(type="pil"), | |
| gr.Textbox(label="What kind of Image is this? (Optional)", placeholder="This is an image of an Official Reciept")] | |
| image_output = gr.Textbox(label="Result") | |
| image_button = gr.Button("Scan") | |
| with gr.Tab("Scan Document"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| document_input = [gr.File(file_types=["pdf","tiff","image","text"]), | |
| gr.Dropdown(["RPFAA Building P1", "RPFAA Building P2", "TDRP"], label="File Type", info="What type of document is this?"), | |
| gr.Textbox(label="Any additional information? (Optional)", placeholder="This is document is an Official Reciept")] | |
| document_output = gr.Textbox(label="Result") | |
| document_button = gr.Button("Scan") | |
| with gr.Tab("Batch Scan"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| batch_document_input = [gr.File(file_types=["pdf","tiff","image","text"], file_count="multiple"), | |
| gr.Dropdown(["RPFAA Building P1", "RPFAA Building P2", "TDRP"], label="File Type", info="What type of document is this?"), | |
| gr.Textbox(label="Any additional information? (Optional)", placeholder="This is document is an Official Reciept")] | |
| batch_document_output = gr.File(label="Result") | |
| batch_document_button = gr.Button("Scan") | |
| with gr.Row(): | |
| with gr.Column(): | |
| retry_button = gr.Button("Retry Unprocessed Documents", label="Retry") | |
| with gr.Column(): | |
| stop_button = gr.Button("Stop Processing Document", label="Stop") | |
| with gr.Tab("Combine JSON"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| json_files_input = gr.File(file_types=[".json"], file_count="multiple", label='Upload JSON files') | |
| combined_json_output = gr.File(label="Result") | |
| combine_button = gr.Button('Combine JSON files') | |
| image_button.click(image, inputs=image_input, outputs=image_output) | |
| document_button.click(document, inputs=document_input, outputs=document_output) | |
| batch_document_event = batch_document_button.click(batch_document, inputs=batch_document_input, outputs=batch_document_output) | |
| retry_button.click(retry_unprocessed_documents, outputs=batch_document_output) | |
| stop_button.click(fn=None, inputs=None, outputs=None, cancels=[batch_document_event]) | |
| combine_button.click(combine_json_files, inputs=json_files_input, outputs=combined_json_output) | |
| app.queue() | |
| app.launch(auth=("username", "password")) | |