""" Product Condition Classification Service using Groq Vision API Analyzes product images to determine if items are resellable, refurbishable, or scrap. Switched from Google Gemini to Groq to avoid free-tier rate limits. Uses meta-llama/llama-4-scout-17b-16e-instruct (with Maverick fallback) via the Groq chat completions endpoint. """ import os import sys import base64 import logging import mimetypes import re from typing import Literal from io import BytesIO logger = logging.getLogger(__name__) # Ensure backend root is importable so we can read voice_config _project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) if _project_root not in sys.path: sys.path.insert(0, _project_root) try: from groq import Groq GROQ_AVAILABLE = True except ImportError: GROQ_AVAILABLE = False logger.warning("groq package not installed. Install with: pip install groq") try: from voice_config import GROQ_API_KEY except ImportError: GROQ_API_KEY = os.environ.get('GROQ_API_KEY', '') ProductCondition = Literal["resale", "refurb", "scrap"] # ──────────────────────────────────────────────────────────────────── # Keyword extraction helper (unchanged from original) # ──────────────────────────────────────────────────────────────────── def _extract_classification_from_text(text: str) -> ProductCondition: """ Extract classification from natural language response using regex / keyword matching. Returns one of: "resale", "refurb", or "scrap". """ text_lower = text.lower() scrap_patterns = [ r'\bscrap\b', r'\bscrapped\b', r'\btoo damaged\b', r'\bbeyond repair\b', r'\bcannot be repaired\b', r'\bseverely damaged\b', r'\bcompletely broken\b', r'\bnot repairable\b', r'\bwrite[\s-]?off\b', r'\bdispose\b', r'\bunusable\b', ] resale_patterns = [ r'\bresale\b', r'\bresell\b', r'\bresellable\b', r'\bperfect condition\b', r'\bexcellent condition\b', r'\blike new\b', r'\bmint condition\b', r'\bpristine\b', r'\bno damage\b', r'\bflawless\b', r'\bnew condition\b', ] refurb_patterns = [ r'\brefurb\b', r'\brefurbish\b', r'\brefurbished\b', r'\brefurbishment\b', r'\bneeds repair\b', r'\bcan be repaired\b', r'\brepairable\b', r'\bminor damage\b', r'\bfixable\b', r'\bcan be fixed\b', r'\bneeds work\b', r'\bneeds restoration\b', ] scrap_score = sum(1 for p in scrap_patterns if re.search(p, text_lower)) resale_score = sum(1 for p in resale_patterns if re.search(p, text_lower)) refurb_score = sum(1 for p in refurb_patterns if re.search(p, text_lower)) logger.info(f"Classification scores - scrap: {scrap_score}, resale: {resale_score}, refurb: {refurb_score}") if scrap_score > 0 and scrap_score >= resale_score and scrap_score >= refurb_score: return "scrap" if resale_score > 0 and resale_score > scrap_score and resale_score >= refurb_score: return "resale" if refurb_score > 0: return "refurb" # Simple substring fallback if 'scrap' in text_lower: return "scrap" if 'resale' in text_lower or 'resell' in text_lower: return "resale" if 'refurb' in text_lower: return "refurb" logger.warning(f"Could not extract classification from: '{text}'. Defaulting to 'refurb'") return "refurb" # ──────────────────────────────────────────────────────────────────── # Groq-based classification service # ──────────────────────────────────────────────────────────────────── class GroqClassificationService: """Product condition classification via Groq Vision API.""" # Models tried in order of preference (vision-capable) _MODEL_PRIORITY = [ "meta-llama/llama-4-scout-17b-16e-instruct", # recommended replacement, fast "meta-llama/llama-4-maverick-17b-128e-instruct", # higher quality fallback ] def __init__(self, api_key: str | None = None): if not GROQ_AVAILABLE: raise ImportError("groq package not installed. Install with: pip install groq") self.api_key = api_key or GROQ_API_KEY or os.environ.get('GROQ_API_KEY', '') if not self.api_key: raise ValueError("GROQ_API_KEY not found. Set it in voice_config.py or as an env var.") self.client = Groq(api_key=self.api_key) self.model_name = self._MODEL_PRIORITY[0] logger.info(f"GroqClassificationService initialized (model: {self.model_name})") # ── helpers ──────────────────────────────────────────────────── @staticmethod def _image_to_data_url(image_file) -> str: """Read a Django UploadedFile into a base64 data-URL.""" image_file.seek(0) raw = image_file.read() # Detect MIME type mime = getattr(image_file, 'content_type', None) if not mime: fname = getattr(image_file, 'name', 'image.jpg') mime, _ = mimetypes.guess_type(fname) if not mime or not mime.startswith('image/'): mime = 'image/jpeg' b64 = base64.b64encode(raw).decode('utf-8') return f"data:{mime};base64,{b64}" def _call_vision(self, image_file, prompt: str, max_tokens: int = 500) -> str: """Send an image + text prompt to Groq vision and return the text reply.""" data_url = self._image_to_data_url(image_file) messages = [ { "role": "user", "content": [ { "type": "image_url", "image_url": {"url": data_url}, }, { "type": "text", "text": prompt, }, ], } ] last_err = None for model in self._MODEL_PRIORITY: try: response = self.client.chat.completions.create( model=model, messages=messages, temperature=0.3, max_tokens=max_tokens, ) text = response.choices[0].message.content.strip() logger.info(f"Groq vision ({model}) response: {text[:200]}") return text except Exception as e: last_err = e logger.warning(f"Groq model {model} failed: {e}") continue raise last_err # all models failed # ── public classification ────────────────────────────────────── def classify_product_condition(self, image_file) -> ProductCondition: """ Classify product condition from image. Returns one of: "resale", "refurb", or "scrap". """ prompt = ( "You are a product condition expert. " "Look at this product image carefully.\n\n" "Your response MUST start with one of these words:\n" "- resale (if excellent / like-new condition)\n" "- refurb (if damaged but fixable)\n" "- scrap (if too damaged to repair economically)\n\n" "Format: \"[classification] - [explanation of damage/condition]\"\n\n" "Analyze and respond:" ) try: result_text = self._call_vision(image_file, prompt) classification = _extract_classification_from_text(result_text) logger.info(f"Extracted classification: {classification}") return classification except Exception as e: logger.error(f"Error during classification: {e}", exc_info=True) logger.warning("Falling back to 'refurb' due to error") return "refurb" def verify_product_name_matches_image( self, image_file, product_name: str, threshold: float = 0.6 ) -> bool: """ Verify if uploaded image likely matches product name. Returns True (match) or False. Defaults to True when the LLM is unreachable (lenient). """ if not product_name or not product_name.strip(): logger.error("Product name is empty") return False prompt = ( f"Look at this product image.\n\n" f"The customer says this image is of: \"{product_name}\".\n\n" f"Does this image show the SAME GENERAL TYPE/CATEGORY of product? " f"For example, if the product name says 'router' the image should show a router or networking device. " f"If the product name says 'headphones' the image should show headphones or earbuds.\n\n" f"Be LENIENT — brand, color, or minor details don't matter. " f"Only say NO if the image clearly shows a COMPLETELY DIFFERENT type of product " f"(e.g., image shows clothing but product name says electronics).\n\n" f"Respond with ONLY 'YES' or 'NO'." ) try: result_text = self._call_vision(image_file, prompt, max_tokens=50) upper = result_text.upper() logger.info(f"Product match raw response: {result_text}") # Check for explicit YES/NO (whole word only) # Use word-boundary regex to avoid matching substrings like "NOISE" as "NO" has_yes = bool(re.search(r'\bYES\b', upper)) has_no = bool(re.search(r'\bNO\b', upper)) if has_yes and not has_no: return True if has_no and not has_yes: return False # Both YES and NO present — use position (whichever comes first) if has_yes and has_no: yes_pos = re.search(r'\bYES\b', upper).start() no_pos = re.search(r'\bNO\b', upper).start() return yes_pos < no_pos # Try to parse a numeric confidence match = re.search(r'(\d+)', upper) if match: confidence = int(match.group(1)) / 100.0 logger.info(f"Parsed confidence: {confidence}") return confidence >= threshold # If we can't parse the answer, be lenient and allow it logger.warning("Could not parse response - defaulting to True (lenient)") return True except Exception as e: logger.error(f"Error during product name verification: {e}", exc_info=True) return True # lenient on errors # ──────────────────────────────────────────────────────────────────── # Singleton & convenience functions (same public API as before) # ──────────────────────────────────────────────────────────────────── _service_instance = None def get_classification_service() -> GroqClassificationService: """Get or create singleton instance of classification service.""" global _service_instance if _service_instance is None: try: _service_instance = GroqClassificationService() logger.info("Created new GroqClassificationService instance") except Exception as e: logger.error(f"Failed to initialize classification service: {e}") raise return _service_instance def classify_product_image(image_file) -> ProductCondition: """Convenience: classify a product image -> 'resale' | 'refurb' | 'scrap'.""" service = get_classification_service() return service.classify_product_condition(image_file) def verify_product_name_matches_image( image_file, product_name: str, threshold: float = 0.6 ) -> bool: """Convenience: check if uploaded image matches product_name.""" service = get_classification_service() return service.verify_product_name_matches_image(image_file, product_name, threshold)