Upload app.py

app.py

@@ -0,0 +1,332 @@
+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)
+