Update app.py

app.py

@@ -1,137 +1,124 @@
-
 import json
 import numpy as np
 from PIL import Image
 import subprocess
 import sys
-
 import logging
 import gradio as gr
 from typing import Tuple, Dict, Optional
-import gradio as gr
 import tempfile
+import cv2
+import moondream as md
+from openai import OpenAI
 
-import json
-import numpy as np
-from PIL import Image
 def install(package):
     subprocess.check_call([sys.executable, "-m", "pip", "install", package])
 
-# OpenCVとGradioをインストール
-install("opencv-python")
+# 依存関係のインストール
+install("opencv-python-headless")
 install("moondream")
 install("openai")
-from openai import OpenAI
-import cv2
-import moondream as md
+
 class PrivacyProtector:
     def __init__(self, moondream_api_key, deepseek_api_key):
-        # Moondream初期化
         self.moon_model = md.vl(api_key=moondream_api_key)
-        
-        # DeepSeekクライアント初期化
         self.deepseek_client = OpenAI(
             api_key=deepseek_api_key,
             base_url="https://api.deepseek.com"
         )
-    
+
     def analyze_risk(self, image_path):
         """画像のリスク分析を行う"""
-        # 画像読み込みとエンコード
-        pil_image = Image.open(image_path)
-        cv_image = cv2.imread(image_path)
-        encoded_image = self.moon_model.encode_image(pil_image)
-        
-        # 画像キャプション生成
-        caption = self.moon_model.caption(encoded_image)["caption"]
-        print(caption)
-        # リスク分析プロンプト
-        analysis_prompt = f"""
-        以下の画像説明を基に個人情報漏洩リスクを分析し、厳密にJSON形式で返答してください：
-        {{
-            "risk_level": "high|medium|low",
-            "risk_reason": "リスクの具体的理由",
-            "objects_to_remove": ["消去すべきオブジェクトリスト"]
-        }}
-        
-        画像説明: {caption}
-        """
-        
-        # DeepSeek API呼び出し
-        response = self.deepseek_client.chat.completions.create(
-            model="deepseek-chat",
-            messages=[
-                {"role": "system", "content": "あなたは優秀なセキュリティ分析AIです。"},
-                {"role": "user", "content": analysis_prompt}
-            ],
-            temperature=0.3,
-            response_format={"type": "json_object"}
-        )
-        
-        # 結果パース
-        result = json.loads(response.choices[0].message.content)
-        return pil_image, cv_image, result
-    
+        try:
+            pil_image = Image.open(image_path)
+            cv_image = cv2.imread(image_path)
+            if cv_image is None:
+                raise ValueError("画像の読み込みに失敗しました")
+
+            encoded_image = self.moon_model.encode_image(pil_image)
+            caption = self.moon_model.caption(encoded_image)["caption"]
+            
+            analysis_prompt = f"""
+            以下の画像説明を基に個人情報漏洩リスクを分析し、厳密にJSON形式で返答してください：
+            {{
+                "risk_level": "high|medium|low",
+                "risk_reason": "リスクの具体的理由",
+                "objects_to_remove": ["消去すべきオブジェクトリスト"]
+            }}
+            
+            画像説明: {caption}
+            """
+            
+            response = self.deepseek_client.chat.completions.create(
+                model="deepseek-chat",
+                messages=[
+                    {"role": "system", "content": "あなたは優秀なセキュリティ分析AIです。"},
+                    {"role": "user", "content": analysis_prompt}
+                ],
+                temperature=0.3,
+                response_format={"type": "json_object"}
+            )
+            
+            result = json.loads(response.choices[0].message.content)
+            return pil_image, cv_image, result
+        
+        except Exception as e:
+            logging.error(f"リスク分析エラー: {str(e)}")
+            raise
+
     def remove_objects(self, pil_image, cv_image, objects_to_remove):
         """オブジェクト検出と消去処理"""
-        # マスク作成
-        mask = np.zeros(cv_image.shape[:2], dtype=np.uint8)
-        h, w = cv_image.shape[:2]
-        
-        for obj_name in objects_to_remove:
-            # オブジェクト検出
-            detection = self.moon_model.detect(pil_image, obj_name)
-            print(detection)
-            for obj in detection["objects"]:
-                # 正規化座標→絶対座標変換
-                x_min = int(obj['x_min'] * w)
-                y_min = int(obj['y_min'] * h)
-                x_max = int(obj['x_max'] * w)
-                y_max = int(obj['y_max'] * h)
-                
-                # マスク領域を矩形で追加
-                cv2.rectangle(mask, (x_min, y_min), (x_max, y_max), 255, -1)
-        
-        # インペイント処理
-        inpainted = cv2.inpaint(cv_image, mask, 3, cv2.INPAINT_TELEA)
-        return inpainted
-    
-    def process_image(self, image_path, output_path):
+        try:
+            mask = np.zeros(cv_image.shape[:2], dtype=np.uint8)
+            h, w = cv_image.shape[:2]
+            
+            for obj_name in objects_to_remove:
+                detection = self.moon_model.detect(pil_image, obj_name)
+                for obj in detection["objects"]:
+                    x_min = int(obj['x_min'] * w)
+                    y_min = int(obj['y_min'] * h)
+                    x_max = int(obj['x_max'] * w)
+                    y_max = int(obj['y_max'] * h)
+                    cv2.rectangle(mask, (x_min, y_min), (x_max, y_max), 255, -1)
+            
+            inpainted = cv2.inpaint(cv_image, mask, 3, cv2.INPAINT_TELEA)
+            return inpainted
+        
+        except Exception as e:
+            logging.error(f"オブジェクト削除エラー: {str(e)}")
+            raise
+
+    def process_image(self, image_path):
         """画像処理フロー全体"""
-        pil_img, cv_img, result = self.analyze_risk(image_path)
-        print(f"リスクレベル: {result['risk_level']}")
-        print(f"理由: {result['risk_reason']}")
-        
-        if result['risk_level'] != 'low':
-            print( result['objects_to_remove'])
-            cleaned = self.remove_objects(pil_img, cv_img, result['objects_to_remove'])
-            cv2.imwrite(output_path, cleaned)
-            print(f"処理済み画像を保存: {output_path}")
-            return True
-        print("リスク対象なし")
-        return False
+        try:
+            pil_img, cv_img, result = self.analyze_risk(image_path)
+            if result['risk_level'] != 'low':
+                cleaned = self.remove_objects(pil_img, cv_img, result['objects_to_remove'])
+                return cleaned, result
+            return cv_img, result
+        
+        except Exception as e:
+            logging.error(f"画像処理エラー: {str(e)}")
+            raise
 
 def gradio_process(input_image):
-    with tempfile.NamedTemporaryFile(suffix=".jpg") as tmp_file:
-        input_path = tmp_file.name
-        output_path = "/content/output.jpg"
-        
-        cv2.imwrite(input_path, cv2.cvtColor(input_image, cv2.COLOR_RGB2BGR))
+    try:
+        # 一時ファイルの処理
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp_file:
+            input_path = tmp_file.name
+            cv2.imwrite(input_path, cv2.cvtColor(input_image, cv2.COLOR_RGB2BGR))
         
         protector = PrivacyProtector(
-            moondream_api_key="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJrZXlfaWQiOiI4NmI0NTUxMi01NWZlLTQ0YzItYTA2Ni1hMTE1NDBlM2EzZTMiLCJpYXQiOjE3Mzc4NTMzMzJ9.8dc3xo_z7nYApJwaOCI2ayYX9pTKpo8QHkoH96ykEcg",
-            deepseek_api_key="sk-61418a466189492c916ec7bcc873b142"
+            moondream_api_key="your_moondream_key",
+            deepseek_api_key="your_deepseek_key"
         )
         
+        # 画像処理実行
+        processed_image, result = protector.process_image(input_path)
         
-        # 分析結果を含めて取得
-        pil_img, cv_img, result = protector.analyze_risk(input_path)
-        processed = protector.process_image(input_path, output_path)
-        
-        # 出力画像
-        output = cv2.cvtColor(cv2.imread(output_path), cv2.COLOR_BGR2RGB) if processed else None
+        # 結果の整形
+        output_image = cv2.cvtColor(processed_image, cv2.COLOR_BGR2RGB)
         
-        # 表示用情報の整形
         info_html = f"""
         <div style="padding:20px; background:#f0f0f0; border-radius:10px;">
             <h3>分析結果</h3>
@@ -142,7 +129,17 @@ def gradio_process(input_image):
         </div>
         """
         
-        return input_image, output, info_html
+        return input_image, output_image, info_html
+    
+    except Exception as e:
+        error_msg = f"処理中にエラーが発生しました: {str(e)}"
+        return input_image, None, f'<div style="color:red;">{error_msg}</div>'
+    
+    finally:
+        # 一時ファイルのクリーンアップ
+        if 'input_path' in locals():
+            import os
+            os.remove(input_path)
 
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
     gr.Markdown("# 🛡️ プライバシー保護画像処理ツール")