init

.gitignore

@@ -0,0 +1,4 @@
+# venv
+.venv
+# pycache
+__pycache__

app.py

@@ -0,0 +1,47 @@
+import requests
+from flask import Flask, request, jsonify
+from smart_search import SmartSearch
+from tempfile import NamedTemporaryFile
+
+app = Flask(__name__)
+
+# Function to download data from API and save to a temporary file
+def download_and_save_data(url: str) -> str:
+    response = requests.get(url)
+    response.raise_for_status()  # Ensure we raise an error for bad responses
+    # Create a temporary file
+    with NamedTemporaryFile(delete=False, suffix='.json') as tmp_file:
+        tmp_file.write(response.content)
+        return tmp_file.name
+
+# URLs for the data
+films_url = 'https://unicone-studio-load-balancer.hf.space/api/film/all'
+tv_series_url = 'https://unicone-studio-load-balancer.hf.space/api/tv/all'
+
+# Download and save data
+films_file = download_and_save_data(films_url)
+tv_series_file = download_and_save_data(tv_series_url)
+
+# Initialize the SmartSearch instance with the paths to the data files
+search_system = SmartSearch(films_file, tv_series_file)
+
+@app.route('/api/search', methods=['POST'])
+def search():
+    # Ensure the request is JSON
+    if not request.is_json:
+        return jsonify({"error": "Request must be JSON"}), 400
+    
+    # Extract the query from the JSON body
+    data = request.get_json()
+    query = data.get('query')
+    
+    if not query:
+        return jsonify({"error": "Missing 'query' field in JSON body"}), 400
+    
+    # Perform the search
+    results = search_system.search(query)
+    
+    return jsonify(results)
+
+if __name__ == '__main__':
+    app.run(debug=True, host="0.0.0.0")

benchmark.py

@@ -0,0 +1,105 @@
+import time
+from smart_search import SmartSearch
+import matplotlib.pyplot as plt
+from sklearn.metrics import precision_recall_fscore_support
+
+# Initialize SmartSearch with the data files
+films_file = 'films.json'
+tv_series_file = 'tv_series.json'
+search_system = SmartSearch(films_file, tv_series_file)
+
+# Define a list of test queries (including misspelled queries) and expected results
+test_queries = [
+    {"query": "my spy", "expected_films": ["My spy 2020", "My spy 2024"], "expected_tv_series": []},
+    {"query": "my sp", "expected_films": ["My spy 2020", "My spy 2024"], "expected_tv_series": []},
+    {"query": "funky monk", "expected_films": ["Funky Monkey 2004"], "expected_tv_series": []},
+    {"query": "yaariyan", "expected_films": ["Yaariayan (2014)", "Yaariyan 2 (2023)"], "expected_tv_series": []},
+    {"query": "grand blu", "expected_films": [], "expected_tv_series": ["Grand Blue"]},
+    {"query": "aho girl", "expected_films": [], "expected_tv_series": ["Aho Girl"]},
+]
+
+# Benchmarking function
+def benchmark_search(search_system, queries):
+    detailed_results = []
+    total_time = 0
+    
+    for test_case in queries:
+        query = test_case['query']
+        expected_films = test_case['expected_films']
+        expected_tv_series = test_case['expected_tv_series']
+        
+        start_time = time.time()
+        result = search_system.search(query)
+        end_time = time.time()
+        
+        elapsed_time = end_time - start_time
+        total_time += elapsed_time
+        
+        # Calculate precision, recall, and F1 score using 'weighted' average
+        films_precision, films_recall, films_f1, _ = precision_recall_fscore_support(
+            expected_films, result['films'], average='weighted')
+        tv_series_precision, tv_series_recall, tv_series_f1, _ = precision_recall_fscore_support(
+            expected_tv_series, result['tv_series'], average='weighted')
+
+        detailed_results.append({
+            "query": query,
+            "result": result,
+            "elapsed_time": elapsed_time,
+            "films_precision": films_precision,
+            "films_recall": films_recall,
+            "films_f1": films_f1,
+            "tv_series_precision": tv_series_precision,
+            "tv_series_recall": tv_series_recall,
+            "tv_series_f1": tv_series_f1
+        })
+    
+    avg_time = total_time / len(queries)
+    return detailed_results, avg_time
+
+# Run the benchmark
+detailed_results, avg_time = benchmark_search(search_system, test_queries)
+
+# Displaying results
+def display_results(detailed_results, avg_time):
+    queries = [result['query'] for result in detailed_results]
+    times = [result['elapsed_time'] for result in detailed_results]
+    films_f1_scores = [result['films_f1'] for result in detailed_results]
+    tv_series_f1_scores = [result['tv_series_f1'] for result in detailed_results]
+
+    print(f"Average search time: {avg_time:.4f} seconds\n")
+    
+    for result in detailed_results:
+        print(f"Query: '{result['query']}'")
+        print(f"Time taken: {result['elapsed_time']:.4f} seconds")
+        print("Films found:", result['result']['films'])
+        print("TV Series found:", result['result']['tv_series'])
+        print(f"Films Precision: {result['films_precision']:.2f}")
+        print(f"Films Recall: {result['films_recall']:.2f}")
+        print(f"Films F1 Score: {result['films_f1']:.2f}")
+        print(f"TV Series Precision: {result['tv_series_precision']:.2f}")
+        print(f"TV Series Recall: {result['tv_series_recall']:.2f}")
+        print(f"TV Series F1 Score: {result['tv_series_f1']:.2f}")
+        print("-" * 50)
+    
+    # Plotting the time taken for each query
+    plt.figure(figsize=(10, 6))
+    plt.barh(queries, times, color='skyblue')
+    plt.xlabel('Time (seconds)')
+    plt.title('Time Taken for Each Search Query')
+    plt.show()
+
+    # Plotting the F1 scores for films and TV series
+    plt.figure(figsize=(10, 6))
+    width = 0.35
+    indices = range(len(queries))
+    plt.bar(indices, films_f1_scores, width, label='Films F1 Score', color='green')
+    plt.bar([i + width for i in indices], tv_series_f1_scores, width, label='TV Series F1 Score', color='orange')
+    plt.xticks([i + width / 2 for i in indices], queries)
+    plt.xlabel('Queries')
+    plt.ylabel('F1 Score')
+    plt.title('F1 Scores for Films and TV Series')
+    plt.legend(loc='best')
+    plt.show()
+
+# Run the display function to show results
+display_results(detailed_results, avg_time)

requirements.txt

@@ -0,0 +1,3 @@
+Flask 
+fuzzywuzzy 
+python-Levenshtein 

smart_search.py

@@ -0,0 +1,68 @@
+import json
+import re
+from fuzzywuzzy import fuzz
+from typing import List, Dict, Tuple, Union
+
+class SmartSearch:
+    def __init__(self, films_file: str, tv_series_file: str):
+        self.films, self.tv_series = self.load_data(films_file, tv_series_file)
+        self.index = self.create_index(self.films, self.tv_series)
+
+    def load_data(self, films_file: str, tv_series_file: str) -> Tuple[List[str], Dict[str, List[Dict[str, str]]]]:
+        with open(films_file, 'r') as f:
+            films = json.load(f)
+            
+        with open(tv_series_file, 'r') as f:
+            tv_series = json.load(f)
+            
+        # Create a mapping for normalized titles to their original titles
+        films_normalized = {}
+        for film in films:
+            normalized_film = re.sub(r'^films/', '', film).lower()
+            films_normalized[normalized_film] = re.sub(r'^films/', '', film)
+        
+        # Normalize TV series titles and episodes, and map to original
+        tv_series_normalized = {}
+        for series, episodes in tv_series.items():
+            series_normalized = re.sub(r'^tv/', '', series).lower()
+            episodes_normalized = [
+                {
+                    "episode": re.sub(r'^tv/.*?/.*?/', '', ep['episode']),
+                    "path": ep['path'],
+                    "season": ep['season']
+                }
+                for ep in episodes
+            ]
+            tv_series_normalized[series_normalized] = {
+                "original": series,
+                "episodes": episodes_normalized
+            }
+        
+        return films_normalized, tv_series_normalized
+
+    def create_index(self, films: Dict[str, str], tv_series: Dict[str, Dict[str, Union[str, List[Dict[str, str]]]]]) -> Dict[str, Union[Dict[str, str], Dict[str, Dict[str, Union[str, List[Dict[str, str]]]]]]]:
+        return {
+            'films': films,
+            'tv_series': tv_series
+        }
+
+    def search(self, query: str) -> Dict[str, List[str]]:
+        query = query.lower()
+        results = {'films': [], 'tv_series': []}
+        
+        # Search films
+        films = self.index['films']
+        results['films'] = [films[film] for film in films if query in film or fuzz.partial_ratio(query, film) > 80]
+        
+        # Search TV series
+        tv_series = self.index['tv_series']
+        for series, data in tv_series.items():
+            if query in series or fuzz.partial_ratio(query, series) > 80:
+                results['tv_series'].append(data['original'])
+            else:
+                for episode in data['episodes']:
+                    if query in episode['episode'].lower() or fuzz.partial_ratio(query, episode['episode'].lower()) > 80:
+                        results['tv_series'].append(data['original'])
+                        break
+        
+        return results

test_api.py

@@ -0,0 +1,38 @@
+import requests
+import json
+
+# Define the API endpoint
+url = "http://127.0.0.1:5000/api/search"
+
+# Define the search queries you want to test
+test_queries = [
+    {"query": "my spy"},
+    {"query": "ahoh girl"},
+    {"query": "yarian"},
+    {"query": "grand blue"},
+    {"query": "Non-existent Title"},
+    {"query": "yariyan 203"},
+    {"query": "My spey"}  # Intentional typo to test fuzzy matching
+]
+
+def test_search_api():
+    for query in test_queries:
+        print(f"Testing query: {query['query']}")
+        
+        # Send the POST request
+        response = requests.post(url, json=query)
+        
+        # Check if the request was successful
+        if response.status_code == 200:
+            # Parse the JSON response
+            results = response.json()
+            print(f"Results for '{query['query']}':")
+            print(json.dumps(results, indent=4))
+        else:
+            print(f"Failed to get results for '{query['query']}'. Status code: {response.status_code}")
+            print(response.text)
+        
+        print("\n" + "="*50 + "\n")
+
+if __name__ == "__main__":
+    test_search_api()