update process document

README.md

@@ -44,24 +44,23 @@ Khi server bat dau, `lifespan` trong `main.py` chay theo thu tu:
 1. Doc bien moi truong tu `core/config.py`.
 2. Tao pool ket noi PostgreSQL (`asyncpg`) va dam bao bang `history` ton tai.
 3. Ket noi Qdrant Cloud.
-4. Neu collection `quy_che_db` chua ton tai: build vectorstore moi bang `core/vectorstore.py`.
-5. Neu da ton tai: tai vectorstore va chunks da luu.
-6. Khoi tao `HybridRetriever` trong `core/retriever.py`.
+4. Khoi tao `CollectionRouterRetriever` de tim theo cac collection dang active tren Qdrant.
+5. Khoi tao Supabase sync coordinator va chay `startup:initial_sync` thong qua `build_vectorstore_improved` trong `core/vectorstore.py` (co the cho toi da theo `SUPABASE_STARTUP_SYNC_WAIT_SECONDS` hoac chay nen).
+6. Bat polling sync dinh ky de dong bo thay doi add/update/delete tu Supabase.
 7. Danh dau app san sang (endpoint `/healthz` se bao `ready=true`).
 
-## 3) Luong ingest tai lieu (xay dung vectorstore)
+## 3) Luong ingest tai lieu (Supabase-only)
 
-Luong nay nam trong `core/vectorstore.py`:
+Luong nay duoc kich hoat boi scheduler/event sync trong `core/supabase_sync_service.py` va ingest trong `core/document_ingest_service.py`:
 
-1. Quet de quy file trong thu muc `data/` (`.pdf`, `.doc`, `.docx`), bao gom ca cac thu muc nam hoc nhu `So tay sinh vien 2022-2023/`.
-2. Trich xuat noi dung (giu bang tu PDF/DOCX).
-3. Lam sach text bang `core/text_utils.py`.
-4. Gan metadata `academic_year` cho tung tai lieu/chunk (neu tim thay mau nam hoc `YYYY-YYYY` trong duong dan hoac ten file).
-5. Chunk van ban thong minh bang `core/chunking.py`.
+1. Lay danh sach object tu Supabase Storage va diff voi snapshot de xac dinh `added/updated/deleted`.
+2. Download tung file can ingest ve file tam.
+3. Trich xuat noi dung tai lieu bang bo ham cu (`load_documents_from_file` trong `core/vectorstore.py`).
+4. Lam sach text bang bo ham cu (`clean_text` trong `core/text_utils.py`).
+5. Chunk van ban bang bo ham cu (`smart_chunking` trong `core/chunking.py`).
 6. Embedding chunks bang model trong `core/models.py`.
-7. Day vector len Qdrant collection `quy_che_db`.
-8. Luu ban sao chunks local vao `vectorstore/chunks.pkl` de startup nhanh hon.
-9. Neu phat hien file moi trong `data/` ma chunks cache chua co, he thong tu dong rebuild de dong bo du lieu.
+7. Upsert vector len Qdrant collection theo folder nam hoc.
+8. Xoa/ghi de theo `object_path` de dam bao incremental sync va tranh duplicate.
 
 ## 3.1) Hoi va tra loi theo nam hoc
 
@@ -120,7 +119,7 @@ Tuong tu luong `/chat`, khac o cho:
 
 ### Du lieu va vector
 
-- `core/vectorstore.py`: Load tai lieu, tien xu ly, chunking, embedding, tao/tai Qdrant vector store, luu chunks local.
+- `core/vectorstore.py`: Cung cap bo ham xu ly tai lieu cu (doc PDF/DOCX, metadata nam hoc) duoc tai su dung trong luong Supabase ingest, dong thoi chua `build_vectorstore_improved`/`load_vectorstore_improved` cho luong Supabase.
 - `core/chunking.py`: Cat van ban thong minh (uu tien giu cau truc bang/danh sach).
 - `core/text_utils.py`: Lam sach va chuan hoa noi dung text truoc khi embedding.
 - `core/models.py`: Khoi tao embedding model va cross-encoder model.
@@ -176,4 +175,4 @@ Sau khi chay, kiem tra:
 ## 9) Ghi chu
 
 - Trong Hugging Face Spaces, frontmatter o dau file README can duoc giu nguyen.
-- Lan chay dau co the cham do qua trinh doc tai lieu, chunk, embedding va day vector len Qdrant.
+- Lan chay dau co the cham do qua trinh initial sync tu Supabase (download + chunk + embedding + upsert Qdrant).

api/admin_documents_router.py

@@ -1,148 +0,0 @@
-import os
-import uuid
-from typing import Any, Dict, List
-
-from fastapi import APIRouter, BackgroundTasks, Depends, File, HTTPException, Query, UploadFile
-from fastapi.concurrency import run_in_threadpool
-from sqlalchemy.orm import Session
-
-from core.config import MAX_UPLOAD_SIZE_MB, UPLOAD_DIR
-from core.document_db import Document, get_document_db
-from core.document_ingest_service import run_document_ingest_task
-
-router = APIRouter(prefix="/admin/documents", tags=["admin-documents"])
-
-_ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt"}
-
-
-class FileTooLargeError(Exception):
-    pass
-
-
-def _save_upload_file_stream(file_obj: Any, destination: str, max_size_bytes: int) -> int:
-    total_size = 0
-    chunk_size = 1024 * 1024
-
-    with open(destination, "wb") as output:
-        while True:
-            chunk = file_obj.read(chunk_size)
-            if not chunk:
-                break
-
-            total_size += len(chunk)
-            if total_size > max_size_bytes:
-                raise FileTooLargeError("Uploaded file exceeds configured maximum size.")
-
-            output.write(chunk)
-
-    return total_size
-
-
-@router.post("/upload")
-async def upload_document(
-    background_tasks: BackgroundTasks,
-    file: UploadFile = File(...),
-    db: Session = Depends(get_document_db),
-) -> Dict[str, Any]:
-    if not file.filename:
-        raise HTTPException(status_code=400, detail="File name is required.")
-
-    extension = os.path.splitext(file.filename)[1].lower()
-    if extension not in _ALLOWED_EXTENSIONS:
-        raise HTTPException(status_code=400, detail="Unsupported file type. Allowed: .pdf, .docx, .txt")
-
-    os.makedirs(UPLOAD_DIR, exist_ok=True)
-    stored_name = f"{uuid.uuid4()}{extension}"
-    stored_path = os.path.abspath(os.path.join(UPLOAD_DIR, stored_name))
-    max_size_bytes = MAX_UPLOAD_SIZE_MB * 1024 * 1024
-
-    try:
-        file.file.seek(0)
-        size = await run_in_threadpool(
-            _save_upload_file_stream,
-            file.file,
-            stored_path,
-            max_size_bytes,
-        )
-    except FileTooLargeError:
-        if os.path.exists(stored_path):
-            os.remove(stored_path)
-        raise HTTPException(
-            status_code=413,
-            detail=f"File is too large. Max allowed size is {MAX_UPLOAD_SIZE_MB} MB.",
-        )
-    except Exception as error:
-        if os.path.exists(stored_path):
-            os.remove(stored_path)
-        raise HTTPException(status_code=500, detail=f"Failed to save file: {error}")
-    finally:
-        await file.close()
-
-    document = Document(
-        original_name=file.filename,
-        stored_name=stored_name,
-        path=stored_path,
-        mime_type=file.content_type or "application/octet-stream",
-        size=size,
-        status="pending",
-        total_chunks=0,
-    )
-    db.add(document)
-    db.commit()
-    db.refresh(document)
-
-    background_tasks.add_task(run_document_ingest_task, document.id)
-
-    return {
-        "status": "success",
-        "document_id": document.id,
-        "original_name": document.original_name,
-        "stored_name": document.stored_name,
-        "path": document.path,
-    }
-
-
-@router.get("/status/{document_id}")
-def get_document_status(document_id: str, db: Session = Depends(get_document_db)) -> Dict[str, Any]:
-    document = db.query(Document).filter(Document.id == document_id).first()
-    if document is None:
-        raise HTTPException(status_code=404, detail="Document not found.")
-
-    return {
-        "status": "success",
-        "document_id": document.id,
-        "processing_status": document.status,
-        "total_chunks": document.total_chunks,
-        "error_message": document.error_message,
-        "created_at": document.created_at,
-    }
-
-
-@router.get("")
-def list_documents(
-    limit: int = Query(default=20, ge=1, le=100),
-    offset: int = Query(default=0, ge=0),
-    db: Session = Depends(get_document_db),
-) -> Dict[str, Any]:
-    records = (
-        db.query(Document)
-        .order_by(Document.created_at.desc())
-        .offset(offset)
-        .limit(limit)
-        .all()
-    )
-
-    return {
-        "status": "success",
-        "items": [
-            {
-                "id": doc.id,
-                "original_name": doc.original_name,
-                "stored_name": doc.stored_name,
-                "status": doc.status,
-                "total_chunks": doc.total_chunks,
-                "created_at": doc.created_at,
-            }
-            for doc in records
-        ],
-    }

core/config.py

@@ -18,12 +18,6 @@ def _is_hf_persistent_storage_available() -> bool:
 _USE_HF_PERSISTENT_STORAGE = _is_hf_persistent_storage_available()
 
 
-def _default_upload_dir() -> str:
-    if _USE_HF_PERSISTENT_STORAGE:
-        return '/data/uploads'
-    return 'uploads'
-
-
 def _default_documents_db_url() -> str:
     if _USE_HF_PERSISTENT_STORAGE:
         return 'sqlite:////data/rag_metadata.db'
@@ -54,10 +48,6 @@ CHUNK_OVERLAP = int(os.getenv('CHUNK_OVERLAP', '150'))
 TOP_K_RESULTS = int(os.getenv('TOP_K_RESULTS', '10'))
 FINAL_TOP_K = int(os.getenv('FINAL_TOP_K', '5'))
 
-DATA_DIR = os.getenv('DATA_DIR', 'data')
-VECTOR_DIR = os.getenv('VECTOR_DIR', 'vectorstore')
-UPLOAD_DIR = os.getenv('UPLOAD_DIR', _default_upload_dir())
-MAX_UPLOAD_SIZE_MB = int(os.getenv('MAX_UPLOAD_SIZE_MB', '20'))
 QDRANT_COLLECTION = os.getenv('QDRANT_COLLECTION', 'rag_docs')
 DOCUMENTS_DATABASE_URL = os.getenv('DOCUMENTS_DATABASE_URL', _default_documents_db_url())
 

core/document_ingest_service.py

@@ -1,13 +1,10 @@
 import logging
 import os
-import re
 import uuid
 from datetime import datetime, timezone
 from typing import List, Optional
 
-from docx import Document as DocxDocument
-from fastapi.concurrency import run_in_threadpool
-from pypdf import PdfReader
+from langchain_core.documents import Document as LangChainDocument
 from qdrant_client import QdrantClient
 from qdrant_client.http.exceptions import UnexpectedResponse
 from qdrant_client.models import (
@@ -20,78 +17,71 @@ from qdrant_client.models import (
     VectorParams,
 )
 
-from .config import CHUNK_OVERLAP, CHUNK_SIZE, QDRANT_API_KEY, QDRANT_COLLECTION, QDRANT_URL
+from .chunking import smart_chunking
+from .config import QDRANT_API_KEY, QDRANT_COLLECTION, QDRANT_URL
 from .document_db import Document, DocumentChunk, SessionLocal
 from .models import embeddings
+from .text_utils import clean_text
+from .vectorstore import extract_academic_year, load_documents_from_file
 
 logger = logging.getLogger(__name__)
 
 _ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt"}
-_WHITESPACE_RE = re.compile(r"\s+")
-_TOKEN_RE = re.compile(r"\S+")
 
 
-def normalize_text(text: str) -> str:
-    if not text:
-        return ""
-
-    cleaned = text.replace("\x00", " ")
-    cleaned = cleaned.replace("\ufeff", " ")
-    cleaned = cleaned.replace("\u200b", " ").replace("\u200c", " ").replace("\u200d", " ")
-    cleaned = _WHITESPACE_RE.sub(" ", cleaned)
-    return cleaned.strip()
-
-
-def read_document_content(path: str, extension: str) -> str:
+def _load_documents_for_ingest(path: str, extension: str) -> List[LangChainDocument]:
     extension = extension.lower()
     if extension not in _ALLOWED_EXTENSIONS:
         raise ValueError(f"Unsupported file extension: {extension}")
 
-    if extension == ".pdf":
-        reader = PdfReader(path)
-        page_texts = [(page.extract_text() or "") for page in reader.pages]
-        return "\n".join(page_texts)
+    return load_documents_from_file(path, os.path.basename(path))
 
-    if extension == ".docx":
-        doc = DocxDocument(path)
-        paragraphs = [p.text for p in doc.paragraphs if p.text]
 
-        for table in doc.tables:
-            for row in table.rows:
-                row_cells = [cell.text.strip() for cell in row.cells]
-                if any(row_cells):
-                    paragraphs.append(" | ".join(row_cells))
+def _clean_documents_for_ingest(docs: List[LangChainDocument], source_name: str) -> List[LangChainDocument]:
+    cleaned_docs: List[LangChainDocument] = []
 
-        return "\n".join(paragraphs)
+    for index, doc in enumerate(docs, 1):
+        cleaned = clean_text(doc.page_content)
+        if not cleaned or len(cleaned.split()) < 20:
+            continue
 
-    with open(path, "r", encoding="utf-8", errors="ignore") as file:
-        return file.read()
+        metadata = doc.metadata.copy() if isinstance(doc.metadata, dict) else {}
+        page_number = metadata.get("page")
+        if page_number is None:
+            page_number = index
 
+        metadata["source_file"] = source_name
+        metadata["page_number"] = page_number
 
-def chunk_text_by_tokens(text: str, chunk_size: int, overlap: int) -> List[str]:
-    if chunk_size <= 0:
-        raise ValueError("CHUNK_SIZE must be > 0")
-    if overlap < 0:
-        raise ValueError("CHUNK_OVERLAP must be >= 0")
-    if overlap >= chunk_size:
-        raise ValueError("CHUNK_OVERLAP must be smaller than CHUNK_SIZE")
+        cleaned_docs.append(
+            LangChainDocument(
+                page_content=cleaned,
+                metadata=metadata,
+            )
+        )
 
-    tokens = _TOKEN_RE.findall(text)
-    if not tokens:
-        return []
+    return cleaned_docs
 
-    step = chunk_size - overlap
-    chunks: List[str] = []
 
-    for start in range(0, len(tokens), step):
-        end = min(start + chunk_size, len(tokens))
-        piece = " ".join(tokens[start:end]).strip()
-        if piece:
-            chunks.append(piece)
-        if end >= len(tokens):
-            break
+def chunk_documents_for_ingest(
+    path: str,
+    extension: str,
+    source_name: str,
+    source_relpath: str,
+) -> List[LangChainDocument]:
+    loaded_docs = _load_documents_for_ingest(path, extension)
+    cleaned_docs = _clean_documents_for_ingest(loaded_docs, source_name)
+    if not cleaned_docs:
+        return []
+
+    academic_year = extract_academic_year(source_relpath) or "ALL"
+    for doc in cleaned_docs:
+        metadata = doc.metadata.copy() if isinstance(doc.metadata, dict) else {}
+        metadata["source_relpath"] = source_relpath
+        metadata["academic_year"] = academic_year
+        doc.metadata = metadata
 
-    return chunks
+    return [doc for doc in smart_chunking(cleaned_docs) if (doc.page_content or "").strip()]
 
 
 def _parse_datetime(value: Optional[str]):
@@ -193,6 +183,7 @@ def process_document_ingest(
 
     effective_file_path = (file_path or "").strip()
     effective_source_path = (source_path or "").strip()
+    source_object_ref = (source_object_path or "").strip()
 
     try:
         document = db.query(Document).filter(Document.id == document_id).first()
@@ -204,18 +195,25 @@ def process_document_ingest(
         document.error_message = None
         db.commit()
 
-        ingest_file_path = effective_file_path or document.path
-        if not ingest_file_path:
-            raise ValueError("Document file path is missing for ingest.")
+        if not effective_file_path:
+            raise ValueError("Supabase-only ingest requires downloaded file_path.")
+        if not source_object_ref:
+            raise ValueError("Supabase-only ingest requires source_object_path.")
 
-        source_object_ref = (source_object_path or document.object_path or "").strip() or None
+        ingest_file_path = effective_file_path
 
         extension_source = source_object_ref or document.stored_name or ingest_file_path
         _, extension = os.path.splitext(extension_source)
 
-        raw_text = read_document_content(ingest_file_path, extension)
-        normalized = normalize_text(raw_text)
-        chunks = chunk_text_by_tokens(normalized, CHUNK_SIZE, CHUNK_OVERLAP)
+        source_name = os.path.basename(source_object_ref or document.stored_name or ingest_file_path)
+        source_relpath = source_object_ref or source_name
+        chunk_docs = chunk_documents_for_ingest(
+            path=ingest_file_path,
+            extension=extension,
+            source_name=source_name,
+            source_relpath=source_relpath,
+        )
+        chunks = [doc.page_content for doc in chunk_docs]
 
         if not chunks:
             raise ValueError("Document has no readable content after normalization.")
@@ -240,7 +238,9 @@ def process_document_ingest(
         points: List[PointStruct] = []
         db_chunk_rows: List[DocumentChunk] = []
 
-        for index, (chunk_text, vector) in enumerate(zip(chunks, vectors)):
+        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,
@@ -250,6 +250,10 @@ def process_document_ingest(
                 "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,
@@ -362,8 +366,3 @@ def delete_vectors_for_object_path(collection_name: str, object_path: str) -> bo
         )
 
     return True
-
-
-async def run_document_ingest_task(document_id: str) -> None:
-    # Heavy ingest work runs in threadpool to keep event loop responsive.
-    await run_in_threadpool(process_document_ingest, document_id)

core/vectorstore.py

@@ -1,33 +1,22 @@
+import asyncio
+import logging
 import os
 import re
-from typing import List, Tuple
-from langchain_qdrant import QdrantVectorStore
-from qdrant_client import QdrantClient
-from qdrant_client.http.models import Distance, VectorParams
-from docx import Document
-from .models import embeddings
-from .text_utils import clean_text
-from .chunking import smart_chunking
-from .config import DATA_DIR, VECTOR_DIR, QDRANT_API_KEY, QDRANT_URL, QDRANT_COLLECTION
-from langchain_core.documents import Document as LangChainDocument
-import zipfile
-import xml.etree.ElementTree as ET
-import pickle
+from typing import Any, Dict, List
+
 import pdfplumber
+from docx import Document
 from docx.document import Document as _Document
-from docx.oxml.text.paragraph import CT_P
 from docx.oxml.table import CT_Tbl
-from docx.table import _Cell, Table
+from docx.oxml.text.paragraph import CT_P
+from docx.table import Table, _Cell
 from docx.text.paragraph import Paragraph
-import logging
+from langchain_core.documents import Document as LangChainDocument
+
+from .text_utils import clean_text
 
-logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-CHUNKS_PICKLE = os.path.join(VECTOR_DIR, "chunks.pkl")
-COLLECTION_NAME = QDRANT_COLLECTION
-# [YEAR-AWARE CHANGE] Ho tro quet de quy va gan metadata nam hoc.
-SUPPORTED_FORMATS = ('.pdf', '.doc', '.docx')
 ACADEMIC_YEAR_PATTERN = re.compile(r"(20\d{2})\s*[-_]\s*(20\d{2})")
 
 
@@ -44,143 +33,52 @@ def extract_academic_year(text: str) -> str:
     return normalize_academic_year(match.group(1), match.group(2))
 
 
-def discover_data_files() -> List[Tuple[str, str, str, str]]:
-    """Quet de quy thu muc data va tra ve (filepath, filename, relpath, academic_year)."""
-    if not os.path.isdir(DATA_DIR):
-        return []
-
-    discovered = []
-    for root, _, files in os.walk(DATA_DIR):
-        for filename in files:
-            if not filename.lower().endswith(SUPPORTED_FORMATS):
-                continue
-
-            filepath = os.path.join(root, filename)
-            relpath = os.path.relpath(filepath, DATA_DIR)
-            year = extract_academic_year(relpath) or "ALL"
-            discovered.append((filepath, filename, relpath, year))
-
-    discovered.sort(key=lambda x: x[2].lower())
-    return discovered
-
-
-def collect_chunk_relpaths(chunks: List) -> set:
-    relpaths = set()
-    for chunk in chunks:
-        metadata = chunk.metadata if isinstance(chunk.metadata, dict) else {}
-        relpath = metadata.get("source_relpath")
-        if relpath:
-            relpaths.add(os.path.normpath(str(relpath)))
-    return relpaths
-
-
-def enrich_chunk_metadata(chunks: List) -> bool:
-    """Bo sung metadata nam hoc cho chunks cu de dam bao loc theo nam hoat dong."""
-    changed = False
-    for chunk in chunks:
-        metadata = chunk.metadata if isinstance(chunk.metadata, dict) else {}
-
-        source = metadata.get("source")
-        source_file = metadata.get("source_file")
-        source_relpath = metadata.get("source_relpath")
-
-        if not source_relpath:
-            if source and os.path.isabs(str(source)):
-                try:
-                    source_relpath = os.path.relpath(source, DATA_DIR)
-                except Exception:
-                    source_relpath = str(source)
-            elif source:
-                source_relpath = str(source)
-            elif source_file:
-                source_relpath = str(source_file)
-
-            if source_relpath:
-                metadata["source_relpath"] = source_relpath
-                changed = True
-
-        if not metadata.get("academic_year"):
-            year = extract_academic_year(source_relpath or source_file or "") or "ALL"
-            metadata["academic_year"] = year
-            changed = True
-
-        if "page_number" not in metadata and metadata.get("page") is not None:
-            metadata["page_number"] = metadata.get("page")
-            changed = True
-
-        chunk.metadata = metadata
-
-    return changed
-
-
-# [YEAR-AWARE CHANGE] Gom doc tu toan bo thu muc data theo cau truc nam hoc.
-def load_and_clean_all_docs() -> List[LangChainDocument]:
-    docs: List[LangChainDocument] = []
-    file_entries = discover_data_files()
-
-    if not file_entries:
-        logger.error(" Không tìm thấy file PDF, DOC, hoặc DOCX!")
-        return docs
-
-    for filepath, filename, relpath, academic_year in file_entries:
-        logger.info(f" Đang đọc: {relpath}")
-        loaded_docs = load_documents_from_file(filepath, filename)
-
-        for i, doc in enumerate(loaded_docs, 1):
-            cleaned = clean_text(doc.page_content)
-            if not cleaned or len(cleaned.split()) < 20:
-                continue
-
-            page_number = doc.metadata.get("page") if isinstance(doc.metadata, dict) else None
-            if page_number is None:
-                page_number = i
-
-            doc.metadata["source_file"] = filename
-            doc.metadata["source_relpath"] = relpath
-            doc.metadata["academic_year"] = academic_year
-            doc.metadata["page_number"] = page_number
-            doc.page_content = cleaned
-            docs.append(doc)
-
-    return docs
-
 def table_to_markdown(data: List[List[str]]) -> str:
     if not data or len(data) < 2:
         return ""
-    header = data[0]
-    header = [str(cell).replace('\n', ' ').strip() if cell else "" for cell in header]
+
+    header = [str(cell).replace("\n", " ").strip() if cell else "" for cell in data[0]]
     separator = ["---"] * len(header)
-    markdown_lines = []
-    markdown_lines.append("| " + " | ".join(header) + " |")
-    markdown_lines.append("| " + " | ".join(separator) + " |")
+    markdown_lines = [
+        "| " + " | ".join(header) + " |",
+        "| " + " | ".join(separator) + " |",
+    ]
+
     for row in data[1:]:
-        clean_row = [str(cell).replace('\n', '<br>').strip() if cell else "" for cell in row]
+        clean_row = [str(cell).replace("\n", "<br>").strip() if cell else "" for cell in row]
         markdown_lines.append("| " + " | ".join(clean_row) + " |")
+
     return "\n".join(markdown_lines) + "\n\n"
 
+
 def read_pdf_with_tables(filepath: str) -> List[LangChainDocument]:
-    docs = []
+    docs: List[LangChainDocument] = []
     try:
         with pdfplumber.open(filepath) as pdf:
-            for i, page in enumerate(pdf.pages, 1):
+            for page_index, page in enumerate(pdf.pages, 1):
                 text = page.extract_text() or ""
                 tables = page.extract_tables()
-                table_texts = []
+                table_texts: List[str] = []
                 if tables:
                     for table in tables:
                         md_table = table_to_markdown(table)
                         if md_table:
                             table_texts.append(md_table)
-                full_content = text + "\n\n[BẢNG DỮ LIỆU TRÍCH XUẤT]:\n" + "\n".join(table_texts)
+
+                full_content = text + "\n\n[BANG DU LIEU TRICH XUAT]:\n" + "\n".join(table_texts)
                 if full_content.strip():
-                    docs.append(LangChainDocument(
-                        page_content=full_content,
-                        metadata={"source": filepath, "page": i}
-                    ))
-    except Exception as e:
-        logger.error(f"Lỗi đọc PDF (pdfplumber) {os.path.basename(filepath)}: {e}")
+                    docs.append(
+                        LangChainDocument(
+                            page_content=full_content,
+                            metadata={"source": filepath, "page": page_index},
+                        )
+                    )
+    except Exception as error:
+        logger.error("Loi doc PDF (pdfplumber) %s: %s", os.path.basename(filepath), error)
+
     return docs
 
+
 def iter_block_items(parent):
     if isinstance(parent, _Document):
         parent_elm = parent.element.body
@@ -194,206 +92,103 @@ def iter_block_items(parent):
         elif isinstance(child, CT_Tbl):
             yield Table(child, parent)
 
+
 def read_docx_with_tables(filepath: str) -> str:
     doc = Document(filepath)
-    full_text = []
+    full_text: List[str] = []
     for block in iter_block_items(doc):
         if isinstance(block, Paragraph):
             if block.text.strip():
                 full_text.append(block.text.strip())
         elif isinstance(block, Table):
-            table_data = []
+            table_data: List[List[str]] = []
             for row in block.rows:
-                row_data = []
+                row_data: List[str] = []
                 for cell in row.cells:
-                    cell_text = clean_text(cell.text)
-                    row_data.append(cell_text)
+                    row_data.append(clean_text(cell.text))
                 table_data.append(row_data)
+
             md_table = table_to_markdown(table_data)
             if md_table:
                 full_text.append(f"\n{md_table}\n")
+
     return "\n".join(full_text)
 
-def extract_text_from_doc_com(filepath: str) -> str:
-    try:
-        import win32com.client
-        word = win32com.client.Dispatch("Word.Application")
-        word.Visible = False
-        doc = word.Documents.Open(os.path.abspath(filepath))
-        text = doc.Range().Text
-        doc.Close()
-        word.Quit()
-        return text.strip()
-    except Exception as e:
-        logger.error(f" COM API lỗi: {str(e)[:40]}")
-        return ""
 
-def extract_text_from_doc(filepath: str) -> str:
-    try:
-        doc = Document(filepath)
-        text = "\n".join([para.text for para in doc.paragraphs])
-        if text.strip():
-            return text
-    except Exception as e:
-        logger.error(f"Lỗi đọc DOC {os.path.basename(filepath)}: {e}")
-    try:
-        with zipfile.ZipFile(filepath, 'r') as zip_ref:
-            xml_content = zip_ref.read('word/document.xml')
-            root = ET.fromstring(xml_content)
-            ns = {'w': 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'}
-            paragraphs = root.findall('.//w:p', ns)
-            text_list = []
-            for para in paragraphs:
-                texts = para.findall('.//w:t', ns)
-                para_text = ''.join([t.text for t in texts if t.text])
-                if para_text.strip():
-                    text_list.append(para_text)
-            return "\n".join(text_list)
-    except Exception as e:
-        logger.error(f" Lỗi đọc DOCX {os.path.basename(filepath)}: {e}")
-        pass
-    if filepath.lower().endswith('.doc'):
-        return extract_text_from_doc_com(filepath)
-    return ""
-
-def load_doc_file(filepath: str) -> List[LangChainDocument]:
-    docs = []
-    try:
-        text = extract_text_from_doc(filepath)
-        if text.strip():
-            docs.append(LangChainDocument(page_content=text, metadata={"source": filepath}))
-        else:
-            logger.warning(f" File rỗng: {os.path.basename(filepath)}")
-    except Exception as e:
-        logger.error(f" Không thể đọc {os.path.basename(filepath)}: {str(e)[:60]}")
-    return docs
+def load_documents_from_file(filepath: str, filename: str) -> List[LangChainDocument]:
+    docs: List[LangChainDocument] = []
+    lower_name = filename.lower()
 
-def load_documents_from_file(filepath: str, filename: str) -> List:
-    docs = []
     try:
-        if filename.lower().endswith('.pdf'):
+        if lower_name.endswith(".pdf"):
             docs = read_pdf_with_tables(filepath)
-        elif filename.lower().endswith('.docx'):
+        elif lower_name.endswith(".docx"):
             text = read_docx_with_tables(filepath)
             if text:
                 docs = [LangChainDocument(page_content=text, metadata={"source": filepath})]
-        elif filename.lower().endswith('.doc'):
-            docs = load_doc_file(filepath)
-        
+        elif lower_name.endswith(".txt"):
+            with open(filepath, "r", encoding="utf-8", errors="ignore") as input_file:
+                text = input_file.read()
+            if text and text.strip():
+                docs = [LangChainDocument(page_content=text, metadata={"source": filepath})]
+
         if docs:
-            logger.info(f" Đã đọc: {filename}")
+            logger.info("Da doc: %s", filename)
+
         return docs
-    except Exception as e:
-        logger.error(f" Lỗi đọc {filename}: {str(e)[:60]}")
+    except Exception as error:
+        logger.error("Loi doc %s: %s", filename, str(error)[:120])
         return []
 
-# [YEAR-AWARE CHANGE] Cho phep tao lai collection khi phat hien file moi.
-def build_vectorstore_improved(recreate_collection: bool = False) -> Tuple[QdrantVectorStore, List]:
-    logger.info(" Đang xây dựng vectorstore...")
-    docs = load_and_clean_all_docs()
-    
-    if not docs:
-        logger.error(" Không có văn bản hợp lệ!")
-        return None, []
-    
-    logger.info(f" Đã đọc {len(docs)} trang hợp lệ")
-    chunks = smart_chunking(docs)
-    logger.info ("Đang kết nối với và đẩy dữ liệu lên Qdrant Cloud ")
-
-    client = QdrantClient(
-        url=QDRANT_URL,
-        api_key=QDRANT_API_KEY
-    )
-    
-    if client.collection_exists(COLLECTION_NAME):
-        if recreate_collection:
-            logger.warning(f"Collection {COLLECTION_NAME} đã tồn tại. Đang tạo lại để đồng bộ dữ liệu mới...")
-            client.delete_collection(collection_name=COLLECTION_NAME)
-            client.create_collection(
-                collection_name=COLLECTION_NAME,
-                vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
-            )
-    else:
-        client.create_collection(
-            collection_name=COLLECTION_NAME,
-            vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
-        )
 
-    db = QdrantVectorStore(
-        client=client,
-        collection_name=COLLECTION_NAME,
-        embedding=embeddings,
+async def build_vectorstore_improved(
+    sync_coordinator: Any,
+    startup_wait_seconds: int = 5,
+) -> Dict[str, Any]:
+    """Supabase build step: trigger one initial sync and optionally wait for completion."""
+    if sync_coordinator is None:
+        raise ValueError("sync_coordinator is required")
+
+    startup_sync_task = asyncio.create_task(
+        sync_coordinator.run_sync(
+            trigger="startup:initial_sync",
+            queue_if_locked=False,
+        )
     )
-    #Đẩy chunks lên cloud 
-    db.add_documents(chunks)
 
-    #Lưu chunk local 
+    if startup_wait_seconds <= 0:
+        return {
+            "task": startup_sync_task,
+            "initial_sync": None,
+            "timed_out": True,
+        }
+
     try:
-        os.makedirs(VECTOR_DIR, exist_ok=True)
-        with open(CHUNKS_PICKLE, 'wb') as f:
-            pickle.dump(chunks, f)
-        logger.info(f" Đã lưu chunks vào {CHUNKS_PICKLE}")
-    except Exception as e:
-        logger.error(f" Không thể lưu chunks: {e}")
-
-    logger.info(" Hoàn tất xây dựng và đưa lên Qdrant Cloud")
-    return db, chunks
-
-def load_vectorstore_improved() -> Tuple[QdrantVectorStore, List]:
-    logger.info("Đang tải vectorstore  t�� Qdrant Cloud")
-    
-    client = QdrantClient(
-        url=QDRANT_URL,
-        api_key=QDRANT_API_KEY
-    )
-    db = QdrantVectorStore(
-        client=client,
-        collection_name=COLLECTION_NAME,
-        embedding=embeddings
-    )
-    # Load chunks từ file pickle nếu có, để tránh phải tái tạo từ file nguồn mỗi lần khởi động
-    if os.path.exists(CHUNKS_PICKLE):
-        try:
-            with open(CHUNKS_PICKLE, 'rb') as f:
-                chunks = pickle.load(f)
-            if enrich_chunk_metadata(chunks):
-                try:
-                    os.makedirs(VECTOR_DIR, exist_ok=True)
-                    with open(CHUNKS_PICKLE, 'wb') as f:
-                        pickle.dump(chunks, f)
-                    logger.info(" Đã cập nhật metadata năm học cho chunks local")
-                except Exception as e:
-                    logger.error(f" Không thể cập nhật {CHUNKS_PICKLE}: {e}")
-
-            # [YEAR-AWARE CHANGE] Neu co file moi theo nam hoc, rebuild de dong bo Qdrant.
-            discovered_relpaths = {os.path.normpath(relpath) for _, _, relpath, _ in discover_data_files()}
-            chunk_relpaths = collect_chunk_relpaths(chunks)
-            missing_relpaths = sorted(discovered_relpaths - chunk_relpaths)
-
-            if missing_relpaths:
-                logger.warning(
-                    f" Phát hiện {len(missing_relpaths)} file mới chưa có trong chunks cache. Đang build lại vectorstore theo dữ liệu hiện tại..."
-                )
-                return build_vectorstore_improved(recreate_collection=True)
-
-            logger.info(f" Đã load {len(chunks)} chunks từ {CHUNKS_PICKLE}")
-            return db, chunks
-        except Exception as e:
-            logger.error(f" Không thể đọc {CHUNKS_PICKLE}: {e} — sẽ thử tái tạo từ file nguồn.")
-
-
-    # Nếu mất file pickle hoặc lỗi, fallback về tái tạo từ file nguồn 
-    docs = load_and_clean_all_docs()
-
-    chunks = smart_chunking(docs)
-    # Lưu lại chunks mới tái tạo vào file pickle để lần sau load nhanh hơn
+        initial_sync = await asyncio.wait_for(
+            asyncio.shield(startup_sync_task),
+            timeout=startup_wait_seconds,
+        )
+        return {
+            "task": startup_sync_task,
+            "initial_sync": initial_sync,
+            "timed_out": False,
+        }
+    except asyncio.TimeoutError:
+        return {
+            "task": startup_sync_task,
+            "initial_sync": None,
+            "timed_out": True,
+        }
+
+
+def load_vectorstore_improved(sync_coordinator: Any) -> Dict[str, Any]:
+    """Supabase load step: return current coordinator health snapshot."""
+    if sync_coordinator is None:
+        return {}
+
     try:
-        os.makedirs(VECTOR_DIR, exist_ok=True)
-        with open(CHUNKS_PICKLE, 'wb') as f:
-            pickle.dump(chunks, f)
-        logger.info(f" Đã tái tạo và lưu {len(chunks)} chunks vào {CHUNKS_PICKLE}")
-    except Exception as e:
-        logger.error(f" Không thể lưu chunks: {e}")
-
-    logger.info(f"Đã tái tạo {len(chunks)} chunks từ file nguồn")
-    return db, chunks
+        state = sync_coordinator.get_health_snapshot()
+        return state if isinstance(state, dict) else {}
+    except Exception:
+        logger.exception("Khong the lay sync state tu coordinator")
+        return {}

main.py

@@ -29,9 +29,9 @@ from core.config import (
 from core.document_db import init_document_db
 from core.supabase_sync_service import SupabaseStorageSyncService, SupabaseSyncCoordinator
 from core.collection_router_retriever import CollectionRouterRetriever
+from core.vectorstore import build_vectorstore_improved, load_vectorstore_improved
 from core.models import embeddings
 from core.qa_pipeline import ask_ai_improved, ask_ai_stream_delta
-from api.admin_documents_router import router as admin_documents_router
 from api.admin_sync_router import router as admin_sync_router
 
 # Hàm log lỗi an toàn
@@ -170,42 +170,48 @@ async def lifespan(app: FastAPI):
                 app.state.supabase_sync_service = sync_service
                 app.state.supabase_sync_coordinator = sync_coordinator
 
-                startup_sync_task = asyncio.create_task(
-                    sync_coordinator.run_sync(
-                    trigger="startup:initial_sync",
-                    queue_if_locked=False,
-                    )
+                build_result = await build_vectorstore_improved(
+                    sync_coordinator=sync_coordinator,
+                    startup_wait_seconds=SUPABASE_STARTUP_SYNC_WAIT_SECONDS,
                 )
+
+                startup_sync_task = build_result.get("task")
                 app.state.supabase_startup_sync_task = startup_sync_task
+                initial_sync = build_result.get("initial_sync")
+                timed_out = bool(build_result.get("timed_out"))
 
-                if SUPABASE_STARTUP_SYNC_WAIT_SECONDS > 0:
-                    try:
-                        initial_sync = await asyncio.wait_for(
-                            asyncio.shield(startup_sync_task),
-                            timeout=SUPABASE_STARTUP_SYNC_WAIT_SECONDS,
-                        )
-                        if initial_sync.get("status") == "failed":
-                            logger.warning(
-                                "Supabase initial sync failed at startup. service will continue and retry in scheduler. error=%s",
-                                initial_sync.get("error"),
-                            )
-                        else:
-                            summary = initial_sync.get("result") if isinstance(initial_sync.get("result"), dict) else {}
-                            logger.info(
-                                "Supabase initial sync completed. added=%s updated=%s deleted=%s failed=%s total_objects=%s",
-                                summary.get("added", 0),
-                                summary.get("updated", 0),
-                                summary.get("deleted", 0),
-                                summary.get("failed", 0),
-                                summary.get("total_objects", 0),
-                            )
-                    except asyncio.TimeoutError:
+                if timed_out:
+                    if SUPABASE_STARTUP_SYNC_WAIT_SECONDS > 0:
                         logger.warning(
                             "Supabase initial sync is still running after %ss. API startup continues and sync will finish in background.",
                             SUPABASE_STARTUP_SYNC_WAIT_SECONDS,
                         )
+                    else:
+                        logger.info("Supabase initial sync chạy nền, không chặn startup (SUPABASE_STARTUP_SYNC_WAIT_SECONDS=0).")
+                elif isinstance(initial_sync, dict) and initial_sync.get("status") == "failed":
+                    logger.warning(
+                        "Supabase initial sync failed at startup. service will continue and retry in scheduler. error=%s",
+                        initial_sync.get("error"),
+                    )
                 else:
-                    logger.info("Supabase initial sync chạy nền, không chặn startup (SUPABASE_STARTUP_SYNC_WAIT_SECONDS=0).")
+                    summary = initial_sync.get("result") if isinstance(initial_sync, dict) and isinstance(initial_sync.get("result"), dict) else {}
+                    logger.info(
+                        "Supabase initial sync completed. added=%s updated=%s deleted=%s failed=%s total_objects=%s",
+                        summary.get("added", 0),
+                        summary.get("updated", 0),
+                        summary.get("deleted", 0),
+                        summary.get("failed", 0),
+                        summary.get("total_objects", 0),
+                    )
+
+                sync_state = load_vectorstore_improved(sync_coordinator)
+                if sync_state:
+                    logger.info(
+                        "Supabase sync state loaded. running=%s queued_events=%s last_sync_at=%s",
+                        sync_state.get("running"),
+                        sync_state.get("queued_events"),
+                        sync_state.get("last_sync_at"),
+                    )
 
                 sync_stop_event = asyncio.Event()
                 sync_task = asyncio.create_task(
@@ -274,7 +280,6 @@ def get_runtime_components(request: Request):
 
 #Cấu hình FastAPI với middleware CORS và lifespan để quản lý trạng thái hệ thống
 app = FastAPI(lifespan=lifespan, title= "RAG API SERVER")
-app.include_router(admin_documents_router)
 app.include_router(admin_sync_router)
 
 #Cho phép truy cập từ mọi nguồn