import gradio as gr import os import random import string from PIL import Image from src.GameMaster import GameMaster from collections import deque import time import threading import queue from src.Founder import Founder import os # os.environ['HTTP_PROXY'] = 'http://localhost:8234' # os.environ['HTTPS_PROXY'] = 'http://localhost:8234' game_master = GameMaster() founder_base = Founder() blank_image_path = "datas/blank_item.jpg" random_data = game_master.random_image_text_data( 12 ) recent_generated_items = deque(random_data , maxlen=12) def refresh_contribution_ladder(): top_contributors = founder_base.get_top_rank(top_k=20) contributions = [] for founder_name, items in top_contributors: random_items = random.sample(items, min(5, len(items))) cultivation_names = [] for item in random_items: result = game_master.remote.search_by_en_keyword(item) if result is not None and "name_in_cultivation" in result: cultivation_names.append(result['name_in_cultivation']) else: cultivation_names.append(item) items_description = ', '.join(cultivation_names) contribution = f"""道友姓名: {founder_name}\n发现了{items_description}等物品""" contributions.append(contribution) while len(contributions) < 20: contributions.append("") return contributions def expensive_generating(save_path, image_feature, backup_results, game_master, founder_name = None): # for now it's a idle implementation for debug # re-search incase redundant generate image_search_result = game_master.remote.top_k_search(image_feature, top_k=1) if image_search_result and len(image_search_result)>0 and image_search_result[0]['similarity'] > game_master.minimal_image_threshold: return global recent_generated_items recent_generated_items.append((save_path, "天机阁长老正在鉴定，道友请耐心等待。。")) search_result, backup_results, image_feature = game_master.search_with_path(save_path) if search_result is None: # real implementation should be cultivation_data = game_master.generate_cultivation_data( \ save_path, image_feature, backup_results ) description = cultivation_data["description_in_cultivation"] print("鉴定得到新物品:", description) else: cultivation_data = search_result description = cultivation_data["description_in_cultivation"] print("鉴定这是记载过的物品，但是第一次发现:", description) suffix = "" if founder_name is not None: global founder_base if "translated_word" in cultivation_data: translated_word = cultivation_data["translated_word"] else: translated_word = cultivation_data["name_in_cultivation"] if founder_base.get_founder(translated_word) is None: if founder_name.strip() != "": founder_base.set_founder(translated_word, founder_name) suffix = f" 由道友 {founder_name} 发现" else: suffix = f" 由道友 {founder_base.get_founder(translated_word)} 发现" # this function will automatically update the database # cultivation_data = backup_results[0] # recent_generated_items.append((save_path, description)) # replace the same save_path in recent_generated_items flag = True for index, item in enumerate(recent_generated_items): if item[0] == save_path: if "name_in_cultivation" in cultivation_data: description = cultivation_data["name_in_cultivation"] + suffix + "--" + description recent_generated_items[index] = (save_path, description) flag = False if flag: if "name_in_cultivation" in cultivation_data: description = cultivation_data["name_in_cultivation"] + suffix + "--" + description recent_generated_items.append((save_path, description)) return ### # Queue for managing tasks to process expensive operations task_queue = queue.Queue() # Function that processes the tasks in the queue def worker(): while True: save_path, image_feature, backup_results, founder_base = task_queue.get() if save_path is None: break expensive_generating(save_path, image_feature, backup_results, game_master, founder_base) task_queue.task_done() # Start the worker thread thread = threading.Thread(target=worker, daemon=True) thread.start() ### def similarity2level( sim, max_val , min_val ): level_num = 3 level = int( (sim - min_val) / (max_val - min_val) * level_num ) level = max(0, min(level_num, level)) return level from src.get_comments_from_level import get_comments_from_level def is_empty_name( founder_name ): empty_names = ["","测试","鲁鲁道祖"] if founder_name.strip() in empty_names: return True # Function to handle image upload and search def process_image(image, founder_name): # Ensure temp_images directory exists os.makedirs('temp_images', exist_ok=True) prefix = "" founder_name = founder_name.strip() if is_empty_name(founder_name): prefix = "道友，请在下面留下您的尊姓大名，提供大量珍宝鉴定的道友，天机阁将在后面送出灵泉玉液（珍珠奶茶）。" try: # Generate a random hash name for the image random_name = ''.join(random.choices(string.ascii_lowercase + string.digits, k=12)) + '.jpg' save_path = os.path.join('temp_images', random_name) # Convert numpy.ndarray to PIL.Image img = Image.fromarray(image) # Resize image to height not exceeding 480 pixels while maintaining aspect ratio img.thumbnail((img.width, 480)) img.save(save_path) except: return "","",prefix + "道友你有什么物品要来鉴定吗？" search_result, backup_results, image_feature = game_master.search_with_path(save_path, threshold = 0) image_similarity = search_result['similarity'] print("image_similarity:", image_similarity) inbase_similarity_level = similarity2level(image_similarity, max_val=0.99, min_val=0.80) suffix = "" if image_similarity > game_master.minimal_image_threshold: result = search_result # remove the temp image # 一般来说这个时候inbase_similarity_level是2或者3 inlibrary_similarity_level = inbase_similarity_level if "translated_word" in search_result: translated_word = search_result["translated_word"] else: translated_word = search_result["name_in_cultivation"] find_founder = founder_base.get_founder(translated_word) if find_founder is None: suffix = "这个物品很早就在天机阁中记载了，道友还有什么物品要鉴定吗？" else: if find_founder == founder_name: suffix = "感谢道友送来这个物品进行鉴定，天机阁的天梯上已经记录了你的贡献。" else: suffix = f"这个物品是由道友{find_founder}发现的，道友还有什么物品要鉴定吗？" else: if len(backup_results) > 0 and "similarity" in backup_results[0]: text_similarity = backup_results[0]['similarity'] print("text_similarity:", text_similarity) inlibrary_similarity_level = similarity2level(text_similarity, max_val=0.79, min_val=0.35) result = backup_results[0] # should call here # expensive_generating(save_path, image_feature, backup_results, game_master) task_queue.put((save_path, image_feature, backup_results, founder_name)) suffix = "道友请移步天机阁内阁查询长老的鉴定结果。" if is_empty_name( founder_name ): suffix += "另外，只有留下姓名的道友，才能最终被记录在天梯获取天机阁奖励。" # print(comments) # update to recent items # global recent_generated_items # recent_generated_items.append((save_path, result["description_in_cultivation"])) # Get the name and description from the first result name = result["name_in_cultivation"] description_in_cultivation = result["description_in_cultivation"] comments = get_comments_from_level( inbase_similarity_level, inlibrary_similarity_level ) # 将comments种的{name} format为name变量 comments = prefix + comments.format(name=name) + suffix # print(comments) return name, description_in_cultivation, comments # Function to refresh and display recent items def refresh_recent_items(): image_paths = [] descriptions = [] # for img_path, desc in recent_generated_items: # loop in inverse order for img_path, desc in reversed(recent_generated_items): img = Image.open(img_path) img.thumbnail((200, 200)) image_paths.append(img) descriptions.append(desc) return image_paths + descriptions # Prepare example images example_images_dir = "datas/example_images" example_images = [os.path.join(example_images_dir, img) for img in os.listdir(example_images_dir) if img.endswith('.jpg')] TODO_list = """ # TODO - [ ] 增加天梯匹配系统 - [ ] 增加readme - [ ] 部署到gitee """ # Gradio Interface with gr.Blocks() as demo: gr.Markdown("# 这个系统是智障，错把日常当修仙\n\nby [李鲁鲁](https://github.com/LC1332)") # Tab 1: Image processing and identification with gr.Tab("鉴定"): with gr.Row(): with gr.Column(scale = 1): image_input = gr.Image(label="上传图片") with gr.Column(scale = 2): comments = gr.Text("道友你有什么物品要来鉴定吗？", label = "") name_output = gr.Textbox(label="鉴定物品名称") description_output = gr.Textbox(label="物品描述") submit_button = gr.Button("鉴定") with gr.Row(): founder_name = gr.Textbox(label="道友尊姓大名？", interactive=True) image_input.upload(process_image, inputs=[image_input,founder_name], outputs=[name_output, description_output, comments]) submit_button.click(process_image, inputs=[image_input,founder_name], outputs=[name_output, description_output, comments]) gr.Examples(examples=example_images, inputs=image_input, label="选择一个示例图片") # Tab 2: Recent items with gr.Tab("天机阁最新鉴定"): refresh_button = gr.Button("询问长老最新鉴定") recent_images = [] recent_descriptions = [] for i in range(4): # Repeat for 4 rows, 3 columns each = 12 items # with gr.Column(): # Create a grid layout with 3 columns, each item having an image and a description below it with gr.Row(): with gr.Column(): recent_image = gr.Image(label=f"物品 { i * 4 + 1}") recent_description = gr.Textbox(label="描述", interactive=False) recent_images.append(recent_image) recent_descriptions.append(recent_description) with gr.Column(): recent_image = gr.Image(label=f"物品 { i * 4 + 2}") recent_description = gr.Textbox(label="描述", interactive=False) recent_images.append(recent_image) recent_descriptions.append(recent_description) with gr.Column(): recent_image = gr.Image(label=f"物品 { i * 4 + 3}") recent_description = gr.Textbox(label="描述", interactive=False) recent_images.append(recent_image) recent_descriptions.append(recent_description) # On clicking the refresh button, update the recent items refresh_button.click(refresh_recent_items, outputs=recent_images + recent_descriptions) with gr.Tab("贡献天梯"): refresh_ladder_button = gr.Button("刷新天梯") contribution_textboxes = [gr.Textbox(label=f"贡献者 {i + 1}", interactive=False) for i in range(20)] refresh_ladder_button.click(refresh_contribution_ladder, outputs=contribution_textboxes) # Launch the demo demo.launch(share=True)