Upload 2 files

app.py

@@ -0,0 +1,298 @@
+import gradio as gr
+import os
+import json
+import shutil
+from pathlib import Path
+from PIL import Image
+import torch
+import clip
+import numpy as np
+import requests
+from io import BytesIO
+import tempfile
+
+class SmartCLIPClassifierNextCloudShare:
+    def __init__(self, share_url, share_password, progress_callback=None):
+        
+        self.share_url = share_url.rstrip('/')
+        self.share_password = share_password
+        self.progress_callback = progress_callback
+        
+        self.session = requests.Session()
+        self.session.auth = (self.get_share_token(), share_password)
+        
+        self.temp_dir = tempfile.mkdtemp()
+        
+        self.categories = [
+            "1_Booth",
+            "2_Business_Interaction", 
+            "3_Buyer_Delegation",
+            "4_Aisle",
+            "5_Conference",
+            "6_Fairground",
+            "7_Products",
+            "8_Registration",
+            "9_Miscellaneous"
+        ]
+        
+        self.log("Loading CLIP model...")
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.model, self.preprocess = clip.load("ViT-B/32", device=self.device)
+        self.log(f"✅ CLIP loaded on {self.device}")
+        
+        self.log("🔍 Scanning NextCloud share...")
+        self.all_files = self.list_files("")
+        self.log(f"Found {len(self.all_files)} total files")
+        
+        self.load_deep_analysis()
+        self.get_image_list()
+    
+    def log(self, message):
+        if self.progress_callback:
+            self.progress_callback(message)
+        print(message)
+    
+    def get_share_token(self):
+        return self.share_url.split('/s/')[-1]
+    
+    def get_webdav_url(self, path=""):
+        token = self.get_share_token()
+        base = self.share_url.rsplit('/s/', 1)[0]
+        if path:
+            return f"{base}/public.php/webdav/{path}"
+        return f"{base}/public.php/webdav/"
+    
+    def download_file(self, filename):
+        url = self.get_webdav_url(filename)
+        response = self.session.get(url)
+        response.raise_for_status()
+        return response.content
+    
+    def upload_file(self, local_path, remote_filename):
+        url = self.get_webdav_url(remote_filename)
+        with open(local_path, 'rb') as f:
+            response = self.session.put(url, data=f)
+        response.raise_for_status()
+    
+    def list_files(self, remote_path=""):
+        url = self.get_webdav_url(remote_path)
+        response = self.session.request('PROPFIND', url, headers={'Depth': '1'})
+        response.raise_for_status()
+        
+        files = []
+        lines = response.text.split('<d:href>')
+        for line in lines:
+            if '</d:href>' in line:
+                href = line.split('</d:href>')[0]
+                if '/webdav/' in href:
+                    filename = href.split('/webdav/')[-1]
+                    if filename and not filename.endswith('/'):
+                        files.append(filename)
+        
+        return files
+    
+    def create_folder(self, folder_name):
+        url = self.get_webdav_url(folder_name)
+        try:
+            self.session.request('MKCOL', url)
+        except:
+            pass
+    
+    def get_image_list(self):
+        self.log("Filtering images...")
+        valid_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
+        
+        self.images = [f for f in self.all_files if Path(f).suffix.lower() in valid_extensions]
+        self.images.sort()
+        self.log(f"✅ Found {len(self.images)} images to classify")
+    
+    def load_deep_analysis(self):
+        self.log("Looking for deep_training_analysis.json...")
+        
+        json_file = None
+        for f in self.all_files:
+            if f.endswith('.json') and 'deep_training_analysis' in f.lower():
+                json_file = f
+                self.log(f"Found: {json_file}")
+                break
+        
+        if not json_file:
+            self.log("⚠️ deep_training_analysis.json not found - using fallback embeddings")
+            self.category_embeddings = {cat: self.create_text_embedding(cat) for cat in self.categories}
+            return
+        
+        try:
+            content = self.download_file(json_file)
+            self.deep_analysis = json.loads(content.decode('utf-8'))
+            
+            self.log("📚 Loaded deep training analysis")
+            
+            self.category_embeddings = {}
+            
+            for category in self.categories:
+                if category in self.deep_analysis:
+                    data = self.deep_analysis[category]
+                    avg_embedding = torch.tensor(data['avg_embedding'], dtype=torch.float32).to(self.device)
+                    avg_embedding = avg_embedding / avg_embedding.norm()
+                    self.category_embeddings[category] = avg_embedding
+                    
+                    self.log(f"  {category}: {data['num_training_images']} training images")
+                else:
+                    self.category_embeddings[category] = self.create_text_embedding(category)
+            
+        except Exception as e:
+            self.log(f"❌ Error loading deep analysis: {e}")
+            self.category_embeddings = {cat: self.create_text_embedding(cat) for cat in self.categories}
+    
+    def create_text_embedding(self, category):
+        descriptions = {
+            "1_Booth": "a photo of an exhibition booth at a trade show",
+            "2_Business_Interaction": "a photo of business people talking at a trade show",
+            "3_Buyer_Delegation": "a photo of a large group visiting a trade show",
+            "4_Aisle": "a photo of a trade show aisle between booths",
+            "5_Conference": "a photo of a conference presentation or seminar",
+            "6_Fairground": "a photo of an exhibition hall or fairground",
+            "7_Products": "a photo of products on display",
+            "8_Registration": "a photo of a registration desk or entry gate",
+            "9_Miscellaneous": "a miscellaneous trade show photo"
+        }
+        
+        text = descriptions.get(category, "a photo")
+        text_input = clip.tokenize([text]).to(self.device)
+        
+        with torch.no_grad():
+            text_features = self.model.encode_text(text_input)
+            text_features = text_features / text_features.norm(dim=-1, keepdim=True)
+        
+        return text_features[0]
+    
+    def classify_image(self, filename):
+        try:
+            img_data = self.download_file(filename)
+            
+            img = Image.open(BytesIO(img_data)).convert('RGB')
+            img_input = self.preprocess(img).unsqueeze(0).to(self.device)
+            
+            with torch.no_grad():
+                img_features = self.model.encode_image(img_input)
+                img_features = img_features / img_features.norm(dim=-1, keepdim=True)
+                img_features = img_features[0]
+                
+                similarities = {}
+                for category, cat_embedding in self.category_embeddings.items():
+                    similarity = (img_features @ cat_embedding).item()
+                    similarities[category] = similarity
+                
+                best_category = max(similarities, key=similarities.get)
+                confidence = similarities[best_category]
+            
+            local_path = os.path.join(self.temp_dir, Path(filename).name)
+            with open(local_path, 'wb') as f:
+                f.write(img_data)
+            
+            category_folder = f"Classified/{best_category}"
+            self.create_folder("Classified")
+            self.create_folder(category_folder)
+            
+            remote_dest = f"{category_folder}/{Path(filename).name}"
+            self.upload_file(local_path, remote_dest)
+            
+            os.remove(local_path)
+            
+            return best_category, confidence
+            
+        except Exception as e:
+            self.log(f"❌ Error: {str(e)}")
+            return "9_Miscellaneous", 0.0
+    
+    def run(self):
+        self.log("🚀 Starting classification...")
+        
+        self.create_folder("Classified")
+        for cat in self.categories:
+            self.create_folder(f"Classified/{cat}")
+        
+        stats = {cat: 0 for cat in self.categories}
+        confidences = {cat: [] for cat in self.categories}
+        
+        for i, filename in enumerate(self.images, 1):
+            self.log(f"[{i}/{len(self.images)}] {Path(filename).name}...")
+            
+            category, confidence = self.classify_image(filename)
+            
+            stats[category] += 1
+            confidences[category].append(confidence)
+            
+            self.log(f"  → {category} (confidence: {confidence:.3f})")
+        
+        self.log("✅ CLASSIFICATION COMPLETE!")
+        self.log("📁 Results uploaded to: Classified/")
+        
+        result_text = "**Results Summary:**\n\n"
+        for cat in self.categories:
+            count = stats[cat]
+            if count > 0:
+                avg_conf = sum(confidences[cat]) / len(confidences[cat])
+                result_text += f"- **{cat}**: {count} images (avg confidence: {avg_conf:.3f})\n"
+        
+        shutil.rmtree(self.temp_dir)
+        
+        return result_text
+
+def classify_photos(share_url, share_password, progress=gr.Progress()):
+    if not share_url or not share_password:
+        return "❌ Please enter both the share URL and password"
+    
+    try:
+        logs = []
+        
+        def log_callback(message):
+            logs.append(message)
+            progress(0.5, desc=message)
+        
+        classifier = SmartCLIPClassifierNextCloudShare(
+            share_url, 
+            share_password,
+            progress_callback=log_callback
+        )
+        
+        result = classifier.run()
+        
+        return result
+        
+    except Exception as e:
+        return f"❌ Error: {str(e)}\n\nPlease check your share URL and password and try again."
+
+# Gradio Interface
+with gr.Blocks(title="Trade Show Photo Classifier") as demo:
+    gr.Markdown("# 🤖 Trade Show Photo Classifier")
+    gr.Markdown("Automatically classify your trade show photos using AI-powered image recognition")
+    
+    with gr.Row():
+        with gr.Column():
+            share_url = gr.Textbox(
+                label="NextCloud Share URL",
+                placeholder="https://cloud2.messefrankfurtexchange.com/s/...",
+                info="Enter the public share link to your NextCloud folder containing the photos"
+            )
+            share_password = gr.Textbox(
+                label="Share Password",
+                type="password",
+                info="Enter the password for the NextCloud share"
+            )
+            classify_btn = gr.Button("🚀 Start Classification", variant="primary")
+    
+    with gr.Row():
+        output = gr.Markdown(label="Results")
+    
+    classify_btn.click(
+        fn=classify_photos,
+        inputs=[share_url, share_password],
+        outputs=output
+    )
+    
+    gr.Markdown("---")
+    gr.Markdown("*Powered by OpenAI CLIP | Deployed on Hugging Face Spaces*")
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,9 @@
+gradio
+torch
+torchvision
+Pillow
+requests
+ftfy
+regex
+tqdm
+git+https://github.com/openai/CLIP.git