Update& change payload

core/collection_router_retriever.py

@@ -1,8 +1,10 @@
 import hashlib
 import logging
-from typing import List
+import re
+from typing import List, Optional
 
 from langchain_core.documents import Document as LangChainDocument
+from qdrant_client.models import Filter, FieldCondition, HasIdCondition, MatchAny
 
 from .collection_utils import collection_matches_year
 from .document_db import SessionLocal, list_active_collection_names
@@ -10,6 +12,47 @@ from .document_db import SessionLocal, list_active_collection_names
 logger = logging.getLogger(__name__)
 
 
+def _build_year_filter(year_scope: Optional[str]) -> Optional[Filter]:
+    """Tạo Qdrant Filter từ year_scope (ví dụ: '2023-2024' hoặc '2023')."""
+    if not year_scope:
+        return None
+    
+    year_targets = []
+    year_scope = year_scope.strip()
+    
+    # Parse year_scope: có thể là "2023-2024" hoặc "2023"
+    if "-" in year_scope:
+        parts = year_scope.split("-")
+        for p in parts:
+            try:
+                year_targets.append(int(p.strip()))
+            except ValueError:
+                pass
+    else:
+        try:
+            year_targets.append(int(year_scope))
+        except ValueError:
+            pass
+    
+    if not year_targets:
+        return None
+    
+    # Sử dụng MatchAny để filter theo danh sách years
+    from qdrant_client.models import HasIdCondition as QdrantHasId
+    try:
+        return Filter(
+            must=[
+                FieldCondition(
+                    key="years",
+                    match=MatchAny(any=year_targets),
+                )
+            ]
+        )
+    except Exception:
+        # Fallback nếu MatchAny không work
+        return None
+
+
 class CollectionRouterRetriever:
     def __init__(
         self,
@@ -61,7 +104,7 @@ class CollectionRouterRetriever:
 
         return active_collections[: self.top_n_collections]
 
-    def _search_target_collections(self, query: str, collections: List[str], limit: int) -> List:
+    def _search_target_collections(self, query: str, collections: List[str], limit: int, year_scope: Optional[str] = None) -> List:
         if not collections:
             return []
 
@@ -71,6 +114,11 @@ class CollectionRouterRetriever:
             logger.exception("Failed to embed query for collection routing")
             return []
 
+        # Tạo filter Qdrant nếu có year_scope
+        year_filter = _build_year_filter(year_scope)
+        if year_filter:
+            logger.info(f"Áp dụng Qdrant Filter cho year_scope: {year_scope}")
+
         scored_docs = []
         for collection_name in collections:
             try:
@@ -79,9 +127,10 @@ class CollectionRouterRetriever:
                     query_vector=query_vector,
                     limit=limit,
                     with_payload=True,
+                    query_filter=year_filter,  # NEW: Áp dụng Qdrant Filter native
                 )
-            except Exception:
-                logger.exception("Qdrant search failed for collection=%s", collection_name)
+            except Exception as e:
+                logger.exception(f"Qdrant search failed for collection={collection_name}: {e}")
                 continue
 
             for point in points:
@@ -95,9 +144,11 @@ class CollectionRouterRetriever:
                     "source_file": payload.get("filename") or payload.get("stored_name") or "",
                     "source_relpath": payload.get("object_path") or payload.get("path") or "",
                     "object_path": payload.get("object_path") or "",
+                    "source_url": payload.get("source_url") or "",  # NEW: Thêm source_url
                     "folder_key": payload.get("folder_key") or "",
                     "collection_name": collection_name,
                     "academic_year": payload.get("academic_year") or "",
+                    "years": payload.get("years") or [],  # NEW: Thêm years array
                     "chunk_index": payload.get("chunk_index"),
                     "page_number": payload.get("page_number"),
                 }
@@ -126,6 +177,7 @@ class CollectionRouterRetriever:
             query=query,
             collections=target_collections,
             limit=candidate_k,
+            year_scope=year_scope,  # NEW: Pass year_scope để Qdrant Filter
         )
 
         if year_scoped:

core/document_ingest_service.py

@@ -1,5 +1,7 @@
+import hashlib
 import logging
 import os
+import re
 import uuid
 from datetime import datetime, timezone
 from typing import List, Optional
@@ -18,7 +20,7 @@ from qdrant_client.models import (
 )
 
 from .chunking import smart_chunking
-from .config import QDRANT_API_KEY, QDRANT_COLLECTION, QDRANT_URL
+from .config import QDRANT_API_KEY, QDRANT_COLLECTION, QDRANT_URL, SUPABASE_URL, SUPABASE_STORAGE_BUCKET
 from .document_db import Document, DocumentChunk, SessionLocal
 from .models import embeddings
 from .text_utils import clean_text
@@ -26,10 +28,38 @@ from .vectorstore import extract_academic_year, load_documents_from_file
 
 logger = logging.getLogger(__name__)
 
+ACTIVE_CODE_PATTERN = re.compile(r"(20\d{2})\s*[-_/]\s*(20\d{2})")
 _ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt"}
 _ENSURED_PAYLOAD_INDEX_COLLECTIONS = set()
 
 
+def _build_supabase_file_url(object_path: str) -> str:
+    """Tạo URL đầy đủ cho tài liệu từ Supabase Storage."""
+    if not SUPABASE_URL or not SUPABASE_STORAGE_BUCKET or not object_path:
+        return ""
+    
+    clean_path = object_path.lstrip("/")
+    return f"{SUPABASE_URL}/storage/v1/object/public/{SUPABASE_STORAGE_BUCKET}/{clean_path}"
+
+
+def _extract_years_from_academic_year(academic_year: str) -> List[int]:
+    """Trích xuất danh sách năm từ chuỗi năm học (ví dụ '2023-2024' -> [2023, 2024])."""
+    if not academic_year or academic_year == "ALL":
+        return []
+    
+    years = []
+    match = ACTIVE_CODE_PATTERN.search(academic_year)
+    if match:
+        try:
+            start_year = int(match.group(1))
+            end_year = int(match.group(2))
+            years = [start_year, end_year]
+        except (ValueError, IndexError):
+            pass
+    
+    return years
+
+
 def _load_documents_for_ingest(path: str, extension: str) -> List[LangChainDocument]:
     extension = extension.lower()
     if extension not in _ALLOWED_EXTENSIONS:
@@ -111,13 +141,28 @@ def _ensure_payload_indexes(client: QdrantClient, collection_name: str) -> None:
     if collection_name in _ENSURED_PAYLOAD_INDEX_COLLECTIONS:
         return
 
-    for field_name in ("object_path", "document_id"):
+    # KEYWORD indexes cho filtering nhanh
+    for field_name in ("object_path", "document_id", "content_hash"):
+        try:
+            client.create_payload_index(
+                collection_name=collection_name,
+                field_name=field_name,
+                field_schema=PayloadSchemaType.KEYWORD,
+                wait=True,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to create KEYWORD index for {field_name}: {e}")
+    
+    # INTEGER array index cho years
+    try:
         client.create_payload_index(
             collection_name=collection_name,
-            field_name=field_name,
-            field_schema=PayloadSchemaType.KEYWORD,
+            field_name="years",
+            field_schema=PayloadSchemaType.INTEGER,
             wait=True,
         )
+    except Exception as e:
+        logger.warning(f"Failed to create INTEGER index for years: {e}")
 
     _ENSURED_PAYLOAD_INDEX_COLLECTIONS.add(collection_name)
 
@@ -127,6 +172,163 @@ def _is_missing_payload_index_error(error: Exception) -> bool:
     return "Index required but not found" in message
 
 
+def _get_or_create_deduplicated_points(
+    client: QdrantClient,
+    collection_name: str,
+    chunk_docs: List[LangChainDocument],
+    vectors: List,
+    source_object_ref: str,
+    document: Document,
+    source_updated_at: Optional[str],
+    source_etag: Optional[str],
+    created_at: str,
+    effective_source_path: Optional[str] = None,
+) -> tuple[List[PointStruct], List[DocumentChunk]]:
+    """
+    Tích hợp MD5 deduplication: nếu content hash trùng, cập nhật years array thay vì tạo mới.
+    """
+    points: List[PointStruct] = []
+    db_chunk_rows: List[DocumentChunk] = []
+    
+    for index, (chunk_doc, vector) in enumerate(zip(chunk_docs, vectors)):
+        chunk_text = chunk_doc.page_content
+        metadata = chunk_doc.metadata if isinstance(chunk_doc.metadata, dict) else {}
+        
+        # Tính content hash
+        content_hash = hashlib.md5(chunk_text.encode('utf-8')).hexdigest()
+        
+        # Trích académie năm học
+        academic_year = metadata.get("academic_year") or "ALL"
+        years = _extract_years_from_academic_year(academic_year)
+        
+        # Tạo source URL
+        source_url = _build_supabase_file_url(source_object_ref)
+        
+        # Kiểm tra xem content_hash đã tồn tại
+        existing_point_id = None
+        try:
+            existing_points = client.scroll(
+                collection_name=collection_name,
+                limit=1,
+                scroll_filter=Filter(
+                    must=[
+                        FieldCondition(
+                            key="content_hash",
+                            match=MatchValue(value=content_hash),
+                        )
+                    ]
+                ),
+            )
+            
+            if existing_points and existing_points[0]:
+                # Nếu tìm thấy point với hash trùng
+                existing_point_id = existing_points[0][0].id
+                logger.info(f"Tìm thấy content đã tồn tại hash={content_hash[:8]}..., sẽ cập nhật years")
+        except Exception as e:
+            logger.debug(f"Không thể tìm kiếm existing points: {e}")
+        
+        if existing_point_id:
+            # Merge years array
+            try:
+                existing_payload = client.retrieve(collection_name, [existing_point_id])[0].payload
+                existing_years = set(existing_payload.get("years", []))
+                merged_years = sorted(list(set(years) | existing_years))
+                
+                # Update payload với years mới
+                updated_payload = {
+                    **existing_payload,
+                    "years": merged_years,
+                    "document_id": document.id,  # Update document_id nếu tài liệu mới
+                    "source_updated_at": source_updated_at or existing_payload.get("source_updated_at"),
+                }
+                
+                client.update_payload(
+                    collection_name=collection_name,
+                    payload_update=updated_payload,
+                    points=[existing_point_id],
+                )
+                logger.info(f"Đã cập nhật years cho hash {content_hash[:8]}...: {merged_years}")
+            except Exception as e:
+                logger.warning(f"Lỗi cập nhật years cho point đã tồn tại: {e}, sẽ tạo point mới")
+                # Fallback: tạo point mới
+                point_id = str(uuid.uuid4())
+                payload = _build_payload(
+                    document, source_object_ref, chunk_text, index, metadata,
+                    academic_year, years, content_hash, source_url,
+                    source_updated_at, source_etag, created_at, effective_source_path
+                )
+                points.append(PointStruct(id=point_id, vector=vector, payload=payload))
+                db_chunk_rows.append(
+                    DocumentChunk(
+                        document_id=document.id,
+                        chunk_index=index,
+                        content_preview=chunk_text[:200],
+                        qdrant_point_id=point_id,
+                    )
+                )
+        else:
+            # Tạo point mới
+            point_id = str(uuid.uuid4())
+            payload = _build_payload(
+                document, source_object_ref, chunk_text, index, metadata,
+                academic_year, years, content_hash, source_url,
+                source_updated_at, source_etag, created_at, effective_source_path
+            )
+            points.append(PointStruct(id=point_id, vector=vector, payload=payload))
+            db_chunk_rows.append(
+                DocumentChunk(
+                    document_id=document.id,
+                    chunk_index=index,
+                    content_preview=chunk_text[:200],
+                    qdrant_point_id=point_id,
+                )
+            )
+    
+    return points, db_chunk_rows
+
+
+def _build_payload(
+    document: Document,
+    source_object_ref: str,
+    chunk_text: str,
+    index: int,
+    metadata: dict,
+    academic_year: str,
+    years: List[int],
+    content_hash: str,
+    source_url: str,
+    source_updated_at: Optional[str],
+    source_etag: Optional[str],
+    created_at: str,
+    effective_source_path: Optional[str] = None,
+) -> dict:
+    """Xây dựng payload dictionary cho point."""
+    source_name = os.path.basename(source_object_ref) if source_object_ref else document.stored_name
+    source_relpath = source_object_ref or source_name
+    
+    return {
+        "document_id": document.id,
+        "filename": document.original_name,
+        "stored_effective_source_path or name": document.stored_name,
+        "path": document.path,
+        "object_path": source_object_ref,
+        "folder_key": document.folder_key,
+        "collection_name": document.collection_name or "",
+        "source_file": metadata.get("source_file") or source_name,
+        "source_relpath": metadata.get("source_relpath") or source_relpath,
+        "source_url": source_url,
+        "academic_year": academic_year,
+        "years": years,
+        "content_hash": content_hash,
+        "page_number": metadata.get("page_number"),
+        "source_updated_at": source_updated_at,
+        "source_etag": source_etag,
+        "chunk_index": index,
+        "created_at": created_at,
+        "content": chunk_text,
+    }
+
+
 def _delete_existing_document_points(
     client: QdrantClient,
     collection_name: str,
@@ -242,41 +444,20 @@ def process_document_ingest(
         _delete_existing_document_points(client, target_collection, source_object_ref, document.id)
 
         created_at = datetime.now(timezone.utc).isoformat()
-        points: List[PointStruct] = []
-        db_chunk_rows: List[DocumentChunk] = []
-
-        for index, (chunk_doc, vector) in enumerate(zip(chunk_docs, vectors)):
-            chunk_text = chunk_doc.page_content
-            metadata = chunk_doc.metadata if isinstance(chunk_doc.metadata, dict) else {}
-            point_id = str(uuid.uuid4())
-            payload = {
-                "document_id": document.id,
-                "filename": document.original_name,
-                "stored_name": document.stored_name,
-                "path": effective_source_path or document.path,
-                "object_path": source_object_ref,
-                "folder_key": document.folder_key,
-                "collection_name": target_collection,
-                "source_file": metadata.get("source_file") or source_name,
-                "source_relpath": metadata.get("source_relpath") or source_relpath,
-                "academic_year": metadata.get("academic_year") or "ALL",
-                "page_number": metadata.get("page_number"),
-                "source_updated_at": source_updated_at,
-                "source_etag": source_etag,
-                "chunk_index": index,
-                "created_at": created_at,
-                "content": chunk_text,
-            }
-
-            points.append(PointStruct(id=point_id, vector=vector, payload=payload))
-            db_chunk_rows.append(
-                DocumentChunk(
-                    document_id=document.id,
-                    chunk_index=index,
-                    content_preview=chunk_text[:200],
-                    qdrant_point_id=point_id,
-                )
-            )
+        
+        # NEW: Sử dụng deduplication logic
+        points, db_chunk_rows = _get_or_create_deduplicated_points(
+            client=client,
+            collection_name=target_collection,
+            chunk_docs=chunk_docs,
+            vectors=vectors,
+            source_object_ref=source_object_ref,
+            document=document,
+            source_updated_at=source_updated_at,
+            source_etag=source_etag,
+            created_at=created_at,
+            effective_source_path=effective_source_path,
+        )
 
         client.upsert(collection_name=target_collection, points=points, wait=True)
 

core/retriever.py

@@ -13,6 +13,57 @@ class HybridRetriever:
         self.rrf_c = 60
         print(" BM25 sẵn sàng!")
 
+    @staticmethod
+    def _filter_by_year_scope(documents: List, year_scope: str | None) -> List:
+        """Filter documents theo year_scope (ví dụ: '2023-2024' hoặc '2023')."""
+        if not year_scope:
+            return documents
+        
+        filtered = []
+        year_targets = set()
+        
+        # Parse year_scope: có thể là "2023-2024" hoặc "2023"
+        if "-" in year_scope:
+            parts = year_scope.split("-")
+            try:
+                year_targets = {int(p.strip()) for p in parts if p.strip()}
+            except ValueError:
+                return documents
+        else:
+            try:
+                year_targets = {int(year_scope.strip())}
+            except ValueError:
+                return documents
+        
+        for doc in documents:
+            metadata = doc.metadata if isinstance(doc.metadata, dict) else {}
+            
+            # Check years array (mới)
+            doc_years = metadata.get("years", [])
+            if isinstance(doc_years, list) and any(y in year_targets for y in doc_years):
+                filtered.append(doc)
+                continue
+            
+            # Check academic_year string (cũ, để backwards compatibility)
+            academic_year = metadata.get("academic_year", "")
+            if academic_year and academic_year != "ALL":
+                doc_year_tokens = set()
+                for potential_year in academic_year.split("-"):
+                    try:
+                        doc_year_tokens.add(int(potential_year.strip()))
+                    except ValueError:
+                        pass
+                
+                if doc_year_tokens.intersection(year_targets):
+                    filtered.append(doc)
+                    continue
+            
+            # Include ALL documents không có year info
+            if not doc_years and academic_year == "ALL":
+                filtered.append(doc)
+        
+        return filtered if filtered else documents
+
     @staticmethod
     def _doc_key(doc) -> str:
         metadata = doc.metadata if isinstance(doc.metadata, dict) else {}
@@ -23,7 +74,6 @@ class HybridRetriever:
         return f"{source}|{page}|{digest}"
 
     def search(self, query: str, k: int = 10, alpha: float = 0.6, year_scope: str | None = None) -> List:
-        del year_scope
         if not self.documents or k <= 0:
             return []
 
@@ -34,7 +84,15 @@ class HybridRetriever:
         # Lấy top k từ BM25
         tokenized_query = query.lower().split()
         candidate_k = min(max(k * 4, k), len(self.documents))
-        bm25_top_docs = self.bm25.get_top_n(tokenized_query, self.documents, n=candidate_k)
+        
+        # Filter documents theo year_scope nếu có
+        docs_to_search = self.documents
+        if year_scope:
+            docs_to_search = self._filter_by_year_scope(self.documents, year_scope)
+            if not docs_to_search:
+                docs_to_search = self.documents  # Fallback nếu không có doc match year
+        
+        bm25_top_docs = self.bm25.get_top_n(tokenized_query, docs_to_search, n=candidate_k)
 
         bm25_ranked = {}
         all_retrieved = {}