import uuid from django.http import JsonResponse from django.shortcuts import render, get_object_or_404 from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST, require_GET from django.utils.decorators import method_decorator from django.views import View import json import base64 from .models import Conversation, Message from ai_engine.services import ConversationManager from ai_engine.services.feedback_learner import record_message_feedback manager = ConversationManager() def _get_session_key(request): """Return the session key, initialising the session if needed.""" if not request.session.session_key: request.session.create() return request.session.session_key def _conversations_qs(request): """ Return conversations belonging to this user OR this session. This is the fix for the empty sidebar bug — previously the queryset filtered only on user, so anonymous or freshly-logged-in users saw nothing. """ session_key = _get_session_key(request) qs = Conversation.objects.order_by("-updated_at") if request.user.is_authenticated: # Authenticated: show conversations that are either owned by the user # OR were created in this session before login (so history persists). return qs.filter(user=request.user) | qs.filter(session_key=session_key, user__isnull=True) else: return qs.filter(session_key=session_key) def chat_home(request): """Render the main chat interface.""" conversation_id = request.session.get("conversation_id") conversations = [] if request.user.is_authenticated: conversations = Conversation.objects.filter(user=request.user)[:20] return render(request, "chat/home.html", { "conversation_id": conversation_id, "conversations": conversations, }) @require_POST @csrf_exempt def chat_send(request): """ Accept a user message, run it through the AI engine, and return the assistant's response plus any product cards. """ import json user_text = "" conversation_id = None image_query = None image_b64 = None image_attached = False content_type = request.content_type or "" if "multipart/form-data" in content_type: user_text = request.POST.get("message", "").strip() conversation_id = request.POST.get("conversation_id") image_file = request.FILES.get("image") if image_file: image_attached = True image_query, image_b64 = _extract_image_query(image_file, user_text) else: try: body = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) user_text = body.get("message", "").strip() conversation_id = body.get("conversation_id") if not user_text and not image_attached: return JsonResponse({"error": "Empty message"}, status=400) if not user_text and image_attached: user_text = "Find products similar to this image." conversation = _get_or_create_conversation(request, conversation_id) if image_attached: # Step 1: confirm intent before searching, to narrow image-based matches. detected_query = dict(image_query or {}) if not detected_query: detected_query = {"q": user_text or "similar product style"} # Mark this as image-driven so downstream prompts can stay strict. detected_query["image_search"] = True summary = _summarize_image_query(detected_query) Message.objects.create( conversation=conversation, role="user", content=user_text, ) if _is_low_confidence_image_query(detected_query): assistant_text = ( "I can search this more accurately with one quick detail. " "What item type is it, for example dress, two-piece set, top, skirt, bag, or shoes?" ) source = "image_clarify" else: assistant_text = ( "I understood your image as " f"{summary}. Do you want exact matches, or similar options?" ) source = "image_confirmation" Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={ "intent": "shopping", "source": source, "pending_image_query": detected_query, "pending_image_b64": image_b64, "pending_consumed": False, }, ) request.session["conversation_id"] = str(conversation.id) return JsonResponse({ "conversation_id": str(conversation.id), "content": assistant_text, "products": None, "metadata": {"intent": "shopping", "source": source}, "image_attached": True, }) pending_bundle = _get_pending_image_bundle(conversation) if pending_bundle and pending_bundle.get("source") == "image_clarify": pending = _apply_image_clarification(dict(pending_bundle.get("query") or {}), user_text) pending = _apply_image_search_mode(pending, "similar") _consume_pending_image_query(conversation) if pending_bundle.get("image_b64"): pending["image_b64"] = pending_bundle["image_b64"] result = manager.handle_message(conversation, user_text, forced_query=pending) elif pending_bundle and pending_bundle.get("source") == "image_confirmation" and _is_image_search_confirmation(user_text): pending = _apply_image_search_mode(dict(pending_bundle.get("query") or {}), _parse_image_search_mode(user_text)) _consume_pending_image_query(conversation) if pending_bundle.get("image_b64"): pending["image_b64"] = pending_bundle["image_b64"] result = manager.handle_message(conversation, user_text, forced_query=pending) else: result = manager.handle_message(conversation, user_text, forced_query=image_query) request.session["conversation_id"] = str(conversation.id) return JsonResponse({ "conversation_id": str(conversation.id), "content": result["content"], "products": result.get("products"), "metadata": result.get("metadata"), "image_attached": image_attached, }) @require_POST @csrf_exempt def chat_new(request): """Start a fresh conversation.""" conversation = Conversation.objects.create( user=request.user if request.user.is_authenticated else None, session_key=request.session.session_key or "", ) request.session["conversation_id"] = str(conversation.id) return JsonResponse({"conversation_id": str(conversation.id)}) @require_GET def chat_history(request, conversation_id): """Return message history for a conversation.""" conversation = get_object_or_404(Conversation, id=conversation_id) messages = conversation.messages.order_by("created_at").values( "role", "content", "product_data", "metadata", "created_at" ) return JsonResponse({ "conversation_id": str(conversation.id), "messages": list(messages), }) def _get_or_create_conversation(request, conversation_id=None): """Resolve or create a Conversation for this request.""" if conversation_id: try: return Conversation.objects.get(id=conversation_id) except Conversation.DoesNotExist: pass if not request.session.session_key: request.session.create() return Conversation.objects.create( user=request.user if request.user.is_authenticated else None, session_key=request.session.session_key or "", ) def _extract_image_query(image_file, user_text=""): allowed_types = {"image/jpeg", "image/jpg", "image/png", "image/webp"} media_type = (image_file.content_type or "").lower() if media_type not in allowed_types: return None, None try: image_bytes = image_file.read() except Exception: return None, None if not image_bytes: return None, None llm = manager.llm image_b64 = base64.b64encode(image_bytes).decode("utf-8") return llm.extract_image_query(image_bytes, media_type, user_text=user_text), image_b64 def _get_pending_image_bundle(conversation): msg = ( conversation.messages .filter(role="assistant") .exclude(metadata__isnull=True) .order_by("-created_at") .first() ) if not msg: return None metadata = msg.metadata or {} if metadata.get("source") not in ("image_confirmation", "image_clarify"): return None if metadata.get("pending_consumed"): return None query = metadata.get("pending_image_query") if not isinstance(query, dict): return None return { "source": metadata.get("source"), "query": query, "image_b64": metadata.get("pending_image_b64"), } def _consume_pending_image_query(conversation): msg = ( conversation.messages .filter(role="assistant") .exclude(metadata__isnull=True) .order_by("-created_at") .first() ) if not msg: return metadata = msg.metadata or {} if metadata.get("source") not in ("image_confirmation", "image_clarify"): return metadata["pending_consumed"] = True msg.metadata = metadata msg.save(update_fields=["metadata"]) def _is_image_search_confirmation(user_text): text = (user_text or "").lower().strip() if not text: return False signals = [ "yes", "yeah", "yep", "okay", "ok", "go ahead", "search", "find", "show me", "exact", "similar", "like this", "this style", ] return any(sig in text for sig in signals) def _parse_image_search_mode(user_text): text = (user_text or "").lower().strip() if any(k in text for k in ["exact", "same", "this exact", "identical"]): return "exact" if any(k in text for k in ["similar", "like this", "close", "anything like", "any similar"]): return "similar" return "similar" def _apply_image_search_mode(query, mode): q_text = str(query.get("q", "")).strip() category = str(query.get("category", "")).strip() color = str(query.get("color", "")).strip() style_terms = [] if color: style_terms.append(color) if category: style_terms.append(category) if q_text: style_terms.append(q_text) if mode == "exact": # Keep stricter, image-grounded terms. if style_terms: query["q"] = " ".join(style_terms[:8]).strip() else: # Slightly broaden for similar suggestions. broadened = " ".join(style_terms + ["similar", "style"]).strip() query["q"] = broadened or "similar style" query["image_match_mode"] = mode return query def _summarize_image_query(query): parts = [] category = query.get("category") color = query.get("color") brand = query.get("brand") q = query.get("q") if color and category: parts.append(f"a {color} {category}") elif category: parts.append(f"a {category}") if brand: parts.append(f"brand {brand}") if q: parts.append(f"style {q}") if not parts: return "a fashion/product style item" return ", ".join(parts) def _is_low_confidence_image_query(query): if not isinstance(query, dict): return True has_core = any(query.get(k) for k in ("category", "brand", "color")) q = str(query.get("q", "")).strip().lower() generic_q = q in ("", "style", "fashion", "outfit", "similar product style") return (not has_core) and generic_q def _apply_image_clarification(query, user_text): text = (user_text or "").strip().lower() if not text: return query category_map = { "dress": "dress", "two piece": "two-piece set", "two-piece": "two-piece set", "set": "set", "top": "top", "skirt": "skirt", "bag": "bag", "shoe": "shoes", "shoes": "shoes", "jacket": "jacket", "trouser": "trousers", "pants": "pants", } for key, value in category_map.items(): if key in text: query["category"] = value break existing_q = str(query.get("q", "")).strip() query["q"] = f"{existing_q} {user_text}".strip() if existing_q else user_text query["image_search"] = True return query # ───────────────────────────────────────────────────────────────────────────── # Conversation list / create # ───────────────────────────────────────────────────────────────────────────── @method_decorator(csrf_exempt, name="dispatch") class ConversationListView(View): """ GET → sidebar conversation list (FIXED) POST → create a new conversation """ def get(self, request): conversations = _conversations_qs(request).values( "id", "title", "created_at", "updated_at" ) data = [ { "id": str(c["id"]), "title": c["title"] or "New conversation", "created_at": c["created_at"].isoformat(), "updated_at": c["updated_at"].isoformat(), } for c in conversations ] return JsonResponse({"conversations": data}) def post(self, request): session_key = _get_session_key(request) user = request.user if request.user.is_authenticated else None try: body = json.loads(request.body or "{}") except json.JSONDecodeError: body = {} conversation = Conversation.objects.create( user=user, session_key=session_key, title=body.get("title", ""), ) return JsonResponse( { "id": str(conversation.id), "title": conversation.title or "New conversation", "created_at": conversation.created_at.isoformat(), }, status=201, ) # ───────────────────────────────────────────────────────────────────────────── # Message feedback # ───────────────────────────────────────────────────────────────────────────── @method_decorator(csrf_exempt, name="dispatch") class MessageFeedbackView(View): """ POST /api/chat/message//feedback/ Body: { "feedback": "up" | "down", "note": "optional text" } Saves the feedback on the Message row and updates the user's preference profile so future results improve. """ def post(self, request, message_id): try: body = json.loads(request.body) except (json.JSONDecodeError, TypeError): return JsonResponse({"error": "Invalid JSON body"}, status=400) feedback_value = body.get("feedback", "").strip().lower() if feedback_value not in ("up", "down"): return JsonResponse( {"error": "feedback must be 'up' or 'down'"}, status=400 ) note = body.get("note", "").strip() try: message = Message.objects.select_related("conversation").get( pk=message_id, role="assistant" ) except Message.DoesNotExist: return JsonResponse({"error": "Message not found"}, status=404) # Verify the requester owns this conversation session_key = _get_session_key(request) conv = message.conversation is_owner = ( (request.user.is_authenticated and conv.user == request.user) or conv.session_key == session_key ) if not is_owner: return JsonResponse({"error": "Forbidden"}, status=403) # Persist and trigger learning user = request.user if request.user.is_authenticated else None record_message_feedback( message_id=message_id, feedback=feedback_value, note=note, user=user, ) return JsonResponse({"status": "ok", "feedback": feedback_value})