), and others. Include category tags, and filter out irrelevant elements like headers and footers."}]}]
prompt_full = current_processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = current_processor(text=[prompt_full], images=[image], return_tensors="pt", padding=True).to(device)
with torch.no_grad():
generated_ids = current_model.generate(**inputs, max_new_tokens=2048, temperature=0.1, top_p=0.9, do_sample=True, repetition_penalty=1.05)
generated_ids_trimmed = [out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)]
output_text = current_processor.batch_decode(generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]
return output_text
def convert_file_to_images(file_path: str, dpi: int = 200) -> List[Image.Image]:
"""
Converts a PDF or image file into a list of PIL Images.
"""
images = []
file_ext = Path(file_path).suffix.lower()
if file_ext in image_suffixes:
images.append(Image.open(file_path).convert("RGB"))
return images
if file_ext not in pdf_suffixes:
raise ValueError(f"Unsupported file type: {file_ext}")
try:
pdf_document = fitz.open(file_path)
zoom = dpi / 72.0
mat = fitz.Matrix(zoom, zoom)
for page_num in range(len(pdf_document)):
page = pdf_document.load_page(page_num)
pix = page.get_pixmap(matrix=mat)
img_data = pix.tobytes("png")
images.append(Image.open(BytesIO(img_data)))
pdf_document.close()
except Exception as e:
logger.error(f"Failed to convert PDF using PyMuPDF: {e}")
raise
return images
def get_initial_state() -> Dict[str, Any]:
"""Returns the default initial state for the application."""
return {"pages": [], "total_pages": 0, "current_page_index": 0, "page_results": []}
def load_and_preview_file(file_path: Optional[str]) -> Tuple[Optional[Image.Image], str, Dict[str, Any]]:
"""
Loads a file, converts all pages to images, and stores them in the state.
"""
state = get_initial_state()
if not file_path:
return None, 'No file loaded
', state
try:
pages = convert_file_to_images(file_path)
if not pages:
return None, 'Could not load file
', state
state["pages"] = pages
state["total_pages"] = len(pages)
page_info_html = f'Page 1 / {state["total_pages"]}
'
return pages[0], page_info_html, state
except Exception as e:
logger.error(f"Failed to load and preview file: {e}")
return None, 'Failed to load preview
', state
async def process_all_pages(state: Dict[str, Any], model_choice: str):
"""
Processes all pages stored in the state and updates the state with results.
"""
if not state or not state["pages"]:
error_msg = "Please upload a file first.
"
return error_msg, "", "", None, "Error: No file to process", state
logger.info(f'Processing {state["total_pages"]} pages with model: {model_choice}')
start_time = time.time()
try:
page_results = []
for i, page_img in enumerate(state["pages"]):
logger.info(f"Parsing page {i + 1}/{state['total_pages']}")
html_result = parse_page(page_img, model_choice)
page_results.append({'raw_html': html_result})
state["page_results"] = page_results
# Create a single markdown file for download with all content
full_html_content = "\n\n".join([f'\n{res["raw_html"]}' for i, res in enumerate(page_results)])
full_markdown = html2text.html2text(full_html_content)
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f:
f.write(full_markdown)
md_path = f.name
parsing_time = time.time() - start_time
cost_time_str = f'Total processing time: {parsing_time:.2f}s'
# Display the results for the current page
current_page_results = get_page_outputs(state)
return *current_page_results, md_path, cost_time_str, state
except Exception as e:
logger.error(f"Parsing failed: {e}", exc_info=True)
error_html = f"An error occurred during processing:
{str(e)}
"
return error_html, "", "", None, f"Error: {str(e)}", state
def navigate_page(direction: str, state: Dict[str, Any]):
"""
Navigates to the previous or next page and updates the UI accordingly.
"""
if not state or not state["pages"]:
return None, 'No file loaded
', *get_page_outputs(state), state
current_index = state["current_page_index"]
total_pages = state["total_pages"]
if direction == "prev":
new_index = max(0, current_index - 1)
elif direction == "next":
new_index = min(total_pages - 1, current_index + 1)
else:
new_index = current_index
state["current_page_index"] = new_index
image_preview = state["pages"][new_index]
page_info_html = f'Page {new_index + 1} / {total_pages}
'
page_outputs = get_page_outputs(state)
return image_preview, page_info_html, *page_outputs, state
def get_page_outputs(state: Dict[str, Any]) -> Tuple[str, str, str]:
"""Helper to get displayable outputs for the current page."""
if not state or not state.get("page_results"):
return "Process the document to see results.
", "", ""
index = state["current_page_index"]
result = state["page_results"][index]
raw_html = result['raw_html']
mmd_source = html2text.html2text(raw_html)
mmd_render = markdown.markdown(mmd_source, extensions=['fenced_code', 'tables'])
return mmd_render, mmd_source, raw_html
def clear_all():
"""Clears all UI components and resets the state."""
return (
None,
None,
"Results will be displayed here after processing.
",
"",
"",
None,
"",
'No file loaded
',
get_initial_state()
)
@click.command()
def main():
"""
Sets up and launches the Gradio user interface for the Logics-Parsing app.
"""
css = """
.main-container { max-width: 1400px; margin: 0 auto; }
.header-text { text-align: center; color: #2c3e50; margin-bottom: 20px; }
.process-button { border: none !important; color: white !important; font-weight: bold !important; background-color: blue !important;}
.process-button:hover { background-color: darkblue !important; transform: translateY(-2px) !important; box-shadow: 0 4px 8px rgba(0,0,0,0.2) !important; }
.page-info { text-align: center; padding: 8px 16px; border-radius: 20px; font-weight: bold; margin: 10px 0; }
"""
with gr.Blocks(theme="bethecloud/storj_theme", css=css, title="Logics-Parsing Demo") as demo:
app_state = gr.State(value=get_initial_state())
gr.HTML("""
""")
with gr.Row(elem_classes=["main-container"]):
with gr.Column(scale=1):
model_choice = gr.Dropdown(choices=["Logics-Parsing", "Gliese-OCR-7B-Post1.0"], label="Select Model⚡️", value="Logics-Parsing")
file_input = gr.File(label="Upload PDF or Image", file_types=[".pdf", ".jpg", ".jpeg", ".png"], type="filepath")
image_preview = gr.Image(label="Preview", type="pil", interactive=False, height=280)
with gr.Row():
prev_page_btn = gr.Button("◀ Previous", size="md")
page_info = gr.HTML('No file loaded
')
next_page_btn = gr.Button("Next ▶", size="md")
example_root = "examples"
if os.path.exists(example_root) and os.path.isdir(example_root):
example_files = [os.path.join(example_root, f) for f in os.listdir(example_root) if f.endswith(tuple(pdf_suffixes + image_suffixes))]
if example_files:
with gr.Accordion("Open Examples⚙️", open=False):
gr.Examples(examples=example_files, inputs=file_input, examples_per_page=10)
with gr.Accordion("Download Details🕧", open=False):
output_file = gr.File(label='Download Markdown Result', interactive=False)
cost_time = gr.Text(label='Time Cost', interactive=False)
process_btn = gr.Button("🚀 Process Document", variant="primary", elem_classes=["process-button"], size="lg")
clear_btn = gr.Button("🗑️ Clear All", variant="secondary")
with gr.Column(scale=2):
with gr.Tabs():
with gr.Tab("Markdown Rendering"):
mmd_html = gr.TextArea(lines=27, label='Markdown Rendering', show_copy_button=True, interactive=True)
with gr.Tab("Markdown Source"):
mmd = gr.TextArea(lines=27, show_copy_button=True, label="Markdown Source", interactive=True)
with gr.Tab("Generated HTML"):
raw_html = gr.TextArea(lines=27, show_copy_button=True, label="Generated HTML")
# --- Event Handlers ---
file_input.change(
fn=load_and_preview_file,
inputs=file_input,
outputs=[image_preview, page_info, app_state],
show_progress="full")
process_btn.click(
fn=process_all_pages,
inputs=[app_state, model_choice],
outputs=[mmd_html, mmd, raw_html,
output_file, cost_time, app_state],
concurrency_limit=15,
show_progress="full")
prev_page_btn.click(
fn=lambda s: navigate_page("prev", s),
inputs=app_state, outputs=[image_preview,
page_info, mmd_html, mmd, raw_html, app_state])
next_page_btn.click(
fn=lambda s: navigate_page("next", s),
inputs=app_state, outputs=[image_preview,
page_info, mmd_html, mmd, raw_html, app_state])
clear_btn.click(
fn=clear_all,
outputs=[file_input, image_preview, mmd_html, mmd, raw_html,
output_file, cost_time, page_info, app_state])
demo.queue().launch(debug=True, show_error=True)
if __name__ == '__main__':
if not os.path.exists("examples"):
os.makedirs("examples")
logger.info("Created 'examples' directory. Please add some sample PDF/image files there.")
main()