feat: improve the initial /thesaurus-custom endpoint

my_ghost_writer/app.py

@@ -24,7 +24,7 @@ from my_ghost_writer.constants import (ALLOWED_ORIGIN_LIST, API_MODE, DOMAIN, IS
    ME_CONFIG_MONGODB_HEALTHCHECK_SLEEP, ME_CONFIG_MONGODB_USE_OK, PORT, RAPIDAPI_HOST, STATIC_FOLDER,
    STATIC_FOLDER_LITEKOBOLDAINET, WORDSAPI_KEY, WORDSAPI_URL, app_logger)
 from my_ghost_writer.pymongo_utils import mongodb_health_check
-from my_ghost_writer.text_parsers2 import find_synonyms_for_phrase, custom_synonyms
+from my_ghost_writer.text_parsers2 import find_synonyms_for_phrase, custom_synonym_handler
 from my_ghost_writer.thesaurus import get_current_info_wordnet
 from my_ghost_writer.type_hints import (RequestQueryThesaurusInflatedBody, RequestQueryThesaurusWordsapiBody,
     RequestSplitText, RequestTextFrequencyBody, MultiWordSynonymResponse, CustomSynonymRequest)
@@ -295,17 +295,8 @@ async def get_synonyms_for_phrase(body: RequestQueryThesaurusInflatedBody):
 async def add_custom_synonyms(body: CustomSynonymRequest):
     """Adds custom synonyms for a given word to the in-memory store."""
     try:
-        word = body.word.lower()  # Store words in lowercase
-        synonyms = [syn.lower() for syn in body.synonyms] # all the synonyms too!
-
-        if word in custom_synonyms:
-            # Update the synonyms list (append new synonyms, avoid duplicates)
-            custom_synonyms[word] = list(set(custom_synonyms[word] + synonyms))
-            return {"message": f"Synonyms for '{body.word}' updated successfully (in-memory)."}
-        else:
-            # Insert a new entry
-            custom_synonyms[word] = synonyms
-            return {"message": f"Synonyms for '{body.word}' added successfully (in-memory)."}
+        custom_synonym_handler.add_entry(body.word, [r.model_dump() for r in body.related])
+        return {"message": f"Custom entry for '{body.word}' added/updated successfully (in-memory)."}
 
     except Exception as e:
         app_logger.error(f"Error adding custom synonyms: {e}")

my_ghost_writer/custom_synonym_handler.py

@@ -0,0 +1,36 @@
+from typing import Any
+
+
+class CustomSynonymHandler:
+    def __init__(self):
+        self.lexicon: dict[str, list[dict[str, Any]]] = {} # {word: {relation_type: [{word: related_word, definition: definition}]}}
+        self.inverted_index: dict[str, set[str]] = {}  # For reverse lookups
+
+    def add_entry(self, word: str, related: list[dict[str, Any]]):
+        word = word.lower()
+        self.lexicon[word] = {}
+        for relation in related:
+            relation_type = relation["type"]
+            related_words = [{"word": w.lower(), "definition": relation.get("definition")} for w in relation["words"]]
+            self.lexicon[word][relation_type] = related_words
+            self._update_inverted_index(word, relation_type, related_words)
+
+    def get_related(self, word: str, relation_type: str) -> list[dict[str, Any]]:
+        word = word.lower()
+        if word in self.lexicon and relation_type in self.lexicon[word]:
+            return self.lexicon[word][relation_type]
+        return []
+
+    def reverse_lookup(self, related_word: str) -> set[str]:
+        related_word = related_word.lower()
+        if related_word in self.inverted_index:
+            return self.inverted_index[related_word]
+        return set()
+
+    def _update_inverted_index(self, word: str, relation_type: str, related_words: list[dict[str, Any]]):
+        """Updates the inverted index for reverse lookups."""
+        for related in related_words:
+            related_word = related["word"]
+            if related_word not in self.inverted_index:
+                self.inverted_index[related_word] = set()
+            self.inverted_index[related_word].add(word)

my_ghost_writer/text_parsers2.py

@@ -9,10 +9,12 @@ from typing import Any, Optional
 from fastapi import HTTPException
 
 from my_ghost_writer.constants import SPACY_MODEL_NAME, app_logger, ELIGIBLE_POS
+from my_ghost_writer.custom_synonym_handler import CustomSynonymHandler
 from my_ghost_writer.type_hints import SynonymInfo, WordSynonymResult, ContextInfo, SynonymGroup
 
 
 custom_synonyms: dict[str, list[str]] = {}
+custom_synonym_handler = CustomSynonymHandler()
 # Load spaCy model
 try:
     nlp = spacy.load(SPACY_MODEL_NAME)
@@ -187,36 +189,37 @@ def get_wordnet_synonyms(word: str, pos_tag: Optional[str] = None) -> list[dict[
     # 1. Custom Synonym Lookup and Preparation
     custom_synset = None  # Initialize to None
     # 1. Direct Lookup: Check if the word is directly in custom_synonyms
-    if word_lower in custom_synonyms:
-        app_logger.info(f"found custom_synonyms:{custom_synonyms[word_lower]} by word:{word_lower}!")
-        # 2. If custom synonyms exist, create the appropriate structure and return
-        synonyms: list[dict[str, Any]] = [{"synonym": syn, "is_custom": True} for syn in custom_synonyms[word_lower]]
-        if synonyms:
-            # Create a dummy synset for the custom synonyms
+    related_synonyms = custom_synonym_handler.get_related(word_lower, "synonym")
+    if related_synonyms:
+        app_logger.info(f"found custom_synonyms:{related_synonyms} by word:{word_lower}!")
+        synonyms_list: list[dict[str, Any]] = []
+        for related in related_synonyms:
+            synonyms_list.append({"synonym": related["word"], "is_custom": True, "definition": related.get("definition")})
+        if synonyms_list:
             custom_synset = {
                 'definition': 'User-defined synonym.',
                 'examples': [],
-                'synonyms': synonyms
+                'synonyms': synonyms_list
             }
             if pos_tag:
                 custom_synset["pos"] = pos_tag
 
     # 2. Reverse Lookup: Check if the word is a *synonym* of any custom word
-    for custom_word, synonym_list in custom_synonyms.items():
-        if word_lower in synonym_list:
-            app_logger.info(f"found reverse match: '{word_lower}' is a synonym of '{custom_word}'")
-            # Found a reverse match!
-            # Include the original custom_word in the synonym list
-            synonyms = [{"synonym": custom_word, "is_custom": True}]  # Start with the original word
-            synonyms.extend([{"synonym": syn, "is_custom": True} for syn in custom_synonyms[custom_word]])  # Add the rest of the synonyms
-
-            custom_synset = {
-                'definition': f'User-defined synonym (reverse match for "{word}").',
-                'examples': [],
-                'synonyms': synonyms
-            }
-            if pos_tag:
-                custom_synset["pos"] = pos_tag
+    reverse_lookup_words = custom_synonym_handler.reverse_lookup(word_lower)
+
+    if reverse_lookup_words:
+        app_logger.info(f"found reverse match: '{word_lower}' is a synonym of '{reverse_lookup_words}'")
+        # Found a reverse match!
+        # The reverse_lookup return the original word, not a list of synonyms
+        synonyms_list: list[dict[str, Any]] = [{"synonym": reverse_word, "is_custom": True} for reverse_word in reverse_lookup_words]
+
+        custom_synset = {
+            'definition': f'User-defined synonym (reverse match for "{word}").',
+            'examples': [],
+            'synonyms': synonyms_list
+        }
+        if pos_tag:
+            custom_synset["pos"] = pos_tag
 
     # 3. WordNet Lookup
     try:

my_ghost_writer/type_hints.py

@@ -1,16 +1,22 @@
-from typing import Any, TypedDict, Optional
+from typing import Any, TypedDict, Optional, Literal
 from pydantic import BaseModel, field_validator
 
 
+class RelatedEntry(BaseModel):
+    type: Literal["synonym", "antonym", "homonym", "homophone", "homograph"]
+    words: list[str]
+    definition: Optional[str] = None  # Definition is now within RelatedEntry
+
+
 class CustomSynonymRequest(BaseModel):
     word: str
-    synonyms: list[str]
+    related: list[RelatedEntry]
 
-    @field_validator("synonyms")
-    def synonyms_must_not_be_empty(cls, v):
-        if not v:
-            raise ValueError("Synonym list cannot be empty.")
-        return v
+    # @field_validator("synonyms")
+    # def synonyms_must_not_be_empty(cls, v):
+    #     if not v:
+    #         raise ValueError("Synonym list cannot be empty.")
+    #     return v
 
 
 class SynonymInfo(TypedDict):
@@ -18,7 +24,6 @@ class SynonymInfo(TypedDict):
     is_custom: bool
 
 
-
 class RequestWordQueryBody(BaseModel):
     word: str
 

tests/events/request_thesaurus_custom.json

@@ -1,4 +1,15 @@
 {
   "word": "happy",
-  "synonyms": ["joyful", "content", "cheerful", "elated"]
+  "related": [
+    {
+      "definition": "Feeling or showing pleasure or contentment.",
+      "type": "synonym",
+      "words": ["joyful", "content", "cheerful", "elated"]
+    },
+    {
+      "definition": "Feeling or showing sorrow; unhappy.",
+      "type": "antonym",
+      "words": ["sad", "unhappy"]
+    }
+  ]
 }

tests/events/request_thesaurus_custom2.json

@@ -0,0 +1,10 @@
+{
+  "word": "joy",
+  "related": [
+    {
+      "definition": "Feeling or showing pleasure or contentment.",
+      "type": "synonym",
+      "words": ["cheer", "happyness"]
+    }
+  ]
+}

tests/events/request_thesaurus_custom3.json

@@ -0,0 +1,9 @@
+{
+  "word": "march",
+  "related": [
+    {
+      "type": "synonym",
+      "words": ["walk", "run"]
+    }
+  ]
+}