From 39242d5e6c49499dc091987a4544219f34f0422a Mon Sep 17 00:00:00 2001 From: Stepan Vladovskiy Date: Wed, 12 Mar 2025 14:13:55 -0300 Subject: [PATCH] debug: add logs in search.py and change and input validation ... index ver too --- services/search.py | 192 +++++++++++++++++++++++++++++++++++++-------- 1 file changed, 161 insertions(+), 31 deletions(-) diff --git a/services/search.py b/services/search.py index c8186802..7b5bfb11 100644 --- a/services/search.py +++ b/services/search.py @@ -3,24 +3,26 @@ import json import logging import os import httpx +import time from services.redis import redis from utils.encoders import CustomJSONEncoder -# Set redis logging level to suppress DEBUG messages +# Set up proper logging logger = logging.getLogger("search") -logger.setLevel(logging.WARNING) +logger.setLevel(logging.INFO) # Change to INFO to see more details REDIS_TTL = 86400 # 1 day in seconds # Configuration for search service SEARCH_ENABLED = bool(os.environ.get("SEARCH_ENABLED", "true").lower() in ["true", "1", "yes"]) -TXTAI_SERVICE_URL = os.environ.get("TXTAI_SERVICE_URL") +TXTAI_SERVICE_URL = os.environ.get("TXTAI_SERVICE_URL", "http://search-txtai.web.1:8000") +MAX_BATCH_SIZE = int(os.environ.get("SEARCH_MAX_BATCH_SIZE", "100")) class SearchService: def __init__(self): - logger.info("Initializing search service...") + logger.info(f"Initializing search service with URL: {TXTAI_SERVICE_URL}") self.available = SEARCH_ENABLED self.client = httpx.AsyncClient(timeout=30.0, base_url=TXTAI_SERVICE_URL) @@ -35,7 +37,9 @@ class SearchService: try: response = await self.client.get("/info") response.raise_for_status() - return response.json() + result = response.json() + logger.info(f"Search service info: {result}") + return result except Exception as e: logger.error(f"Failed to get search info: {e}") return {"status": "error", "message": str(e)} @@ -69,63 +73,163 @@ class SearchService: shout.media or "" ])) + if not text.strip(): + logger.warning(f"No text content to index for shout {shout.id}") + return + + # Log the document being indexed + logger.info(f"Indexing document: ID={shout.id}, Text length={len(text)}") + # Send to txtai service response = await self.client.post( "/index", json={"id": str(shout.id), "text": text} ) response.raise_for_status() - logger.info(f"Post {shout.id} successfully indexed") + result = response.json() + logger.info(f"Post {shout.id} successfully indexed: {result}") except Exception as e: logger.error(f"Indexing error for shout {shout.id}: {e}") async def bulk_index(self, shouts): """Index multiple documents at once""" if not self.available or not shouts: + logger.warning(f"Bulk indexing skipped: available={self.available}, shouts_count={len(shouts) if shouts else 0}") return + + start_time = time.time() + logger.info(f"Starting bulk indexing of {len(shouts)} documents") - documents = [] - for shout in shouts: - text = " ".join(filter(None, [ - shout.title or "", - shout.subtitle or "", - shout.lead or "", - shout.body or "", - shout.media or "" - ])) - documents.append({"id": str(shout.id), "text": text}) + # Process documents in batches + batch_size = MAX_BATCH_SIZE + total_indexed = 0 + total_skipped = 0 + + for i in range(0, len(shouts), batch_size): + batch = shouts[i:i+batch_size] + logger.info(f"Processing batch {i//batch_size + 1} of {(len(shouts)-1)//batch_size + 1}, size {len(batch)}") - try: - response = await self.client.post( - "/bulk-index", - json={"documents": documents} - ) - response.raise_for_status() - logger.info(f"Bulk indexed {len(documents)} documents") - except Exception as e: - logger.error(f"Bulk indexing error: {e}") + documents = [] + for shout in batch: + try: + # Clean and combine all text fields + text_fields = [] + for field_name in ['title', 'subtitle', 'lead', 'body']: + field_value = getattr(shout, field_name, None) + if field_value and isinstance(field_value, str) and field_value.strip(): + text_fields.append(field_value.strip()) + + # Process media field if it exists + media = getattr(shout, 'media', None) + if media: + if isinstance(media, str): + # Try to parse if it's JSON + try: + media_json = json.loads(media) + if isinstance(media_json, dict) and 'title' in media_json: + text_fields.append(media_json['title']) + if isinstance(media_json, dict) and 'body' in media_json: + text_fields.append(media_json['body']) + except json.JSONDecodeError: + text_fields.append(media) + elif isinstance(media, dict): + if 'title' in media: + text_fields.append(media['title']) + if 'body' in media: + text_fields.append(media['body']) + + # Combine fields into one text + text = " ".join(text_fields) + + if not text.strip(): + logger.debug(f"Skipping shout {shout.id}: no text content") + total_skipped += 1 + continue + + # Add to batch + documents.append({ + "id": str(shout.id), + "text": text + }) + total_indexed += 1 + + except Exception as e: + logger.error(f"Error processing shout {getattr(shout, 'id', 'unknown')} for indexing: {e}") + total_skipped += 1 + + if not documents: + logger.warning(f"No valid documents in batch {i//batch_size + 1}") + continue + + try: + # Log a sample of the batch for debugging + if documents: + sample = documents[0] + logger.info(f"Sample document: id={sample['id']}, text_length={len(sample['text'])}") + + # Send batch to txtai service + logger.info(f"Sending batch of {len(documents)} documents to search service") + response = await self.client.post( + "/bulk-index", + json={"documents": documents} + ) + response.raise_for_status() + result = response.json() + logger.info(f"Batch {i//batch_size + 1} indexed successfully: {result}") + except Exception as e: + logger.error(f"Bulk indexing error for batch {i//batch_size + 1}: {e}") + + elapsed = time.time() - start_time + logger.info(f"Bulk indexing completed in {elapsed:.2f}s: {total_indexed} indexed, {total_skipped} skipped") async def search(self, text, limit, offset): """Search documents""" if not self.available: + logger.warning("Search not available") return [] + + # Validate input + if not isinstance(text, str) or not text.strip(): + logger.warning(f"Invalid search text: {text}") + return [] + + logger.info(f"Searching for: '{text}' (limit={limit}, offset={offset})") # Check Redis cache first redis_key = f"search:{text}:{offset}+{limit}" cached = await redis.get(redis_key) if cached: - return json.loads(cached) - - logger.info(f"Searching: {text} {offset}+{limit}") + cached_results = json.loads(cached) + logger.info(f"Retrieved {len(cached_results)} results from cache for '{text}'") + return cached_results try: + # Log request + logger.info(f"Sending search request: text='{text}', limit={limit}, offset={offset}") + + # Send search request to txtai service response = await self.client.post( "/search", json={"text": text, "limit": limit, "offset": offset} ) response.raise_for_status() + + # Log raw response for debugging + logger.info(f"Raw search response: {response.text}") + + # Parse response result = response.json() + logger.info(f"Parsed search response: {result}") + + # Extract results formatted_results = result.get("results", []) + logger.info(f"Search for '{text}' returned {len(formatted_results)} results") + + # Log sample results for debugging + if formatted_results: + logger.info(f"Sample result: {formatted_results[0]}") + else: + logger.warning(f"No results found for '{text}'") # Cache results if formatted_results: @@ -137,7 +241,7 @@ class SearchService: ) return formatted_results except Exception as e: - logger.error(f"Search error: {e}") + logger.error(f"Search error for '{text}': {e}", exc_info=True) return [] @@ -157,6 +261,32 @@ async def search_text(text: str, limit: int = 50, offset: int = 0): async def initialize_search_index(shouts_data): """Initialize search index with existing data during application startup""" if SEARCH_ENABLED: - logger.info("Initializing search index with existing data...") + if not shouts_data: + logger.warning("No shouts data provided for search indexing") + return + + logger.info(f"Initializing search index with {len(shouts_data)} documents") + + # Check if search service is available first + info = await search_service.info() + if info.get("status") in ["error", "unavailable", "disabled"]: + logger.error(f"Cannot initialize search index: {info}") + return + + # Start the bulk indexing process await search_service.bulk_index(shouts_data) - logger.info(f"Search index initialized with {len(shouts_data)} documents") \ No newline at end of file + + # Verify indexing worked by testing with a search + try: + test_query = "test" + logger.info(f"Verifying search index with query: '{test_query}'") + test_results = await search_text(test_query, 5) + + if test_results: + logger.info(f"Search verification successful: found {len(test_results)} results") + else: + logger.warning("Search verification returned no results. Index may be empty or not working.") + except Exception as e: + logger.error(f"Error verifying search index: {e}") + else: + logger.info("Search indexing skipped (SEARCH_ENABLED=False)") \ No newline at end of file