#!/usr/bin/python3 # -*- coding: utf-8 -*- import argparse import base64 import json import logging import os from pathlib import Path import platform import re import urllib import urllib.parse import zipfile import tempfile # os.environ["DOTNET_SYSTEM_GLOBALIZATION_INVARIANT"] = "1" # Process terminated. Couldn't find a valid ICU package installed on the system. Set the configuration flag System.Globalization.Invariant to true if you want to run with no globalization support. import difflib import gradio as gr import log from project_settings import project_path, log_directory from toolbox.os.command import Command from toolbox.to_markdown.base_to_markdown import BaseToMarkdown from toolbox.unstructured_eval.unstructured_eval import ChunkSearcher, ChunkSimilarity log.setup_size_rotating(log_directory=log_directory) logger = logging.getLogger("main") def get_args(): parser = argparse.ArgumentParser() parser.add_argument( "--to_md_example_file_dir", default=(project_path / "data/files").as_posix(), type=str ) parser.add_argument( "--chunk_similar_examples_json_file", default=(project_path / "examples_chunk_similar.json").as_posix(), type=str ) args = parser.parse_args() return args latex_delimiters = [ {"left": "$$", "right": "$$", "display": True}, {"left": '$', "right": '$', "display": False} ] def image_to_base64(image_path): with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode('utf-8') def replace_image_with_base64(markdown_text: str, image_dir: str): pattern1 = r'\!\[(?:.*?)\]\((.+?)\)' def replace(match): relative_path = match.group(1) relative_path = urllib.parse.unquote(relative_path) full_path = os.path.join(image_dir, relative_path) base64_image = image_to_base64(full_path) return f"![{relative_path}](data:image/jpeg;base64,{base64_image})" markdown_text = re.sub(pattern1, replace, markdown_text) pattern2 = '' def replace(match): relative_path = match.group(1) full_path = os.path.join(image_dir, relative_path) base64_image = image_to_base64(full_path) return f"![{relative_path}](data:image/jpeg;base64,{base64_image})" markdown_text = re.sub(pattern2, replace, markdown_text) return markdown_text def load_markdown_from_zip(filename: str) -> str: filename = Path(filename) unzip_dir = filename.parent / filename.stem with zipfile.ZipFile(filename.as_posix(), "r") as f: f.extractall(unzip_dir) md_file = unzip_dir / f"{filename.stem}.md" with open(md_file, "r", encoding="utf-8") as f: md_content = f.read() return md_content to_md_engine_map = { "default": { "xlsx": "pandas", "html": "html_markdownify", "pdf": "pymupdf4llm", "pptx": "pptx2md", "docx": "docx2md", }, "aliyun": { "pdf": "aliyun", "html": "aliyun", }, } def change_to_md_engine(engine: str): if engine not in to_md_engine_map.keys(): raise ValueError(f"Unsupported engine: {engine}") choices = to_md_engine_map[engine] choices = list(choices.keys()) return gr.Dropdown( choices=choices, value=choices[0], interactive=True, ) def run_to_markdown(filename: str, engine: str, extension): logger.info(f"engine: {engine}, extension: {extension}, filename: {filename}") tmp_dir = Path(tempfile.gettempdir()) / "document_loaders" tmp_dir.mkdir(parents=True, exist_ok=True) to_md_engine_name = to_md_engine_map[engine][extension] to_md: BaseToMarkdown = BaseToMarkdown.by_name(to_md_engine_name)(filename) output_zip_file = to_md.save_to_zip(output_dir=tmp_dir.as_posix()) text_content = load_markdown_from_zip(output_zip_file) image_dir = tmp_dir / Path(output_zip_file).stem md_content = replace_image_with_base64(text_content, image_dir=image_dir.as_posix()) return output_zip_file, md_content, text_content def run_chunk_similar(page_content: str, chunk: str, win_size_radio: float): searcher = ChunkSearcher() match_content = searcher.chunk_search( chunk, page_content, win_size_radio, ) # edit diff diff_list = difflib.Differ().compare(chunk, match_content) markdown_diff = "" for diff in diff_list: operation = diff[0] char = diff[-1] if operation == "-": if char in ["\t", "\n", "\r", "\f", "\v", "\u200B"]: char = f"{repr(char).strip("'")}{char}" markdown_diff += f"{char}" elif operation == "+": if char in ["\t", "\n", "\r", "\f", "\v", "\u200B"]: char = f"{repr(char).strip("'")}{char}" markdown_diff += f"{char}" else: markdown_diff += char # scores chunk_similarity = ChunkSimilarity() scores = chunk_similarity.similar(chunk, match_content) markdown_scores = "" for idx, score in enumerate(scores): metric, score, note = score row = f"| {metric} | {score} | {note} |\n" markdown_scores += row if idx == 0: row= "| --- | --- | --- |\n" markdown_scores += row return match_content, markdown_diff, markdown_scores def shell(cmd: str): return Command.popen(cmd) def main(): args = get_args() # examples to_md_example_file_dir = Path(args.to_md_example_file_dir) to_md_example_file_list = list() for filename in to_md_example_file_dir.glob("**/*.*"): filename = Path(filename).as_posix() to_md_example_file_list.append(filename) with open(args.chunk_similar_examples_json_file, "r", encoding="utf-8") as f: chunk_similar_examples = json.load(f) # ui with gr.Blocks() as blocks: with gr.Tabs(): with gr.TabItem("to_md"): with gr.Row(): with gr.Column(variant="panel", scale=5): engine_choices = list(to_md_engine_map.keys()) extension_choices = set() for engine in engine_choices: extension_choices.update(to_md_engine_map[engine].keys()) extension_choices = list(extension_choices) to_md_engine = gr.Dropdown(choices=engine_choices, value=engine_choices[0], label="engine") to_md_extension = gr.Dropdown(choices=extension_choices, value=extension_choices[0], label="extension") to_md_file = gr.File(value=None, label="file") with gr.Row(): to_md_button = gr.Button(value="convert", variant="primary") to_md_clear = gr.ClearButton(components=[to_md_file], value="clear") gr.Examples( examples=to_md_example_file_list, inputs=[to_md_file], ) with gr.Column(variant="panel", scale=5): to_md_output_file = gr.File(label="convert result", interactive=False) with gr.Tabs(): with gr.Tab("Markdown rendering"): to_md_md = gr.Markdown( label="Markdown rendering", height=900, show_copy_button=True, latex_delimiters=latex_delimiters, line_breaks=True ) with gr.Tab("Markdown text"): to_md_md_text = gr.TextArea(lines=45, show_copy_button=True) to_md_button.click( fn=run_to_markdown, inputs=[to_md_file, to_md_engine, to_md_extension], outputs=[to_md_output_file, to_md_md, to_md_md_text], ) to_md_clear.add([ to_md_file, to_md_engine, to_md_extension, to_md_output_file, to_md_md, to_md_md_text, ]) to_md_engine.change( change_to_md_engine, inputs=to_md_engine, outputs=to_md_extension, ) with gr.TabItem("chunk_similar"): with gr.Row(): with gr.Column(variant="panel", scale=5): cs_page_content = gr.TextArea( label="page_content", lines=18, max_lines=100, ) with gr.Column(variant="panel", scale=5): cs_chunk = gr.TextArea( label="chunk", ) cs_match = gr.TextArea( label="match", ) cs_diff = gr.Markdown( label="diff", latex_delimiters=latex_delimiters, line_breaks=True ) with gr.Column(variant="panel", scale=5): cs_win_size_radio = gr.Slider(minimum=1, maximum=3, value=1.6, step=0.1, label="win_size_radio") with gr.Row(): cs_button = gr.Button(value="run", variant="primary") cs_clear = gr.ClearButton(value="clear") cs_scores = gr.Markdown( label="scores", ) gr.Examples( examples=chunk_similar_examples, inputs=[cs_page_content, cs_chunk, cs_win_size_radio], outputs=[cs_match, cs_diff, cs_scores], fn=run_chunk_similar, ) cs_button.click( fn=run_chunk_similar, inputs=[cs_page_content, cs_chunk, cs_win_size_radio], outputs=[cs_match, cs_diff, cs_scores], ) cs_clear.add(components=[ cs_page_content, cs_chunk, cs_match, cs_scores ]) with gr.TabItem("shell"): shell_text = gr.Textbox(label="cmd") shell_button = gr.Button("run") shell_output = gr.Textbox(label="output") shell_button.click( shell, inputs=[ shell_text, ], outputs=[ shell_output ], ) # http://127.0.0.1:7860/ # blocks.queue().launch( share=False if platform.system() == "Windows" else False, # server_name="127.0.0.1" if platform.system() == "Windows" else "0.0.0.0", server_name="0.0.0.0", server_port=7860, ) return if __name__ == "__main__": main()