upload document from admin

api/admin_documents_router.py

@@ -0,0 +1,148 @@
+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, List[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

@@ -26,6 +26,10 @@ FINAL_TOP_K = int(os.getenv('FINAL_TOP_K', '3'))
 
 DATA_DIR = os.getenv('DATA_DIR', 'data')
 VECTOR_DIR = os.getenv('VECTOR_DIR', 'vectorstore')
+UPLOAD_DIR = os.getenv('UPLOAD_DIR', 'uploads')
+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', 'sqlite:///./rag_metadata.db')
 
 # External service configs
 QDRANT_URL = os.getenv('QDRANT_URL')

core/document_db.py

@@ -0,0 +1,59 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, create_engine
+from sqlalchemy.orm import declarative_base, relationship, sessionmaker
+
+from .config import DOCUMENTS_DATABASE_URL
+
+Base = declarative_base()
+
+_connect_args = {"check_same_thread": False} if DOCUMENTS_DATABASE_URL.startswith("sqlite") else {}
+engine = create_engine(DOCUMENTS_DATABASE_URL, connect_args=_connect_args)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class Document(Base):
+    __tablename__ = "documents"
+
+    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    original_name = Column(String(512), nullable=False)
+    stored_name = Column(String(512), nullable=False)
+    path = Column(String(1024), nullable=False)
+    mime_type = Column(String(255), nullable=False)
+    size = Column(Integer, nullable=False)
+    status = Column(String(32), nullable=False, default="pending")
+    total_chunks = Column(Integer, nullable=False, default=0)
+    error_message = Column(Text, nullable=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=utcnow)
+
+    chunks = relationship("DocumentChunk", back_populates="document", cascade="all, delete-orphan")
+
+
+class DocumentChunk(Base):
+    __tablename__ = "document_chunks"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    document_id = Column(String(36), ForeignKey("documents.id", ondelete="CASCADE"), nullable=False)
+    chunk_index = Column(Integer, nullable=False)
+    content_preview = Column(String(200), nullable=False)
+    qdrant_point_id = Column(String(64), nullable=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=utcnow)
+
+    document = relationship("Document", back_populates="chunks")
+
+
+def init_document_db() -> None:
+    Base.metadata.create_all(bind=engine)
+
+
+def get_document_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

core/document_ingest_service.py

@@ -0,0 +1,181 @@
+import logging
+import os
+import re
+import uuid
+from datetime import datetime, timezone
+from typing import List
+
+from docx import Document as DocxDocument
+from fastapi.concurrency import run_in_threadpool
+from pypdf import PdfReader
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, PointStruct, VectorParams
+
+from .config import CHUNK_OVERLAP, CHUNK_SIZE, QDRANT_API_KEY, QDRANT_COLLECTION, QDRANT_URL
+from .document_db import Document, DocumentChunk, SessionLocal
+from .models import embeddings
+
+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:
+    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)
+
+    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))
+
+        return "\n".join(paragraphs)
+
+    with open(path, "r", encoding="utf-8", errors="ignore") as file:
+        return file.read()
+
+
+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")
+
+    tokens = _TOKEN_RE.findall(text)
+    if not tokens:
+        return []
+
+    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
+
+    return chunks
+
+
+def _ensure_qdrant_collection(client: QdrantClient, vector_size: int) -> None:
+    if not client.collection_exists(collection_name=QDRANT_COLLECTION):
+        client.create_collection(
+            collection_name=QDRANT_COLLECTION,
+            vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE),
+        )
+
+
+def process_document_ingest(document_id: str) -> None:
+    db = SessionLocal()
+    document = db.query(Document).filter(Document.id == document_id).first()
+
+    if document is None:
+        db.close()
+        logger.error("Document not found for ingest: %s", document_id)
+        return
+
+    try:
+        document.status = "processing"
+        document.error_message = None
+        db.commit()
+
+        _, extension = os.path.splitext(document.stored_name)
+        raw_text = read_document_content(document.path, extension)
+        normalized = normalize_text(raw_text)
+        chunks = chunk_text_by_tokens(normalized, CHUNK_SIZE, CHUNK_OVERLAP)
+
+        if not chunks:
+            raise ValueError("Document has no readable content after normalization.")
+
+        if not QDRANT_URL:
+            raise ValueError("QDRANT_URL is required for ingest.")
+
+        client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY or None)
+        vectors = embeddings.embed_documents(chunks)
+
+        if not vectors or not vectors[0]:
+            raise ValueError("Failed to create embeddings for chunks.")
+
+        _ensure_qdrant_collection(client, len(vectors[0]))
+
+        created_at = datetime.now(timezone.utc).isoformat()
+        points: List[PointStruct] = []
+        db_chunk_rows: List[DocumentChunk] = []
+
+        for index, (chunk_text, vector) in enumerate(zip(chunks, vectors)):
+            point_id = str(uuid.uuid4())
+            payload = {
+                "document_id": document.id,
+                "filename": document.original_name,
+                "stored_name": document.stored_name,
+                "path": document.path,
+                "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,
+                )
+            )
+
+        client.upsert(collection_name=QDRANT_COLLECTION, points=points, wait=True)
+
+        db.query(DocumentChunk).filter(DocumentChunk.document_id == document.id).delete()
+        db.bulk_save_objects(db_chunk_rows)
+
+        document.total_chunks = len(chunks)
+        document.status = "done"
+        db.commit()
+
+        logger.info("Document ingest success. document_id=%s total_chunks=%s", document.id, len(chunks))
+    except Exception as error:
+        db.rollback()
+
+        failed_doc = db.query(Document).filter(Document.id == document_id).first()
+        if failed_doc is not None:
+            failed_doc.status = "failed"
+            failed_doc.error_message = str(error)
+            db.commit()
+
+        logger.exception("Document ingest failed. document_id=%s", document_id)
+    finally:
+        db.close()
+
+
+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

@@ -8,7 +8,7 @@ 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
+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
@@ -25,7 +25,7 @@ logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 CHUNKS_PICKLE = os.path.join(VECTOR_DIR, "chunks.pkl")
-COLLECTION_NAME = "quy_che_db"
+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})")

main.py

@@ -11,10 +11,12 @@ import asyncpg
 from starlette.concurrency import iterate_in_threadpool
 from qdrant_client import QdrantClient
 #Import các model và các hàm cần thiết từ core 
-from core.config import QDRANT_URL, QDRANT_API_KEY, DATABASE_URL
+from core.config import QDRANT_URL, QDRANT_API_KEY, DATABASE_URL, QDRANT_COLLECTION
+from core.document_db import init_document_db
 from core.vectorstore import build_vectorstore_improved, load_vectorstore_improved
 from core.retriever import HybridRetriever
 from core.qa_pipeline import ask_ai_improved, ask_ai_stream_delta
+from api.admin_documents_router import router as admin_documents_router
 
 # Hàm log lỗi an toàn
 logging.basicConfig(level=logging.INFO)
@@ -109,6 +111,8 @@ async def lifespan(app: FastAPI):
     logger.info("Đang khởi tạo API SERVER ...")
     pool = None
     try:
+        init_document_db()
+
         pool = await asyncpg.create_pool(
             dsn=DATABASE_URL,
             min_size=POOL_MIN_SIZE,
@@ -118,7 +122,7 @@ async def lifespan(app: FastAPI):
         await init_db_asyncpg(pool)
 
         client = QdrantClient(url = QDRANT_URL, api_key=QDRANT_API_KEY)
-        collection_name= "quy_che_db"
+        collection_name = QDRANT_COLLECTION
         if not client.collection_exists(collection_name):
             logger.warning(f"Chưa có collection {collection_name} trên Qdrant Cloud. Đang xây dựng vectorstore mới...")
             db, all_chunks= build_vectorstore_improved()
@@ -151,6 +155,7 @@ 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)
 
 #Cho phép truy cập từ mọi nguồn 
 allow_origins = [origin.strip() for origin in os.getenv("ALLOW_ORIGINS", "*").split(",") if origin.strip()]

requirements.txt

@@ -12,6 +12,7 @@ google-generativeai>=0.7.0
 
 # Database & Vector Store
 asyncpg>=0.29.0
+sqlalchemy>=2.0.0
 qdrant-client>=1.9.0
 
 #Embedding Models & Transformers