FastAPI/app/api.py

from typing import List
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
import numpy as np
from pydantic import BaseModel, Field, conlist

from app.utils.embedding import get_embedding


app = FastAPI()


origins = [
    "http://localhost:3000",
    "http://localhost:8000",
    "localhost:8000",
    "https://your-space-name.hf.space",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Mount static files
app.mount(
    "/static", StaticFiles(directory="app/build/static", html=True), name="static"
)


# Serve index.html at the root
@app.get("/")
def read_root():
    return FileResponse("app/build/index.html")


words_db = []


@app.get("/api/words", tags=["words"])
async def get_words() -> dict:

    return {"data": words_db}


@app.post("/api/add-word", tags=["words"])
async def add_word(word: dict) -> dict:
    try:
        word_embedding = get_embedding(word["item"])
        words_db.append(word)
        word["embedding"] = word_embedding.tolist()

        return JSONResponse(
            status_code=200,
            content={"message": "Item created successfully", "success": True},
        )
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/delete-word/{word_id}", tags=["words"])
async def delete_word(word_id: str) -> dict:
    word_id = int(word_id)
    for word in words_db:
        if int(word["id"]) == word_id:
            words_db.remove(word)

            return {"data": {"Succesful"}}

    return {"data": {"Word not found"}}


#### Category Words

common_category_words = []


@app.get("/api/common-category-words", tags=["category-words"])
async def get_common_category_words() -> dict:

    return {"data": common_category_words}


@app.post("/api/add-common-category-words", tags=["category-words"])
async def add_common_category_words(new_word: dict) -> dict:
    try:
        for word in common_category_words:
            if new_word["item"] == word["item"]:
                raise HTTPException(status_code=400, detail="Word already exists")

        word_embedding = get_embedding(new_word["item"])
        new_word["embedding"] = word_embedding.tolist()
        common_category_words.append(new_word)
        return JSONResponse(
            status_code=200,
            content={"message": "Item created successfully", "success": True},
        )
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/get-embedding", tags=["category-words"])
async def get_embedding_api() -> dict:
    if len(common_category_words) > 1:
        vectors = [word["embedding"] for word in common_category_words]
        variances = np.var(vectors, axis=0)
        low_variance_dims = np.argsort(variances)[:3]
        result = {
            "variances": variances.tolist(),
            "top_variance_dims": low_variance_dims.tolist(),
        }
        return JSONResponse(status_code=200, content={"data": result, "success": True})
    return JSONResponse(
        status_code=200,
        content={
            "data": {"variances": [0] * 50, "top_variance_dims": [0, 1, 2]},
            "message": "Not enough words for analysis",
            "success": False,
        },
    )


@app.delete("/api/delete-common-category-words/{word_id}", tags=["category-words"])
async def delete_common_category_words(word_id: str) -> dict:
    word_id = int(word_id)
    for word in common_category_words:
        if int(word["id"]) == word_id:
            common_category_words.remove(word)

            return {"data": {"Succesful"}}

    return {"data": {"Word not found"}}


### Difference Semantic

from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY


class WordPair(BaseModel):
    word1: str = Field(..., min_length=2)
    word2: str = Field(..., min_length=2)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = exc.errors()
    detailed_messages = [f"Error in {err['loc'][1]}: {err['msg']}" for err in errors]
    print(detailed_messages)
    return JSONResponse(
        status_code=HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": ", ".join(detailed_messages), "body": exc.body},
    )


semantic_difference_db = {}


@app.get("/api/difference-semantic-words", tags=["difference-semantic"])
async def get_semantic_difference() -> dict:

    return JSONResponse(
        status_code=200, content={"data": semantic_difference_db, "success": True}
    )


@app.post("/api/add-difference-semantic-words", tags=["difference-semantic"])
async def add_semantic_difference(new_words: WordPair):
    try:
        first_word = new_words.word1
        second_word = new_words.word2

        combined_word = f"{first_word}-{second_word}"
        word_one_embedding = get_embedding(first_word)
        word_two_embedding = get_embedding(second_word)

        embedding = word_one_embedding - word_two_embedding

        if (
            combined_word in semantic_difference_db
            or combined_word[::-1] in semantic_difference_db
        ):
            raise HTTPException(
                status_code=400, detail="Semantic difference already exists"
            )
        semantic_difference_db[combined_word] = {
            "id": len(semantic_difference_db) + 1,
            "word-1": first_word,
            "word-2": second_word,
            "embedding": embedding.tolist(),
        }
        return JSONResponse(
            status_code=200,
            content={"message": "Item created successfully", "success": True},
        )
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/get-embedding-difference", tags=["difference-semantic"])
async def get_embedding_difference() -> dict:
    try:
        if len(semantic_difference_db) > 1:
            vectors = [word["embedding"] for word in semantic_difference_db.values()]
            variances = np.var(vectors, axis=0)
            low_variance_dims = np.argsort(variances)[:3]
            result = {
                "variance": variances.tolist(),
                "top_variance_dims": low_variance_dims.tolist(),
            }
            return JSONResponse(
                status_code=200, content={"data": result, "success": True}
            )
        return JSONResponse(
            status_code=200,
            content={
                "data": {"variance": [0] * 50, "top_variance_dims": [0, 1, 2]},
                "message": "Not enough words for analysis",
                "success": False,
            },
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete(
    "/api/delete-difference-semantic-words/{word_id}", tags=["difference-semantic"]
)
async def delete_semantic_difference(word_id: str) -> dict:
    try:
        word_id = int(word_id)
        for word in semantic_difference_db.values():
            if int(word["id"]) == word_id:
                del semantic_difference_db[word["word-1"] + "-" + word["word-2"]]
                return {"data": {"Succesful"}}
        return JSONResponse(status_code=404, content={"data": {"Word not found"}})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)