From fcba05ac4b9b86a2434c3b7b89ae170127dfa1d4 Mon Sep 17 00:00:00 2001 From: Aryan Rangarajan Date: Fri, 10 Jul 2026 22:53:54 +0530 Subject: [PATCH] Implement watchdog client and automatic file indexing --- .dockerignore | 1 - .gitignore | 7 +- .../ecca7b879961_genesis_migration.py | 1 + .../efb17bbdead1_add_watched_folders_table.py | 39 ++++ app/main.py | 2 + app/models/__init__.py | 3 +- app/models/watched_folder.py | 27 +++ app/routes/files.py | 74 +++++++- app/routes/upload.py | 8 +- app/routes/watch.py | 95 ++++++++++ app/services/archive.py | 21 +++ app/services/indexer.py | 176 ++++++++++++++---- docker-compose.yml | 4 +- watchdog_client/__init__.py | 0 watchdog_client/api.py | 81 ++++++++ watchdog_client/config.py | 2 + watchdog_client/hashing.py | 11 ++ watchdog_client/main.py | 32 ++++ watchdog_client/scanner.py | 36 ++++ watchdog_client/watcher.py | 84 +++++++++ 20 files changed, 664 insertions(+), 40 deletions(-) create mode 100644 alembic/versions/efb17bbdead1_add_watched_folders_table.py create mode 100644 app/models/watched_folder.py create mode 100644 app/routes/watch.py create mode 100644 app/services/archive.py create mode 100644 watchdog_client/__init__.py create mode 100644 watchdog_client/api.py create mode 100644 watchdog_client/config.py create mode 100644 watchdog_client/hashing.py create mode 100644 watchdog_client/main.py create mode 100644 watchdog_client/scanner.py create mode 100644 watchdog_client/watcher.py diff --git a/.dockerignore b/.dockerignore index dda73b3..6add21f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -2,4 +2,3 @@ __pycache__ .git .env -alembic/versions \ No newline at end of file diff --git a/.gitignore b/.gitignore index 4898d55..077eb3a 100644 --- a/.gitignore +++ b/.gitignore @@ -52,4 +52,9 @@ docker-compose.override.yml # ====================== .DS_Store Thumbs.db -test_data/ \ No newline at end of file +test_data/ + +# ====================== +#Runtime files +# ====================== +temp/ \ No newline at end of file diff --git a/alembic/versions/ecca7b879961_genesis_migration.py b/alembic/versions/ecca7b879961_genesis_migration.py index 689ceb2..e34943b 100644 --- a/alembic/versions/ecca7b879961_genesis_migration.py +++ b/alembic/versions/ecca7b879961_genesis_migration.py @@ -18,6 +18,7 @@ branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None +op.execute("CREATE EXTENSION IF NOT EXISTS vector") def upgrade() -> None: """Upgrade schema.""" diff --git a/alembic/versions/efb17bbdead1_add_watched_folders_table.py b/alembic/versions/efb17bbdead1_add_watched_folders_table.py new file mode 100644 index 0000000..e2acdfd --- /dev/null +++ b/alembic/versions/efb17bbdead1_add_watched_folders_table.py @@ -0,0 +1,39 @@ +"""add watched folders table + +Revision ID: efb17bbdead1 +Revises: ecca7b879961 +Create Date: 2026-07-02 12:29:07.453901 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import pgvector.sqlalchemy + + +# revision identifiers, used by Alembic. +revision: str = 'efb17bbdead1' +down_revision: Union[str, Sequence[str], None] = 'ecca7b879961' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('watched_folders', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('folder_path', sa.Text(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('folder_path') + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table('watched_folders') + # ### end Alembic commands ### diff --git a/app/main.py b/app/main.py index fbd693a..1b49429 100644 --- a/app/main.py +++ b/app/main.py @@ -2,6 +2,7 @@ from app.routes import search from app.routes import upload from app.routes import files +from app.routes import watch app = FastAPI( title="WhereTF Backend" @@ -10,6 +11,7 @@ app.include_router(search.router) app.include_router(upload.router) app.include_router(files.router) +app.include_router(watch.router) @app.get("/health", tags=["System"]) def health_check(): diff --git a/app/models/__init__.py b/app/models/__init__.py index bf82e82..f00cad8 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -1,3 +1,4 @@ from app.models.file import File from app.models.content import FileContent -from app.models.relationship import FileRelationship \ No newline at end of file +from app.models.relationship import FileRelationship +from app.models.watched_folder import WatchedFolder diff --git a/app/models/watched_folder.py b/app/models/watched_folder.py new file mode 100644 index 0000000..1b3bc6c --- /dev/null +++ b/app/models/watched_folder.py @@ -0,0 +1,27 @@ +from datetime import datetime + +from sqlalchemy import DateTime, Text +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.sql import func + +from app.database import Base + + +class WatchedFolder(Base): + __tablename__ = "watched_folders" + + id: Mapped[int] = mapped_column( + primary_key=True, + autoincrement=True + ) + + folder_path: Mapped[str] = mapped_column( + Text, + unique=True, + nullable=False + ) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now() + ) \ No newline at end of file diff --git a/app/routes/files.py b/app/routes/files.py index 6f98cb1..b2d2215 100644 --- a/app/routes/files.py +++ b/app/routes/files.py @@ -66,4 +66,76 @@ def delete_file(file_id: uuid.UUID, db: Session = Depends(get_db)): db.delete(db_file) db.commit() - return {"status": "success", "message": f"File and vectors wiped successfully."} \ No newline at end of file + return {"status": "success", "message": f"File and vectors wiped successfully."} + + +class FilePathRequest(BaseModel): + file_path: str + + +@router.post("/delete-by-path") +def delete_file_by_path(req: FilePathRequest, db: Session = Depends(get_db)): + + db_file = db.scalar( + select(File).where(File.file_path == req.file_path) + ) + + if not db_file: + raise HTTPException(status_code=404, detail="File not found") + + db.delete(db_file) + db.commit() + + return { + "status": "success", + "message": "File deleted successfully." + } + +class FileRenameRequest(BaseModel): + old_path: str + new_path: str + +@router.post("/rename") +def rename_file(req: FileRenameRequest, db: Session = Depends(get_db)): + + db_file = db.scalar( + select(File).where(File.file_path == req.old_path) + ) + + if not db_file: + raise HTTPException(status_code=404, detail="File not found") + + db_file.file_path = req.new_path + + db.commit() + + return { + "status": "success", + "message": "File renamed successfully." + } + +class FileCheckRequest(BaseModel): + file_path: str + file_hash: str + +@router.post("/needs-indexing") +def needs_indexing(req: FileCheckRequest, db: Session = Depends(get_db)): + db_file = db.scalar( + select(File).where( + File.file_path == req.file_path + ) + ) + + if db_file is None: + return { + "needs_indexing": True + } + + if db_file.file_hash == req.file_hash: + return { + "needs_indexing": False + } + + return { + "needs_indexing": True + } \ No newline at end of file diff --git a/app/routes/upload.py b/app/routes/upload.py index 7d7fad8..17e6de2 100644 --- a/app/routes/upload.py +++ b/app/routes/upload.py @@ -1,7 +1,7 @@ import os import shutil -from fastapi import APIRouter, UploadFile, File, BackgroundTasks, HTTPException +from fastapi import APIRouter, Form, UploadFile, File, BackgroundTasks, HTTPException from app.services.indexer import background_index_file @@ -11,7 +11,8 @@ @router.post("/upload/") async def upload_file( background_tasks: BackgroundTasks, - file: UploadFile = File(...) + file: UploadFile = File(...), + original_path: str = Form(...) ): """ Accept a document and send it for background indexing. @@ -27,7 +28,8 @@ async def upload_file( background_tasks.add_task( background_index_file, - temp_path + temp_path, + original_path ) return { diff --git a/app/routes/watch.py b/app/routes/watch.py new file mode 100644 index 0000000..ae4a2fd --- /dev/null +++ b/app/routes/watch.py @@ -0,0 +1,95 @@ +from pathlib import Path + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from app.database import SessionLocal +from app.models import WatchedFolder + +router = APIRouter(tags=["Watch"]) + + +class WatchFolderRequest(BaseModel): + folder_path: str + + +@router.post("/watch/folder") +def add_watch_folder(req: WatchFolderRequest): + + db = SessionLocal() + + try: + existing = db.query(WatchedFolder).filter_by( + folder_path=req.folder_path + ).first() + + if existing: + raise HTTPException( + status_code=409, + detail="Folder is already being watched." + ) + + watched_folder = WatchedFolder( + folder_path=req.folder_path + ) + + db.add(watched_folder) + db.commit() + db.refresh(watched_folder) + + return { + "success": True, + "id": watched_folder.id, + "folder_path": watched_folder.folder_path + } + + finally: + db.close() + + +@router.get("/watch/folders") +def get_watched_folders(): + + db = SessionLocal() + + try: + folders = db.query(WatchedFolder).all() + + return [ + { + "id": folder.id, + "folder_path": folder.folder_path, + "created_at": folder.created_at + } + for folder in folders + ] + + finally: + db.close() + + +@router.delete("/watch/folder") +def remove_watch_folder(req: WatchFolderRequest): + + db = SessionLocal() + + try: + folder = db.query(WatchedFolder).filter_by( + folder_path=req.folder_path + ).first() + + if not folder: + raise HTTPException( + status_code=404, + detail="Folder not found." + ) + + db.delete(folder) + db.commit() + + return { + "success": True + } + + finally: + db.close() \ No newline at end of file diff --git a/app/services/archive.py b/app/services/archive.py new file mode 100644 index 0000000..266ca7a --- /dev/null +++ b/app/services/archive.py @@ -0,0 +1,21 @@ +import os +import tempfile +import zipfile + + +def extract_zip(zip_path: str): + + temp_dir = tempfile.mkdtemp() + + with zipfile.ZipFile(zip_path, "r") as archive: + archive.extractall(temp_dir) + + extracted_files = [] + + for root, _, files in os.walk(temp_dir): + for file in files: + extracted_files.append( + os.path.join(root, file) + ) + + return temp_dir, extracted_files \ No newline at end of file diff --git a/app/services/indexer.py b/app/services/indexer.py index 8e24576..4757b3e 100644 --- a/app/services/indexer.py +++ b/app/services/indexer.py @@ -1,56 +1,168 @@ import os +import shutil +import zipfile import logging + from app.database import SessionLocal -from app.models import File, FileContent +from app.models import File, FileContent from app.processing import process +from app.services.archive import extract_zip logger = logging.getLogger(__name__) -def background_index_file(temp_file_path: str): + +def save_processed_data(db, file_rows, content_rows): """ - Background task that processes a document, extracts AI embeddings, - saves them to pgvector, and cleans up the temporary file. + Save processed file metadata and chunks into the database. """ + + for fr in file_rows: + existing = db.query(File).filter_by(file_path=fr["file_path"]).first() + + if existing: + for k, v in fr.items(): + setattr(existing, k, v) + else: + db.add(File(**fr)) + + db.flush() + + for cr in content_rows: + + file_path = cr.pop("file_path") + + file_obj = ( + db.query(File) + .filter_by(file_path=file_path) + .one() + ) + + db.add( + FileContent( + file_id=file_obj.id, + chunk_index=cr["chunk_index"], + content_text=cr["content_text"], + embedding=cr["embedding"], + ) + ) + + +def background_index_file(temp_file_path: str, original_path: str): + """ + Processes a document (or ZIP archive), generates embeddings, + stores everything in PostgreSQL, and cleans up temporary files. + """ + db = SessionLocal() - + try: - logger.info(f"Starting background indexing for: {temp_file_path}") - + + logger.info(f"Starting indexing for: {temp_file_path}") + + # -------------------------------------------------------- + # ZIP FILE + # -------------------------------------------------------- + + if zipfile.is_zipfile(temp_file_path): + + logger.info("ZIP archive detected.") + + temp_extract_dir, extracted_files = extract_zip( + temp_file_path + ) + + try: + + for extracted in extracted_files: + + extension = os.path.splitext(extracted)[1].lower() + + supported = { + ".pdf", + ".docx", + ".txt", + ".md", + ".pptx", + ".csv", + ".py", + } + + if extension not in supported: + logger.info(f"Skipping unsupported file: {extracted}") + continue + + logger.info(f"Processing: {extracted}") + + archive_path = ( + f"{original_path}::" + f"{os.path.relpath(extracted, temp_extract_dir)}" + ) + + file_rows, content_rows = process(extracted) + + for fr in file_rows: + fr["file_path"] = archive_path + + for cr in content_rows: + cr["file_path"] = archive_path + + save_processed_data( + db, + file_rows, + content_rows + ) + + db.commit() + + logger.info("ZIP archive indexed successfully.") + + finally: + + shutil.rmtree(temp_extract_dir) + + return + + # -------------------------------------------------------- + # NORMAL FILE + # -------------------------------------------------------- + file_rows, content_rows = process(temp_file_path) + for fr in file_rows: - existing = db.query(File).filter_by(file_path=fr["file_path"]).first() - if existing: - for k, v in fr.items(): - setattr(existing, k, v) - else: - db.add(File(**fr)) - - db.flush() + fr["file_path"] = original_path for cr in content_rows: - file_path = cr.pop("file_path") - file_obj = db.query(File).filter_by(file_path=file_path).one() - - db.add(FileContent( - file_id=file_obj.id, - chunk_index=cr["chunk_index"], - content_text=cr["content_text"], - embedding=cr["embedding"] - )) + cr["file_path"] = original_path + + save_processed_data( + db, + file_rows, + content_rows + ) - # permanently saving the vectors db.commit() - logger.info(f"Successfully indexed {len(file_rows)} file(s) and {len(content_rows)} chunks.") + + logger.info( + f"Successfully indexed " + f"{len(file_rows)} file(s) " + f"and {len(content_rows)} chunk(s)." + ) except Exception as e: - # undo the changes if anything fails + db.rollback() - logger.error(f"Failed to index file {temp_file_path}: {str(e)}") - + + logger.error( + f"Failed to index {temp_file_path}: {e}" + ) + finally: + db.close() - - # Server Cleanup,deleting the temp files + if os.path.exists(temp_file_path): os.remove(temp_file_path) - logger.info(f"Cleaned up temporary file: {temp_file_path}") \ No newline at end of file + + logger.info( + f"Removed temporary file: {temp_file_path}" + ) \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 5b0bee0..9894dbf 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -26,6 +26,8 @@ services: condition: service_healthy environment: DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/wheretf + volumes: + - .:/app -volumes: +volumes: postgres_data: diff --git a/watchdog_client/__init__.py b/watchdog_client/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/watchdog_client/api.py b/watchdog_client/api.py new file mode 100644 index 0000000..a23241a --- /dev/null +++ b/watchdog_client/api.py @@ -0,0 +1,81 @@ +from pathlib import Path +import requests + +from .config import API_URL + +def get_watched_folders(): + + response = requests.get( + f"{API_URL}/watch/folders" + ) + + response.raise_for_status() + + return response.json() + +def upload(file_path: str): + + with open(file_path, "rb") as f: + + response = requests.post( + f"{API_URL}/upload/", + files={ + "file": (Path(file_path).name, f) + }, + data={ + "original_path": file_path + } + ) + + response.raise_for_status() + + return response.json() + +def delete(file_path: str): + + response = requests.post( + f"{API_URL}/files/delete-by-path", + json={ + "file_path": file_path + } + ) + + response.raise_for_status() + + return response.json() + +def modify(file_path: str): + + # Remove the old indexed version + delete(file_path) + + # Upload and re-index the current version + return upload(file_path) + +def rename(old_path: str, new_path: str): + + response = requests.post( + f"{API_URL}/files/rename", + json={ + "old_path": old_path, + "new_path": new_path + } + ) + + response.raise_for_status() + + return response.json() + +def needs_indexing(file_path: str, file_hash: str): + + response = requests.post( + f"{API_URL}/files/needs-indexing", + json={ + "file_path": file_path, + "file_hash": file_hash + } + ) + + response.raise_for_status() + + return response.json()["needs_indexing"] \ No newline at end of file diff --git a/watchdog_client/config.py b/watchdog_client/config.py new file mode 100644 index 0000000..92fa5c6 --- /dev/null +++ b/watchdog_client/config.py @@ -0,0 +1,2 @@ +API_URL = "http://localhost:8000" +POLL_INTERVAL = 10 diff --git a/watchdog_client/hashing.py b/watchdog_client/hashing.py new file mode 100644 index 0000000..db95d12 --- /dev/null +++ b/watchdog_client/hashing.py @@ -0,0 +1,11 @@ +import hashlib + + +def calculate_file_hash(file_path: str) -> str: + sha256 = hashlib.sha256() + + with open(file_path, "rb") as f: + while chunk := f.read(8192): + sha256.update(chunk) + + return sha256.hexdigest() \ No newline at end of file diff --git a/watchdog_client/main.py b/watchdog_client/main.py new file mode 100644 index 0000000..b2654a0 --- /dev/null +++ b/watchdog_client/main.py @@ -0,0 +1,32 @@ +import time +from .scanner import scan_folder +from watchdog.observers import Observer +from .api import get_watched_folders +from .watcher import FileWatcher + +folders = get_watched_folders() +observer = Observer() + +for folder in folders: + + folder_path = folder["folder_path"] + + scan_folder(folder_path) + + observer.schedule( + FileWatcher(), + folder_path, + recursive=True + ) +observer.start() + +print("Watching folders...") + +try: + while True: + time.sleep(1) + +except KeyboardInterrupt: + observer.stop() + +observer.join() \ No newline at end of file diff --git a/watchdog_client/scanner.py b/watchdog_client/scanner.py new file mode 100644 index 0000000..bf14909 --- /dev/null +++ b/watchdog_client/scanner.py @@ -0,0 +1,36 @@ +import os + +from .api import upload, needs_indexing +from .hashing import calculate_file_hash + + +def scan_folder(folder_path: str): + """ + Scan an existing watched folder and upload only new/changed files. + """ + + for root, _, files in os.walk(folder_path): + + for filename in files: + + file_path = os.path.join(root, filename) + + try: + file_hash = calculate_file_hash(file_path) + + if needs_indexing(file_path, file_hash): + + print(f"Indexing: {file_path}") + + response = upload(file_path) + + print(response) + + else: + + print(f"Skipping: {file_path}") + + except Exception as e: + + print(f"Failed: {file_path}") + print(e) \ No newline at end of file diff --git a/watchdog_client/watcher.py b/watchdog_client/watcher.py new file mode 100644 index 0000000..3e9584e --- /dev/null +++ b/watchdog_client/watcher.py @@ -0,0 +1,84 @@ +from watchdog.events import FileSystemEventHandler +from .api import upload, delete, modify, rename +import time + +DEBOUNCE_SECONDS = 1.5 # Time to wait before processing a file event +last_processed={} + +class FileWatcher(FileSystemEventHandler): + + def on_created(self, event): + + if event.is_directory: + return + + now = time.time() + + last_time = last_processed.get(event.src_path) + + if last_time and (now - last_time) < DEBOUNCE_SECONDS: + return + + last_processed[event.src_path] = now + + print("Created:", event.src_path) + + try: + time.sleep(1) + response = upload(event.src_path) + print(response) + + except Exception as e: + print("Upload failed:", e) + + + def on_modified(self, event): + + if event.is_directory: + return + + now = time.time() + + last_time = last_processed.get(event.src_path) + + if last_time and (now - last_time) < DEBOUNCE_SECONDS: + return + + last_processed[event.src_path] = now + + print("Modified:", event.src_path) + + try: + modify(event.src_path) + print("Successfully re-indexed.") + + except Exception as e: + print("Modify failed:", e) + + def on_deleted(self, event): + + if event.is_directory: + return + + print("Deleted:", event.src_path) + + try: + response = delete(event.src_path) + print(response) + + except Exception as e: + print("Delete failed:", e) + + def on_moved(self, event): + + if event.is_directory: + return + + print(f"Renamed: {event.src_path} -> {event.dest_path}") + + try: + response = rename(event.src_path, event.dest_path) + print(response) + + except Exception as e: + print("Rename failed:", e) \ No newline at end of file