api.py

import os

from fastapi import FastAPI, Query
from typing import Union
from fastapi.responses import JSONResponse, FileResponse
from typing import Annotated
from core.concept_extractor import extract
from pydantic import BaseModel, Field
import logging
from logging.handlers import RotatingFileHandler

############# LOGGER ##################
# Create a custom logger
logger = logging.getLogger("app")
logger.setLevel(logging.INFO)  # Capture all logs

# Console handler - prints everything
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_formatter = logging.Formatter("%(asctime)s - [%(levelname)s] - %(name)s - %(message)s")
console_handler.setFormatter(console_formatter)

# File handler - saves only ERROR and above
file_handler = RotatingFileHandler("logs.log", maxBytes=10000000, backupCount=5)
file_handler.setLevel(logging.INFO)
file_formatter = logging.Formatter("%(asctime)s - [%(levelname)s] - %(name)s - %(message)s")
file_handler.setFormatter(file_formatter)

# Add handlers to the logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)

class QueryParams(BaseModel):
    text: str = Field(description="Text to extract concepts from")
    filter_tags: Union[list[str],None] = Field(default=None, description="Text to extract concepts from")
    exclude: bool = Field(default=False, description="Use filter tags as an exclusion list?")
    fuzzy_threshold: int = Field(default=100, description="Fuzzy matching threshold (0-100)")
    use_premium: bool = Field(default=False, description="Use premium translation?")

app = FastAPI(title="Concept Extractor",
              contact={
                  "name": "Concept Extractor",
                  "url": "https://github.com/detsutut",
                  "email": "buonocore.tms@gmail.com",
              }
              )

@app.get("/")
def read_root():
    return {}

@app.get('/favicon.ico', include_in_schema=False)
async def favicon():
    return FileResponse("favicon.ico")

@app.get("/extract", response_model=dict)
def extract_from_text(text: str,
                      f: Annotated[list[str] | None, Query(description="List of tags to filter by")] = None,
                      e: bool = Query(default=False, description="Use filter tags as an exclusion list?"),
                      o: int = Query(default=100, description="Fuzzy matching threshold (0-100)"),
                      p: bool = Query(default=False, description="Use premium translation?")):
    if not text:
        return JSONResponse(content={"error": "Please provide a text to extract from."}, status_code=400)
    logger.info(f"Received extraction request:\n\tText:{text}\n\tfilter_tags:{f}, exclude:{e}, fuzzy_threshold:{o}, use_premium:{p}\n")
    results_dataframe = extract(text=text, filter_tags=f or [], exclude=e, fuzzy_threshold=o, use_premium=p)
    if results_dataframe is None:
        return {}
    results_dict = results_dataframe.to_dict()
    logger.info(f"Extraction completed:{results_dict}\n")
    return results_dict

@app.post("/extract", response_model=dict)
def extract_from_text(query: QueryParams):
    if not query.text:
        return JSONResponse(content={"error": "Please provide a text to extract from."}, status_code=400)
    logger.info(f"Received extraction request:\n\tText:{query.text}\n\tfilter_tags:{query.filter_tags}, exclude:{query.exclude}, fuzzy_threshold:{query.fuzzy_threshold}, use_premium:{query.use_premium}\n")
    results_dataframe = extract(text=query.text,
                                filter_tags=query.filter_tags or [],
                                exclude=query.exclude,
                                fuzzy_threshold=query.fuzzy_threshold,
                                use_premium=query.use_premium)
    if results_dataframe is None:
        return {}
    results_dict = results_dataframe.to_dict()
    logger.info(f"Extraction completed:{results_dict}\n")
    return results_dict