Update app.py

app.py

@@ -7,25 +7,26 @@ import onnxruntime as rt
 import pandas as pd
 from PIL import Image
 from huggingface_hub import login
-
-from translator import translate_texts
+from translator import translate_texts, translate_texts_with_dynamic_keys
 
 # ------------------------------------------------------------------
-# 模型配置
+# 模型与认证配置
 # ------------------------------------------------------------------
 MODEL_REPO = "SmilingWolf/wd-eva02-large-tagger-v3"
 MODEL_FILENAME = "model.onnx"
 LABEL_FILENAME = "selected_tags.csv"
+OWNER_USERNAME = os.environ.get("OWNER_USERNAME", "")
 
 HF_TOKEN = os.environ.get("HF_TOKEN", "")
 if HF_TOKEN:
-    login(token=HF_TOKEN)
+    try:
+        login(token=HF_TOKEN)
+        print("✅ HF_TOKEN 登录成功")
+    except Exception as e:
+        print(f"⚠️ HF_TOKEN 登录失败: {e}")
 else:
     print("⚠️ 未检测到 HF_TOKEN，私有模型可能下载失败")
 
-# ------------------------------------------------------------------
-# Tagger 类 (全局实例化)
-# ------------------------------------------------------------------
 class Tagger:
     def __init__(self):
         self.hf_token = HF_TOKEN
@@ -59,7 +60,6 @@ class Tagger:
             raise RuntimeError(f"模型初始化失败: {e}")
 
 
-    # ------------------------- preprocess -------------------------
     def _preprocess(self, img: Image.Image) -> np.ndarray:
         if img is None:
             raise ValueError("输入图像不能为空")
@@ -70,9 +70,8 @@ class Tagger:
         canvas.paste(img, ((size - img.width) // 2, (size - img.height) // 2))
         if size != self.input_size:
             canvas = canvas.resize((self.input_size, self.input_size), Image.BICUBIC)
-        return np.array(canvas)[:, :, ::-1].astype(np.float32)  # to BGR
+        return np.array(canvas)[:, :, ::-1].astype(np.float32)
 
-    # --------------------------- predict --------------------------
     def predict(self, img: Image.Image, gen_th: float = 0.35, char_th: float = 0.85):
         if self.model is None:
             raise RuntimeError("模型未成功加载，无法进行预测。")
@@ -111,125 +110,41 @@ class Tagger:
 
         return res, tag_categories_for_translation
 
-# 全局 Tagger 实例
 try:
     tagger_instance = Tagger()
 except RuntimeError as e:
     print(f"应用启动时Tagger初始化失败: {e}")
-    tagger_instance = None # 允许应用启动，但在处理时会失败
+    tagger_instance = None
 
-# ------------------------------------------------------------------
-# Gradio UI
-# ------------------------------------------------------------------
 custom_css = """
-.label-container {
-    max-height: 300px;
-    overflow-y: auto;
-    border: 1px solid #ddd;
-    padding: 10px;
-    border-radius: 5px;
-    background-color: #f9f9f9;
-}
-.tag-item {
-    display: flex;
-    justify-content: space-between;
-    align-items: center;
-    margin: 2px 0;
-    padding: 2px 5px;
-    border-radius: 3px;
-    background-color: #fff;
-    transition: background-color 0.2s;
-}
-.tag-item:hover {
-    background-color: #f0f0f0;
-}
-.tag-en {
-    font-weight: bold;
-    color: #333;
-    cursor: pointer; /* Indicates clickable */
-}
-.tag-zh {
-    color: #666;
-    margin-left: 10px;
-}
-.tag-score {
-    color: #999;
-    font-size: 0.9em;
-}
-.btn-analyze-container { /* Custom class for analyze button container */
-    margin-top: 15px;
-    margin-bottom: 15px;
-}
+.label-container { max-height: 300px; overflow-y: auto; border: 1px solid #ddd; padding: 10px; border-radius: 5px; background-color: #f9f9f9; }
+.tag-item { display: flex; justify-content: space-between; align-items: center; margin: 2px 0; padding: 2px 5px; border-radius: 3px; background-color: #fff; transition: background-color 0.2s; }
+.tag-item:hover { background-color: #f0f0f0; }
+.tag-en { font-weight: bold; color: #333; cursor: pointer; }
+.tag-zh { color: #666; margin-left: 10px; }
+.tag-score { color: #999; font-size: 0.9em; }
+.btn-analyze-container { margin-top: 15px; margin-bottom: 15px; }
 """
 
 _js_functions = """
 function copyToClipboard(text) {
-    // --- 调试信息 ---
-    console.log('copyToClipboard function was called.');
-    console.log('Received text:', text);
-    // console.trace(); // 如果需要更详细的调用栈信息，可以取消这行注释
-
-    // --- 保护性检查 ---
-    // 如果 text 未定义或为 null，则不执行后续操作，并打印警告
     if (typeof text === 'undefined' || text === null) {
-        console.warn('copyToClipboard was called with undefined or null text. Aborting this specific copy operation.');
-        // 在这种情况下，我们不应该尝试复制，也不应该显示“已复制”的提示
+        console.warn('copyToClipboard was called with undefined or null text.');
         return;
     }
-
     navigator.clipboard.writeText(text).then(() => {
-        // console.log('Tag copied to clipboard: ' + text); // 成功复制的日志（可选）
         const feedback = document.createElement('div');
-        
-        // 确保 text 是���符串类型，再进行 substring 操作
-        let displayText = String(text); // 将 text 转换为字符串以防万一
+        let displayText = String(text);
         displayText = displayText.substring(0, 30) + (displayText.length > 30 ? '...' : '');
-        
         feedback.textContent = '已复制: ' + displayText;
-        feedback.style.position = 'fixed';
-        feedback.style.bottom = '20px';
-        feedback.style.left = '50%';
-        feedback.style.transform = 'translateX(-50%)';
-        feedback.style.backgroundColor = '#4CAF50';
-        feedback.style.color = 'white';
-        feedback.style.padding = '10px 20px';
-        feedback.style.borderRadius = '5px';
-        feedback.style.zIndex = '10000';
-        feedback.style.transition = 'opacity 0.5s ease-out';
+        feedback.style.cssText = 'position: fixed; bottom: 20px; left: 50%; transform: translateX(-50%); background-color: #4CAF50; color: white; padding: 10px 20px; border-radius: 5px; z-index: 10000; transition: opacity 0.5s ease-out;';
         document.body.appendChild(feedback);
         setTimeout(() => {
             feedback.style.opacity = '0';
-            setTimeout(() => {
-                if (document.body.contains(feedback)) { // 确保元素还在DOM中
-                    document.body.removeChild(feedback);
-                }
-            }, 500);
+            setTimeout(() => { if (document.body.contains(feedback)) document.body.removeChild(feedback); }, 500);
         }, 1500);
     }).catch(err => {
         console.error('Failed to copy tag. Error:', err, 'Attempted to copy text:', text);
-        // 可以考虑也给用户一个错误提示，但原版 alert 可能体验不佳
-        // alert('复制失败: ' + err); 
-        const errorFeedback = document.createElement('div');
-        errorFeedback.textContent = '复制操作失败!'; // 更友好的错误提示
-        errorFeedback.style.position = 'fixed';
-        errorFeedback.style.bottom = '20px';
-        errorFeedback.style.left = '50%';
-        errorFeedback.style.transform = 'translateX(-50%)';
-        errorFeedback.style.backgroundColor = '#D32F2F'; // 红色背景表示错误
-        errorFeedback.style.color = 'white';
-        errorFeedback.style.padding = '10px 20px';
-        errorFeedback.style.borderRadius = '5px';
-        errorFeedback.style.zIndex = '10000';
-        errorFeedback.style.transition = 'opacity 0.5s ease-out';
-        document.body.appendChild(errorFeedback);
-        setTimeout(() => {
-            errorFeedback.style.opacity = '0';
-            setTimeout(() => {
-                if (document.body.contains(errorFeedback)) {
-                    document.body.removeChild(errorFeedback);
-                }
-            }, 500);
-        }, 2500);
     });
 }
 """
@@ -246,13 +161,22 @@ with gr.Blocks(theme=gr.themes.Soft(), title="AI 图像标签分析器", css=cus
     with gr.Row():
         with gr.Column(scale=1):
             img_in = gr.Image(type="pil", label="上传图片", height=300)
-            
             btn = gr.Button("🚀 开始分析", variant="primary", elem_classes=["btn-analyze-container"])
             
             with gr.Accordion("⚙️ 高级设置", open=False):
                 gen_slider = gr.Slider(0, 1, value=0.35, step=0.01, label="通用标签阈值", info="越高 → 标签更少更准")
                 char_slider = gr.Slider(0, 1, value=0.85, step=0.01, label="角色标签阈值", info="推荐保持较高阈值")
                 show_tag_scores = gr.Checkbox(True, label="在列表中显示标签置信度")
+                
+                with gr.Group(visible=False) as guest_api_group:
+                    gr.Markdown("### 访客翻译API配置\n由于您不是本空间所有者，需要提供自己的翻译API密钥才能使用翻译功能。")
+                    guest_tencent_id = gr.Textbox(label="腾讯云 Secret ID", type="password")
+                    guest_tencent_key = gr.Textbox(label="腾讯云 Secret Key", type="password")
+                    guest_baidu_json = gr.TextArea(
+                        label="百度翻译凭证 JSON",
+                        placeholder='[{"app_id": "...", "secret_key": "..."}, ...]',
+                        lines=3
+                    )
             
             with gr.Accordion("📊 标签汇总设置", open=True):
                 gr.Markdown("选择要包含在下方汇总文本框中的标签类别：")
@@ -283,209 +207,121 @@ with gr.Blocks(theme=gr.themes.Soft(), title="AI 图像标签分析器", css=cus
                 show_copy_button=True
             )
 
-    # ----------------- 辅助函数 -----------------
     def format_tags_html(tags_dict, translations_list, category_name, show_scores=True, show_translation_in_list=True):
-        if not tags_dict:
-            return "<p>暂无标签</p>"
-        
+        if not tags_dict: return "<p>暂无标签</p>"
         html = '<div class="label-container">'
-        
-        if not isinstance(translations_list, list):
-            translations_list = []
-
         tag_keys = list(tags_dict.keys())
-
         for i, tag in enumerate(tag_keys):
             score = tags_dict[tag]
-            escaped_tag = tag.replace("'", "\\'") # Escape for JS
-            
+            escaped_tag = tag.replace("'", "\\'")
             html += '<div class="tag-item">'
             tag_display_html = f'<span class="tag-en" onclick="copyToClipboard(\'{escaped_tag}\')">{tag}</span>'
-            
             if show_translation_in_list and i < len(translations_list) and translations_list[i]:
                 tag_display_html += f'<span class="tag-zh">({translations_list[i]})</span>'
-            
             html += f'<div>{tag_display_html}</div>'
-            if show_scores:
-                html += f'<span class="tag-score">{score:.3f}</span>'
+            if show_scores: html += f'<span class="tag-score">{score:.3f}</span>'
             html += '</div>'
         html += '</div>'
         return html
 
-    def generate_summary_text_content(
-        current_res, current_translations_dict,
-        s_gen, s_char, s_rat, s_sep_type, s_show_zh
-    ):
-        if not current_res:
-            return "请先分析图像或选择要汇总的标签类别。"
-
-        summary_parts = []
-        separators = {"逗号": ", ", "换行": "\n", "空格": " "}
-        separator = separators.get(s_sep_type, ", ")
-
+    def generate_summary_text_content(current_res, current_translations_dict, s_gen, s_char, s_rat, s_sep_type, s_show_zh):
+        if not current_res: return "请先分析图像或选择要汇总的标签类别。"
+        summary_parts, separator = [], {"逗号": ", ", "换行": "\n", "空格": " "}.get(s_sep_type, ", ")
         categories_to_summarize = []
         if s_gen: categories_to_summarize.append("general")
         if s_char: categories_to_summarize.append("characters")
         if s_rat: categories_to_summarize.append("ratings")
-
-        if not categories_to_summarize:
-            return "请至少选择一个标签类别进行汇总。"
+        if not categories_to_summarize: return "请至少选择一个标签类别进行汇总。"
 
         for cat_key in categories_to_summarize:
             if current_res.get(cat_key):
                 tags_to_join = []
                 cat_tags_en = list(current_res[cat_key].keys())
                 cat_translations = current_translations_dict.get(cat_key, [])
-
                 for i, en_tag in enumerate(cat_tags_en):
                     if s_show_zh and i < len(cat_translations) and cat_translations[i]:
                         tags_to_join.append(f"{en_tag}({cat_translations[i]})")
                     else:
                         tags_to_join.append(en_tag)
-                if tags_to_join: # only add if there are tags for this category
-                     summary_parts.append(separator.join(tags_to_join))
-        
-        # Join parts with double newline for readability if multiple categories present and separator is not newline
-        joiner = "\n\n" if separator != "\n" and len(summary_parts) > 1 else separator if separator == "\n" else " "
-        
+                if tags_to_join: summary_parts.append(separator.join(tags_to_join))
+        joiner = "\n\n" if separator != "\n" and len(summary_parts) > 1 else separator
         final_summary = joiner.join(summary_parts)
         return final_summary if final_summary else "选定的类别中没有找到标签。"
 
-
-    # ----------------- 主要处理回调 -----------------
-    def process_image_and_generate_outputs(
-        img, g_th, c_th, s_scores, # Main inputs
-        s_gen, s_char, s_rat, s_sep, s_zh_in_sum
-        ):
+    def process_image_and_generate_outputs(img, g_th, c_th, s_scores, s_gen, s_char, s_rat, s_sep, s_zh_in_sum, guest_tc_id, guest_tc_key, guest_bd_json, request: gr.Request):
         if img is None:
-            yield (
-                gr.update(interactive=True, value="🚀 开始分析"),
-                gr.update(visible=True, value="❌ 请先上传图片。"),
-                "", "", "", "",
-                gr.update(placeholder="请先上传图片并开始分析..."),
-                {}, {}, {} 
-            )
+            yield (gr.update(interactive=True, value="🚀 开始分析"), gr.update(visible=True, value="❌ 请先上传图片。"), "", "", "", "", gr.update(placeholder="请先上传图片并开始分析..."), {}, {}, {})
             return
-        
         if tagger_instance is None:
-            yield (
-                gr.update(interactive=True, value="🚀 开始分析"),
-                gr.update(visible=True, value="❌ 分析器未成功初始化，请检查控制台错误。"),
-                "", "", "", "",
-                gr.update(placeholder="分析器初始化失败..."),
-                {}, {}, {}
-            )
+            yield (gr.update(interactive=True, value="🚀 开始分析"), gr.update(visible=True, value="❌ 分析器未成功初始化，请检查控制台错误。"), "", "", "", "", gr.update(placeholder="分析器初始化失败..."), {}, {}, {})
             return
 
-        yield (
-            gr.update(interactive=False, value="🔄 处理中..."),
-            gr.update(visible=True, value="🔄 正在分析图像，请稍候..."),
-            gr.HTML(value="<p>分析中...</p>"), # General
-            gr.HTML(value="<p>分析中...</p>"), # Character
-            gr.HTML(value="<p>分析中...</p>"), # Rating
-            gr.update(value="分析中，请稍候..."), # Summary
-            {}, {}, {} # Clear states initially
-        )
+        yield (gr.update(interactive=False, value="🔄 处理中..."), gr.update(visible=True, value="🔄 正在分析图像..."), gr.HTML(value="<p>分析中...</p>"), gr.HTML(value="<p>分析中...</p>"), gr.HTML(value="<p>分析中...</p>"), gr.update(value="分析中..."), {}, {}, {})
         
         try:
-            # 1. Predict tags
             res, tag_categories_original_order = tagger_instance.predict(img, g_th, c_th)
-
-            all_tags_to_translate = []
-            for cat_key in ["general", "characters", "ratings"]:
-                all_tags_to_translate.extend(tag_categories_original_order.get(cat_key, []))
+            all_tags_to_translate = [tag for cat_key in ["general", "characters", "ratings"] for tag in tag_categories_original_order.get(cat_key, [])]
             
             all_translations_flat = []
             if all_tags_to_translate:
-                all_translations_flat = translate_texts(all_tags_to_translate, src_lang="auto", tgt_lang="zh")
-            
+                is_owner = request.username and request.username == OWNER_USERNAME
+                if is_owner:
+                    print("- [Auth] 所有者身份，使用预置密钥进行翻译。")
+                    all_translations_flat = translate_texts(all_tags_to_translate)
+                else:
+                    print("- [Auth] 访客身份，使用用户提供的密钥进行翻译。")
+                    if not guest_tc_id and not guest_bd_json:
+                        print("  - [Warning] 访客未提供任何API密钥，将跳过翻译。")
+                        all_translations_flat = all_tags_to_translate
+                    else:
+                        all_translations_flat = translate_texts_with_dynamic_keys(all_tags_to_translate, guest_tc_id, guest_tc_key, guest_bd_json)
+
             current_translations_dict = {}
             offset = 0
             for cat_key in ["general", "characters", "ratings"]:
-                cat_original_tags = tag_categories_original_order.get(cat_key, [])
-                num_tags_in_cat = len(cat_original_tags)
-                if num_tags_in_cat > 0:
-                    current_translations_dict[cat_key] = all_translations_flat[offset : offset + num_tags_in_cat]
-                    offset += num_tags_in_cat
-                else:
-                    current_translations_dict[cat_key] = []
+                num_tags = len(tag_categories_original_order.get(cat_key, []))
+                current_translations_dict[cat_key] = all_translations_flat[offset : offset + num_tags]
+                offset += num_tags
             
-            general_html = format_tags_html(res.get("general", {}), current_translations_dict.get("general", []), "general", s_scores, True)
-            char_html = format_tags_html(res.get("characters", {}), current_translations_dict.get("characters", []), "characters", s_scores, True)
-            rating_html = format_tags_html(res.get("ratings", {}), current_translations_dict.get("ratings", []), "ratings", s_scores, True)
-
-            summary_text = generate_summary_text_content(
-                res, current_translations_dict,
-                s_gen, s_char, s_rat, s_sep, s_zh_in_sum 
-            )
+            general_html = format_tags_html(res.get("general", {}), current_translations_dict.get("general", []), "general", s_scores)
+            char_html = format_tags_html(res.get("characters", {}), current_translations_dict.get("characters", []), "characters", s_scores)
+            rating_html = format_tags_html(res.get("ratings", {}), current_translations_dict.get("ratings", []), "ratings", s_scores)
+            summary_text = generate_summary_text_content(res, current_translations_dict, s_gen, s_char, s_rat, s_sep, s_zh_in_sum)
 
-            yield (
-                gr.update(interactive=True, value="🚀 开始分析"),
-                gr.update(visible=True, value="✅ 分析完成!"), 
-                general_html,
-                char_html,
-                rating_html,
-                gr.update(value=summary_text),
-                res,
-                current_translations_dict,
-                tag_categories_original_order
-            )
+            yield (gr.update(interactive=True, value="🚀 开始分析"), gr.update(visible=True, value="✅ 分析完成!"), general_html, char_html, rating_html, gr.update(value=summary_text), res, current_translations_dict, tag_categories_original_order)
             
         except Exception as e:
             import traceback
             tb_str = traceback.format_exc()
             print(f"处理时发生��误: {e}\n{tb_str}")
-            yield (
-                gr.update(interactive=True, value="🚀 开始分析"),
-                gr.update(visible=True, value=f"❌ 处理失败: {str(e)}"),
-                "<p>处理出错</p>", "<p>处理出错</p>", "<p>处理出错</p>",
-                gr.update(value=f"错误: {str(e)}", placeholder="分析失败..."),
-                {}, {}, {}
-            )
+            yield (gr.update(interactive=True, value="🚀 开始分析"), gr.update(visible=True, value=f"❌ 处理失败: {str(e)}"), "<p>处理出错</p>", "<p>处理出错</p>", "<p>处理出错</p>", gr.update(value=f"错误: {str(e)}", placeholder="分析失败..."), {}, {}, {})
 
-    # ----------------- 更新汇总文本的回调 -----------------
-    def update_summary_display(
-        s_gen, s_char, s_rat, s_sep, s_zh_in_sum,
-        current_res_from_state, current_translations_from_state
-    ):
-        if not current_res_from_state:
-            return gr.update(placeholder="请先完成一次图像分析以生成汇总。", value="")
-
-        new_summary_text = generate_summary_text_content(
-            current_res_from_state, current_translations_from_state,
-            s_gen, s_char, s_rat, s_sep, s_zh_in_sum
-        )
+    def update_summary_display(s_gen, s_char, s_rat, s_sep, s_zh_in_sum, current_res, current_translations):
+        if not current_res: return gr.update(placeholder="请先完成一次图像分析以生成汇总。", value="")
+        new_summary_text = generate_summary_text_content(current_res, current_translations, s_gen, s_char, s_rat, s_sep, s_zh_in_sum)
         return gr.update(value=new_summary_text)
 
-    # ----------------- 绑定事件 -----------------
+    def check_user_auth(request: gr.Request):
+        if not OWNER_USERNAME: print("⚠️ 警告: 未设置 OWNER_USERNAME 环境变量。所有用户都将被视为访客。")
+        if not request.username or request.username != OWNER_USERNAME:
+            print(f"- [Auth] 访客 '{request.username}' 已连接，显示 API Key 输入框。")
+            return gr.update(visible=True)
+        print(f"- [Auth] 所有者 '{request.username}' 已连接。")
+        return gr.update(visible=False)
+
+    demo.load(fn=check_user_auth, inputs=None, outputs=[guest_api_group])
+
     btn.click(
         process_image_and_generate_outputs,
-        inputs=[
-            img_in, gen_slider, char_slider, show_tag_scores,
-            sum_general, sum_char, sum_rating, sum_sep, sum_show_zh
-        ],
-        outputs=[
-            btn, processing_info,
-            out_general, out_char, out_rating,
-            out_summary,
-            state_res, state_translations_dict, state_tag_categories_for_translation
-        ],
-        # show_progress="full" # Gradio's built-in progress
+        inputs=[img_in, gen_slider, char_slider, show_tag_scores, sum_general, sum_char, sum_rating, sum_sep, sum_show_zh, guest_tencent_id, guest_tencent_key, guest_baidu_json],
+        outputs=[btn, processing_info, out_general, out_char, out_rating, out_summary, state_res, state_translations_dict, state_tag_categories_for_translation]
     )
 
     summary_controls = [sum_general, sum_char, sum_rating, sum_sep, sum_show_zh]
     for ctrl in summary_controls:
-        ctrl.change(
-            fn=update_summary_display,
-            inputs=summary_controls + [state_res, state_translations_dict],
-            outputs=[out_summary],
-            # show_progress=False # Typically fast, no need for progress indicator
-        )
+        ctrl.change(fn=update_summary_display, inputs=summary_controls + [state_res, state_translations_dict], outputs=[out_summary])
     
-# ------------------------------------------------------------------
-# 启动
-# ------------------------------------------------------------------
 if __name__ == "__main__":
     if tagger_instance is None:
         print("CRITICAL: Tagger 未能初始化，应用功能将受限。请检查之前的错误信息。")
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch(server_name="0.0.0.0", server_port=7860)