dependencies.py

import os
import sys
import json
from dotenv import load_dotenv
import pika

load_dotenv()

from utils.db_interfaces import PostgresInterface
from utils.file_manager import FileManager
from utils.file_manager_metadata import FileManagerMetadata
from utils.tags_generation import TagsGeneration
from utils.ai_adapters import AIAdapter
from utils.pic_manager import PicManager
from utils.categories_manager import CategoriesManager


dependent_status = {
    "postgres-services": {
        "uri": f'{os.getenv("POSTGRES_HOST")}:{os.getenv("POSTGRES_PORT")}',
        "status": True,
        "message": "running..."
    },
    "milvus-services": {
        "uri": os.getenv("MILVUS_URI"),
        "status": True,
        "message": "running..."
    },
    "rabbitmq-services": {
        "uri": f'{os.getenv("RABBITMQ_URI")}',
        "status": True,
        "message": "running..."
    },
    "ai-engine": {
        "engine": os.getenv("AI_ENGINE"),
        "status": True,
        "message": "running..."
    },
    "vector-engine": {
        "model": os.getenv("VECTOR_MODEL"),
        "device": "cpu",
        "status": True,
        "message": "running..."
    },
    "pdf-engine": {
        "loader": os.getenv("PDF_LOADER_ENGINE"),
        "splitter": os.getenv("PDF_SPLITTER_ENGINE"),
        "chunk_size": int(os.getenv("CHUNK_SIZE")),
        "chunk_overlap": int(os.getenv("CHUNK_OVERLAP")),
        "status": True,
        "message": "running..."
    }
}

# -------------------------------------------------------------------
# Postgres Interface
# -------------------------------------------------------------------
try:
    # Define connection to postgresql database
    postgres_interface = PostgresInterface(
            host=os.getenv("POSTGRES_HOST"),
            port=os.getenv("POSTGRES_PORT"),
            user=os.getenv("POSTGRES_USER"),
            password=os.getenv("POSTGRES_PASSWORD"),
            database=os.getenv("POSTGRES_DB")
        )
    # Define postgresql management functions 
    file_manager_interface = FileManager(postgres_interface)
    file_manager_md_interface = FileManagerMetadata(postgres_interface)
    tags_generator_interface = TagsGeneration(postgres_interface)
    pic_manager_interface = PicManager(postgres_interface)
    categories_manager_interface = CategoriesManager(postgres_interface)
    
    # Define get postgresql management functions
    def get_file_manager():
        return file_manager_interface
    def get_file_manager_md():
        return file_manager_md_interface
    def get_tags_generator():
        return tags_generator_interface
    def get_pic_manager():
        return pic_manager_interface
    def get_categories_manager():
        return categories_manager_interface
except Exception as e:
    dependent_status['postgres-services']['status'] = False
    dependent_status['postgres-services']['message'] = str(e)

# -------------------------------------------------------------------
# Milvus Interface
# -------------------------------------------------------------------

from utils.db_interfaces import MilvusInterface
from utils.docs_retrieval import DocsRetrieval

try:
    milvus_interface = MilvusInterface(
        uri=os.getenv("MILVUS_URI"),
        token=os.getenv("MILVUS_TOKEN"),
        database=os.getenv("MILVUS_DB")
    )

    docs_retriever = DocsRetrieval(milvus_interface)

    def get_docs_retriever():
        return docs_retriever
except Exception as e:
    dependent_status['milvus-services']['status'] = False
    dependent_status['milvus-services']['message'] = str(e)

# -------------------------------------------------------------------
# Document Loader and Processor - PDF
# -------------------------------------------------------------------
from utils.docs_loader import FMPDFLoader
from utils.docs_processor import DocsProcessor

try:
    pdf_loader = FMPDFLoader(loader_type=os.getenv("PDF_LOADER_ENGINE"))
    pdf_processor = DocsProcessor(
        splitter_engine=os.getenv("PDF_SPLITTER_ENGINE"),
        chunk_size=int(os.getenv("CHUNK_SIZE")),
        chunk_overlap=int(os.getenv("CHUNK_OVERLAP"))
    )
    
    def get_pdf_loader():
        return pdf_loader
    def get_pdf_processor():
        return pdf_processor
except Exception as e:
    dependent_status['pdf-engine']['loader'] = os.getenv("PDF_LOADER_ENGINE")
    dependent_status['pdf-engine']['splitter'] = os.getenv("PDF_SPLITTER_ENGINE")
    dependent_status['pdf-engine']['status'] = False
    dependent_status['pdf-engine']['message'] = str(e)

# -------------------------------------------------------------------
# AI Adapter
# -------------------------------------------------------------------
def select_api_key(ai_engine: str):
    if ai_engine == "gemini":
        return os.getenv("GEMINI_KEY")
    elif ai_engine == "groq":
        return os.getenv("GROQ_KEY")
    elif ai_engine == "deepseek":
        return os.getenv("DEEPSEEK_KEY")
    elif ai_engine == "openai":
        return os.getenv("OPENAI_KEY")
    elif ai_engine == "anthropic":
        return os.getenv("CLAUDE_KEY")
    elif ai_engine == "perplexity":
        return os.getenv("PERPLEXITY_KEY")
    else:
        raise ValueError(f"Unsupported engine: {ai_engine}")

try:
    ai_adapter = AIAdapter(
        ai_engine=os.getenv("AI_ENGINE"),
        api_key=select_api_key(os.getenv("AI_ENGINE")),
        tags_prompt_version=os.getenv("TAGS_PROMPT_VERSION")
    )

    def get_ai_adapter():
        return ai_adapter
except Exception as e:
    dependent_status['ai-engine']['status'] = False
    dependent_status['ai-engine']['message'] = str(e)

# -------------------------------------------------------------------
# RabbitMQ Services
# -------------------------------------------------------------------
import pika

params = pika.URLParameters(os.getenv("RABBITMQ_URI"))
try:
    connection = pika.BlockingConnection(params)
    connection.close()
except Exception as e:
    dependent_status['rabbitmq-services']['status'] = False
    dependent_status['rabbitmq-services']['message'] = str(e)

# -------------------------------------------------------------------
# Vector Retrieval and Search
# -------------------------------------------------------------------
import torch
from utils.vector_adapters import VectorModelManager

device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
model_path = os.getenv("VECTOR_MODEL")

try:
    vector_model_manager = VectorModelManager(model_path, device)
    dependent_status['vector-engine']['device'] = device

    def get_vector_model():
        return vector_model_manager
except Exception as e:
    dependent_status['vector-engine']['status'] = False
    dependent_status['vector-engine']['message'] = str(e)

# -------------------------------------------------------------------
# Dependencies Checker Running
# -------------------------------------------------------------------
if (
    dependent_status['postgres-services']['status'] == True and
    dependent_status['milvus-services']['status'] == True and
    dependent_status['ai-engine']['status'] == True and
    dependent_status['vector-engine']['status'] == True and
    dependent_status['rabbitmq-services']['status'] == True
):
    print("--------------------------------------")
    print("File Manager Services Status:")
    print("--------------------------------------")
    print(json.dumps(dependent_status, indent=4))
    print("--------------------------------------")
    print("Attention: ✅  All dependencies services status ok!")
    print("--------------------------------------")
else:
    print("--------------------------------------")
    print("File Manager Service Status:")
    print("--------------------------------------")
    print(json.dumps(dependent_status, indent=4))
    print("--------------------------------------")
    print("Attention: ⚠️  Please check dependencies log above!")
    print("--------------------------------------")
    sys.exit()