chore: remove legacy files and scripts no longer part of the main architecture

.cursorrules

@@ -0,0 +1,10 @@
+You are a Senior Python Systems Architect specializing in Machine Learning Engineering.
+Your goal is to refactor a research-prototype code base into a production-grade system.
+
+Guidelines:
+1.  **Code Structure**: Follow Clean Architecture principles. Separate concerns strictly between Data Access, Business Logic, and Interface.
+2.  **Type Hinting**: All new or refactored functions MUST have Python type hints and docstrings.
+3.  **No "Glue Scripts"**: Avoid using `subprocess.run` to call other Python scripts. Import classes and call methods instead.
+4.  **Error Handling**: Use specific exception handling, not bare `except Exception`.
+5.  **Paths**: Always use `pathlib.Path`, never `os.path.join`.
+6.  **Refactoring Safety**: When refactoring, ensure existing logic (feature engineering, recall calculation) is preserved unless explicitly asked to simplify.

legacy/README.md

@@ -1,10 +0,0 @@
-# Legacy — Not part of main architecture
-
-Code moved here is preserved but not used in the main flow (src.main FastAPI + React).
-
-| File | Note |
-|:---|:---|
-| app.py | Gradio UI (replaced by React + FastAPI) |
-| agent/ | Shopping agent (broken imports, not used) |
-| deploy.sh | Old Hugging Face deployment script |
-| download_fix.py | Temporary fix script |

legacy/agent/agent_core.py

@@ -1,55 +0,0 @@
-from intent_parser import IntentParser
-from rag_retriever import ProductRetriever
-from dialogue_manager import DialogueManager
-from llm_generator import LLMGenerator
-import os
-
-class ShoppingAgent:
-    def __init__(self, index_path: str, metadata_path: str, llm_model: str = None):
-        self.parser = IntentParser()
-        self.retriever = ProductRetriever(index_path, metadata_path)
-        self.dialogue_manager = DialogueManager()
-        self.llm = LLMGenerator(model_name=llm_model) # Defaults to mock
-        
-    def process_query(self, query: str):
-        print(f"\nUser: {query}")
-        
-        # 1. Parse Intent
-        intent = self.parser.parse(query)
-        # print(f"[Debug] Intent: {intent}")
-        
-        # 2. Enrich Query (incorporating history could happen here)
-        search_query = query
-        if intent['category']:
-            search_query += f" {intent['category']}"
-        
-        # 3. Retrieve
-        results = self.retriever.search(search_query, k=3)
-        
-        # 4. Generate Response using LLM + History
-        history_str = self.dialogue_manager.get_context_string()
-        response = self.llm.generate_response(query, results, history_str)
-        
-        # 5. Update Memory
-        self.dialogue_manager.add_turn(query, response)
-            
-        print("[Agent]:")
-        print(response)
-        return response
-
-    def reset(self):
-        self.dialogue_manager.clear_history()
-
-if __name__ == "__main__":
-    if not os.path.exists("data/product_index.faiss"):
-        print("Index not found. Please run rag_indexer.py first.")
-    else:
-        # Pass "mock" to force CPU-friendly mock generation, 
-        # or pass a model name like "gpt2" (small) if you have 'transformers' installed to test pipeline.
-        agent = ShoppingAgent("data/product_index.faiss", "data/product_metadata.pkl", llm_model="mock")
-        
-        print("--- Turn 1 ---")
-        agent.process_query("I need a gaming laptop under $1000")
-        
-        print("\n--- Turn 2 ---")
-        agent.process_query("Do you have anything cheaper?")

legacy/agent/data_loader.py

@@ -1,61 +0,0 @@
-import pandas as pd
-import random
-
-def generate_synthetic_data(num_samples: int = 100) -> pd.DataFrame:
-    """
-    Generates synthetic e-commerce product data.
-    """
-    categories = ['Electronics', 'Clothing', 'Home & Kitchen', 'Books', 'Toys']
-    adjectives = ['Premium', 'Budget', 'High-end', 'Durable', 'Stylish', 'Compact', 'Professional']
-    products_map = {
-        'Electronics': ['Smartphone', 'Laptop', 'Headphones', 'Smartwatch', 'Camera'],
-        'Clothing': ['T-Shirt', 'Jeans', 'Jacket', 'Sneakers', 'Dress'],
-        'Home & Kitchen': ['Blender', 'Coffee Maker', 'Desk Lamp', 'Sofa', 'Curtains'],
-        'Books': ['Novel', 'Textbook', 'Biography', 'Cookbook', 'Comic'],
-        'Toys': ['Lego Set', 'Action Figure', 'Board Game', 'Puzzle', 'Doll']
-    }
-    
-    data = []
-    for i in range(num_samples):
-        cat = random.choice(categories)
-        prod = random.choice(products_map[cat])
-        adj = random.choice(adjectives)
-        
-        title = f"{adj} {prod} {i+1}"
-        price = round(random.uniform(10.0, 1000.0), 2)
-        description = f"This is a {adj.lower()} {prod.lower()} perfect for your needs. It features high quality materials and modern design."
-        features = f"Feature A, Feature B, {adj} Quality"
-        
-        data.append({
-            'product_id': f"P{str(i).zfill(4)}",
-            'title': title,
-            'category': cat,
-            'price': price,
-            'description': description,
-            'features': features,
-            'review_text': f"Great {prod}! I loved the {adj.lower()} aspect."
-        })
-        
-    return pd.DataFrame(data)
-
-def load_data(file_path: str = None) -> pd.DataFrame:
-    """
-    Loads data from a file or generates synthetic data if path is None.
-    """
-    if file_path:
-        # Check extension and load accordingly
-        if file_path.endswith('.csv'):
-            return pd.read_csv(file_path)
-        elif file_path.endswith('.json'):
-            return pd.read_json(file_path)
-        else:
-            raise ValueError("Unsupported file format")
-    else:
-        print("No file path provided. Generating synthetic data...")
-        return generate_synthetic_data()
-
-if __name__ == "__main__":
-    df = load_data()
-    print(df.head())
-    df.to_csv("synthetic_products.csv", index=False)
-    print("Saved synthetic_products.csv")

legacy/agent/dialogue_manager.py

@@ -1,39 +0,0 @@
-from typing import List, Dict
-
-class DialogueManager:
-    def __init__(self, max_history: int = 5):
-        self.history: List[Dict[str, str]] = []
-        self.max_history = max_history
-
-    def add_turn(self, user_input: str, system_response: str):
-        """
-        Adds a single turn to the history.
-        """
-        self.history.append({"role": "user", "content": user_input})
-        self.history.append({"role": "assistant", "content": system_response})
-        
-        # Keep history within limit (rolling buffer)
-        if len(self.history) > self.max_history * 2:
-            self.history = self.history[-(self.max_history * 2):]
-
-    def get_history(self) -> List[Dict[str, str]]:
-        """
-        Returns the conversation history.
-        """
-        return self.history
-
-    def clear_history(self):
-        """
-        Resets the conversation.
-        """
-        self.history = []
-
-    def get_context_string(self) -> str:
-        """
-        Returns history formatted as a string for simple prompts.
-        """
-        context = ""
-        for turn in self.history:
-            role = "User" if turn["role"] == "user" else "Agent"
-            context += f"{role}: {turn['content']}\n"
-        return context

legacy/agent/intent_parser.py

@@ -1,64 +0,0 @@
-import re
-from typing import Dict, Optional
-
-class IntentParser:
-    def __init__(self):
-        # In a real scenario, this would be an LLM-based parser
-        pass
-        
-    def parse(self, query: str) -> Dict[str, Optional[str]]:
-        """
-        Parses the user query into structured slots.
-        """
-        query = query.lower()
-        
-        intent = {
-            'category': None,
-            'budget': None,
-            'style': None,
-            'original_query': query
-        }
-        
-        # Rule-based Category Extraction
-        categories = ['laptop', 'phone', 'smartphone', 'headphone', 'camera', 'jeans', 'shirt', 'dress', 'shoe', 'blender', 'coffee', 'lamp', 'sofa', 'desk', 'toy', 'lego', 'book', 'novel']
-        for cat in categories:
-            if cat in query:
-                intent['category'] = cat
-                break # Take the first match for now
-        
-        # Rule-based Budget Extraction
-        # Look for "under $100", "cheap", "expensive", "budget"
-        if "cheap" in query or "budget" in query:
-            intent['budget'] = "low"
-        elif "expensive" in query or "premium" in query:
-            intent['budget'] = "high"
-        
-        match = re.search(r'under \$?(\d+)', query)
-        if match:
-            intent['budget'] = f"<{match.group(1)}"
-            
-        # Rule-based Style/Feature Extraction (naïve) 
-        # Everything else that is an adjective could be style
-        styles = ['gaming', 'professional', 'casual', 'formal', 'black', 'red', 'blue', 'wireless', 'bluetooth']
-        found_styles = []
-        for style in styles:
-            if style in query:
-                found_styles.append(style)
-        
-        if found_styles:
-            intent['style'] = ", ".join(found_styles)
-            
-        return intent
-
-if __name__ == "__main__":
-    parser = IntentParser()
-    queries = [
-        "I want a cheap gaming laptop",
-        "Looking for a blue dress under $50",
-        "wireless headphones for travel"
-    ]
-    
-    for q in queries:
-        print(f"Query: {q}")
-        print(f"Parsed: {parser.parse(q)}")
-        print("-" * 20)

legacy/agent/llm_generator.py

@@ -1,92 +0,0 @@
-from typing import List, Dict, Optional
-import os
-
-class LLMGenerator:
-    def __init__(self, model_name: str = None, device: str = "cpu"):
-        """
-        Initialize LLM.
-        Args:
-            model_name: HuggingFace model name (e.g., 'meta-llama/Meta-Llama-3-8B-Instruct').
-                        If None, uses a Mock generator.
-            device: 'cpu' or 'cuda'.
-        """
-        self.model_name = model_name
-        self.device = device
-        self.pipeline = None
-        
-        if self.model_name and self.model_name != "mock":
-            try:
-                from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer
-                import torch
-                
-                print(f"Loading LLM: {model_name} on {device}...")
-                # Note: In a real script, we would handle quantization (bitsandbytes) here
-                # based on the device capabilities we discussed.
-                dtype = torch.float16 if device == 'cuda' else torch.float32
-                
-                self.pipeline = pipeline(
-                    "text-generation",
-                    model=model_name,
-                    torch_dtype=dtype,
-                    device_map="auto" if device == 'cuda' else "cpu"
-                )
-            except Exception as e:
-                print(f"Failed to load model {model_name}: {e}")
-                print("Falling back to Mock Generator.")
-                self.model_name = "mock"
-
-    def generate_response(self, user_query: str, retrieved_items: List[Dict], history_str: str) -> str:
-        """
-        Generates a natural language response based on context.
-        """
-        # 1. Format retrieved items
-        items_str = ""
-        for i, item in enumerate(retrieved_items):
-            items_str += f"{i+1}. {item['title']} (${item['price']}): {item['description']}\n"
-            
-        # 2. Construct Prompt (Simple Template)
-        prompt = f"""You are a helpful shopping assistant.
-        
-Context History:
-{history_str}
-
-Retrieved Products related to the user's request:
-{items_str}
-
-User's Query: {user_query}
-
-Instructions:
-- Recommend the best products from the list above.
-- Explain WHY they fit the user's request (budget, style, category).
-- Be concise and friendly.
-
-Response:"""
-
-        if self.model_name == "mock" or self.model_name is None:
-            return self._mock_generation(items_str)
-        else:
-            # Real LLM Generation
-            try:
-                outputs = self.pipeline(
-                    prompt, 
-                    max_new_tokens=200, 
-                    do_sample=True, 
-                    temperature=0.7,
-                    truncation=True
-                )
-                generated_text = outputs[0]['generated_text']
-                # Extract only the response part if the model echos the prompt (common in base pipelines)
-                if "Response:" in generated_text:
-                    return generated_text.split("Response:")[-1].strip()
-                return generated_text
-            except Exception as e:
-                return f"[Error generating response: {e}]"
-
-    def _mock_generation(self, items_str):
-        """
-        Fallback logic for testing without a GPU.
-        """
-        if not items_str:
-            return "I couldn't find any products matching your specific criteria. Could you try different keywords?"
-        
-        return f"Based on your request, I found these great options:\n{items_str}\nI recommend checking the first one as it offers the best value!"

legacy/agent/rag_indexer.py

@@ -1,56 +0,0 @@
-import os
-import faiss
-import numpy as np
-import pickle
-from sentence_transformers import SentenceTransformer
-import pandas as pd
-try:
-    from src.data_loader import load_data
-except ImportError:
-    from data_loader import load_data # Fallback for direct execution
-
-class RAGIndexer:
-    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
-        self.model = SentenceTransformer(model_name)
-        self.index = None
-        self.metadata = []
-
-    def build_index(self, data: pd.DataFrame):
-        """
-        Builds the Faiss index from the product dataframe.
-        """
-        print("Encoding product data...")
-        # Create a rich text representation for embedding
-        # Title + Description + Features + Category + Price (as text)
-        documents = data.apply(lambda x: f"{x['title']} {x['description']} Category: {x['category']} Price: {x['price']}", axis=1).tolist()
-        
-        embeddings = self.model.encode(documents, show_progress_bar=True)
-        dimension = embeddings.shape[1]
-        
-        self.index = faiss.IndexFlatL2(dimension)
-        self.index.add(embeddings.astype('float32'))
-        
-        self.metadata = data.to_dict('records')
-        print(f"Index built with {len(self.metadata)} items.")
-
-    def save(self, index_path: str, metadata_path: str):
-        """
-        Saves the index and metadata to disk.
-        """
-        if self.index:
-            faiss.write_index(self.index, index_path)
-            with open(metadata_path, 'wb') as f:
-                pickle.dump(self.metadata, f)
-            print(f"Saved index to {index_path} and metadata to {metadata_path}")
-        else:
-            print("No index to save.")
-
-if __name__ == "__main__":
-    # Scaffolding run
-    df = load_data() # Generates synthetic
-    indexer = RAGIndexer()
-    indexer.build_index(df)
-    
-    # Ensure output dir exists
-    os.makedirs("data", exist_ok=True)
-    indexer.save("data/product_index.faiss", "data/product_metadata.pkl")

legacy/agent/rag_retriever.py

@@ -1,42 +0,0 @@
-import faiss
-import pickle
-import numpy as np
-from sentence_transformers import SentenceTransformer
-from typing import List, Dict
-
-class ProductRetriever:
-    def __init__(self, index_path: str, metadata_path: str, model_name: str = 'all-MiniLM-L6-v2'):
-        self.model = SentenceTransformer(model_name)
-        
-        print(f"Loading index from {index_path}...")
-        self.index = faiss.read_index(index_path)
-        
-        print(f"Loading metadata from {metadata_path}...")
-        with open(metadata_path, 'rb') as f:
-            self.metadata = pickle.load(f)
-            
-    def search(self, query: str, k: int = 5) -> List[Dict]:
-        """
-        Searches for the top-k most relevant products.
-        """
-        query_vector = self.model.encode([query]).astype('float32')
-        distances, indices = self.index.search(query_vector, k)
-        
-        results = []
-        for i, idx in enumerate(indices[0]):
-            if idx < len(self.metadata):
-                item = self.metadata[idx]
-                item['score'] = float(distances[0][i])
-                results.append(item)
-                
-        return results
-
-if __name__ == "__main__":
-    # Test run
-    retriever = ProductRetriever("data/product_index.faiss", "data/product_metadata.pkl")
-    query = "cheap gaming laptop"
-    results = retriever.search(query)
-    
-    print(f"Query: {query}")
-    for res in results:
-        print(f" - {res['title']} (${res['price']}) [Score: {res['score']:.4f}]")

legacy/app.py

@@ -1,264 +0,0 @@
-import gradio as gr
-import logging
-import os
-import requests
-import json
-from typing import List, Tuple, Any
-from src.utils import setup_logger
-
-# --- Configuration ---
-API_URL = os.getenv("API_URL", "http://localhost:6006")  # Localhost via SSH Tunnel
-
-# --- Initialize Logger ---
-logger = setup_logger(__name__)
-
-# --- Module Initialization ---
-# (We no longer load model locally; we query the remote API)
-categories = ["All", "Fiction", "History", "Science", "Technology"] # Fallback/Mock for now
-tones = ["All", "Happy", "Surprising", "Angry", "Suspenseful", "Sad"]
-
-def fetch_tones():
-    try:
-        resp = requests.get(f"{API_URL}/tones", timeout=3)
-        if resp.status_code == 200:
-            data = resp.json()
-            tns = data.get("tones") if isinstance(data, dict) else None
-            if isinstance(tns, list) and len(tns) > 0:
-                return tns
-    except Exception as e:
-        logger.warning(f"fetch_tones failed: {e}")
-    return tones
-
-def fetch_categories():
-    try:
-        resp = requests.get(f"{API_URL}/categories", timeout=3)
-        if resp.status_code == 200:
-            data = resp.json()
-            cats = data.get("categories") if isinstance(data, dict) else None
-            if isinstance(cats, list) and len(cats) > 0:
-                return cats
-    except Exception as e:
-        logger.warning(f"fetch_categories failed: {e}")
-    return categories
-
-# Try to fetch real categories on startup
-categories = fetch_categories()
-tones = fetch_tones()
-
-# Initialize Shopping Agent (Mock or Real)
-# Note: Real agent requires FAISS index. We'll handle checks later.
-try:
-    # from legacy.agent.agent_core import ShoppingAgent
-    # shopping_agent = ShoppingAgent(...)
-    pass 
-except ImportError:
-    logger.warning("Shopping Agent module not found or failed to import.")
-
-# --- Business Logic: Tab 1 (Discovery) ---
-def recommend_books(query: str, category: str, tone: str):
-    """Fetch recommendations and return both gallery items and raw data."""
-    try:
-        if not query.strip():
-            return [], []
-        
-        payload = {
-            "query": query,
-            "category": category if category else "All",
-            "tone": tone if tone else "All"
-        }
-        
-        logger.info(f"Sending request to {API_URL}/recommend")
-        response = requests.post(f"{API_URL}/recommend", json=payload, timeout=25)
-        
-        if response.status_code == 200:
-            data = response.json()
-            results = data.get("recommendations", [])
-            gallery_items = [(item["thumbnail"], f"{item['title']}\n{item['authors']}") for item in results]
-            return gallery_items, results
-        else:
-            logger.error(f"API Error: {response.text}")
-            return [], []
-            
-    except Exception as e:
-        logger.error(f"Error in recommend_books: {e}")
-        return [], []
-
-
-def show_book_details(evt: Any, recs: List[dict]):
-    """Populate detail panel when a gallery item is selected and prep a QA hint."""
-    try:
-        if recs is None:
-            return "", "", "", "", "", -1
-        idx = evt.index if evt and hasattr(evt, "index") else None
-        if idx is None or idx >= len(recs):
-            return "", "", "", "", "", -1
-        book = recs[idx]
-        title_block = f"### {book['title']}\n**Authors:** {book['authors']}\n**ISBN:** {book['isbn']}"
-        desc_block = f"**Description**\n\n{book['description']}"
-        rank_block = f"**Rank:** #{idx + 1}"  # simple positional rank
-        comments_block = "**Reviews (sample):**\n- Exceptional pacing and character depth.\n- A must-read for this genre."
-        qa_hint = f"Ask the assistant: Tell me more about '{book['title']}' by {book['authors']}."
-        return title_block, rank_block, comments_block, desc_block, qa_hint, idx
-    except Exception as e:
-        logger.error(f"Error showing book details: {e}")
-        return "", "", "", "", "", -1
-
-def clear_discovery():
-    return "", "All", "All", []
-
-
-def add_to_favorites(selected_idx: int, recs: List[dict]):
-    try:
-        if selected_idx is None or selected_idx < 0 or not recs or selected_idx >= len(recs):
-            return "Please select a book from the gallery first."
-        book = recs[selected_idx]
-        payload = {"user_id": "local", "isbn": book["isbn"]}
-        resp = requests.post(f"{API_URL}/favorites/add", json=payload, timeout=8)
-        if resp.status_code == 200:
-            data = resp.json()
-            return f"✅ Added to favorites: {book['title']} ({data.get('favorites_count', '?')} books in collection)"
-        return f"❌ Failed to add: {resp.text}"
-    except Exception as e:
-        logger.error(f"add_to_favorites error: {e}")
-        return "❌ Error adding to favorites. Try again later."
-
-
-def generate_highlights(selected_idx: int, recs: List[dict]):
-    try:
-        if selected_idx is None or selected_idx < 0 or not recs or selected_idx >= len(recs):
-            return "(Hint) Please select a book from the gallery, then click Generate Highlights."
-        book = recs[selected_idx]
-        payload = {"isbn": book["isbn"], "user_id": "local"}
-        resp = requests.post(f"{API_URL}/marketing/highlights", json=payload, timeout=12)
-        if resp.status_code != 200:
-            return "Failed to generate highlights. Try again later."
-        data = resp.json()
-        persona = data.get("persona", {})
-        highlights = data.get("highlights", [])
-        header = f"### Personalized Highlights ({book['title']})\n"
-        persona_md = f"> Your Profile: {persona.get('summary','N/A')}\n\n" if persona else ""
-        bullets = "\n".join([f"- {h}" for h in highlights]) if highlights else "- No highlights available"
-        return header + persona_md + bullets
-    except Exception as e:
-        logger.error(f"generate_highlights error: {e}")
-        return "Error generating highlights. Try again later."
-
-# --- Business Logic: Tab 2 (Assistant) ---
-def chat_response(message, history):
-    """Answer book questions using the recommender API as a knowledge source."""
-    try:
-        if not message.strip():
-            return "Please describe the book or question you have."
-
-        # Use the same recommend endpoint as retrieval to ground answers
-        payload = {"query": message, "category": "All", "tone": "All"}
-        resp = requests.post(f"{API_URL}/recommend", json=payload, timeout=20)
-        if resp.status_code != 200:
-            return "Unable to retrieve book information. Try again later."
-
-        data = resp.json()
-        recs = data.get("recommendations", [])
-        if not recs:
-            return "No matching books found. Try a different query."
-
-        top = recs[0]
-        answer = [
-            f"**{top.get('title','')}**",
-            f"Author: {top.get('authors','Unknown')}",
-            f"Summary: {top.get('description','No summary available')}"
-        ]
-        # If more results, suggest to check discovery tab
-        if len(recs) > 1:
-            answer.append("More results available in the Find Books tab.")
-        return "\n\n".join(answer)
-    except Exception as e:
-        logger.error(f"chat_response error: {e}")
-        return "Error processing your question. Try again later."
-
-# --- Business Logic: Tab 3 (Marketing) ---
-def generate_marketing_copy(product_name, features, target_audience):
-    # Placeholder for Marketing Content Engine
-    # from src.marketing.guardrails import SafetyCheck...
-    return f"""
-    📣 **CALLING ALL {target_audience.upper()}!**
-    
-    Presenting **{product_name}** — the treasure you've been seeking.
-    
-    ✨ **Why you'll love it:**
-    {features}
-    
-    Perfect for your collection. Add it to your shelf today.
-    """
-
-# --- UI Construction ---
-with gr.Blocks(title="Paper Shelf - Book Discovery", theme=gr.themes.Soft()) as dashboard:
-    
-    gr.Markdown("# 📚 Paper Shelf")
-    gr.Markdown("Intelligent book discovery powered by semantic search: **Find Books**, **Ask Questions**, **Generate Marketing Copy**.")
-    
-    with gr.Tabs():
-        
-        # --- Tab 1: Discovery ---
-        with gr.TabItem("🔍 Find Books (Search & Recommendations)"):
-            rec_state = gr.State([])  # store full recommendation data
-            qa_hint = gr.State("")
-            sel_idx = gr.State(-1)
-            with gr.Row():
-                with gr.Column(scale=3):
-                    q_input = gr.Textbox(label="What are you looking for?", placeholder="e.g., a mystery novel with fast pacing")
-                with gr.Column(scale=1):
-                    cat_input = gr.Dropdown(label="Category", choices=categories, value="All")
-                    tone_input = gr.Dropdown(label="Mood/Tone", choices=tones, value="All")
-            
-            btn_rec = gr.Button("Find Books", variant="primary")
-            gallery = gr.Gallery(label="Results", columns=4, height="auto")
-            with gr.Row():
-                with gr.Column(scale=2):
-                    title_info = gr.Markdown(label="Book Info")
-                    desc_info = gr.Markdown(label="Description")
-                with gr.Column(scale=1):
-                    rank_info = gr.Markdown(label="Ranking")
-                    comments_info = gr.Markdown(label="Reviews")
-            qa_hint_md = gr.Markdown(label="Ask the Assistant", value="(Click a book to see suggested questions)")
-
-            with gr.Row():
-                btn_fav = gr.Button("⭐ Add to Favorites", variant="secondary")
-                btn_high = gr.Button("✨ Generate Highlights", variant="primary")
-            fav_status = gr.Markdown(label="Status")
-            highlights_md = gr.Markdown(label="Personalized Highlights")
-
-            btn_rec.click(recommend_books, [q_input, cat_input, tone_input], [gallery, rec_state])
-            gallery.select(show_book_details, [rec_state], [title_info, rank_info, comments_info, desc_info, qa_hint_md, sel_idx])
-            btn_fav.click(add_to_favorites, [sel_idx, rec_state], [fav_status])
-            btn_high.click(generate_highlights, [sel_idx, rec_state], [highlights_md])
-
-        # --- Tab 2: AI Assistant ---
-        with gr.TabItem("💬 Ask Questions (RAG Assistant)"):
-            chatbot = gr.ChatInterface(
-                fn=chat_response,
-                examples=["Is there a mystery with time travel?", "Recommend sci-fi with female protagonists"],
-                title="Intelligent Book Assistant",
-                description="Search and learn about books through conversational AI."
-            )
-
-        # --- Tab 3: Marketing ---
-        with gr.TabItem("✍️ Create Marketing Copy (GenAI)"):
-            with gr.Row():
-                m_name = gr.Textbox(label="Book Title/Hook", value="The Hobbit - First Edition, Near Mint")
-                m_feat = gr.Textbox(label="Key Features/Condition", value="Near mint condition, no markings, ships worldwide")
-                m_aud = gr.Textbox(label="Target Audience", value="Fantasy enthusiasts, collectors")
-            
-            btn_gen = gr.Button("Generate Listing", variant="primary")
-            m_out = gr.Markdown(label="Generated Copy")
-            
-            btn_gen.click(generate_marketing_copy, [m_name, m_feat, m_aud], m_out)
-
-if __name__ == "__main__":
-    import os
-    assets_path = os.path.join(os.path.dirname(__file__), "assets")
-    dashboard.launch(
-        server_name="0.0.0.0", 
-        server_port=7860,
-        allowed_paths=[assets_path],
-        share=True
-    )

legacy/deploy.sh

@@ -1,47 +0,0 @@
-#!/bin/bash
-
-echo "🚀 准备部署到 Hugging Face Spaces..."
-
-# 检查必要文件
-echo "📋 检查必要文件..."
-required_files=("gradio-dashboard.py" "books_with_emotions.csv" "books_descriptions.txt" "cover-not-found.jpg" "requirements.txt")
-
-for file in "${required_files[@]}"; do
-    if [ -f "$file" ]; then
-        echo "✅ $file 存在"
-    else
-        echo "❌ $file 缺失"
-        exit 1
-    fi
-done
-
-# 重命名主文件为 app.py (Hugging Face 标准)
-if [ -f "gradio-dashboard.py" ]; then
-    cp gradio-dashboard.py app.py
-    echo "✅ 已创建 app.py"
-fi
-
-# 检查 Git 状态
-echo "📝 检查 Git 状态..."
-if [ -d ".git" ]; then
-    echo "✅ Git 仓库已初始化"
-    git status
-else
-    echo "⚠️  未检测到 Git 仓库，请先运行："
-    echo "   git init"
-    echo "   git add ."
-    echo "   git commit -m '准备部署'"
-    echo "   git remote add origin https://github.com/你的用户名/book-recommender.git"
-    echo "   git push -u origin main"
-fi
-
-echo ""
-echo "🎯 下一步操作："
-echo "1. 访问 https://huggingface.co/spaces"
-echo "2. 点击 'Create new Space'"
-echo "3. 选择 'Gradio' SDK"
-echo "4. 连接你的 GitHub 仓库"
-echo "5. 在 Settings 中添加 HUGGINGFACEHUB_API_TOKEN"
-echo "6. 等待自动部署完成"
-echo ""
-echo "📖 详细说明请查看 HUGGINGFACE_DEPLOYMENT.md"

legacy/download_fix.py

@@ -1,11 +0,0 @@
-import os
-os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
-from huggingface_hub import snapshot_download
-
-print("🚀 Downloading model from hf-mirror...")
-snapshot_download(
-    repo_id="sentence-transformers/all-MiniLM-L6-v2",
-    ignore_patterns=["*.bin", "*.h5", "*.ot"],  # 只下载 safetensors，省流
-    resume_download=True
-)
-print("✅ Download Complete!")

scripts/__init__.py

@@ -0,0 +1 @@
+# Pipeline scripts package

scripts/data/__init__.py

@@ -0,0 +1 @@
+# Data processing scripts

scripts/data/build_books_basic_info.py

@@ -1,48 +1,48 @@
-import pandas as pd
 import csv
+import logging
+from pathlib import Path
+
+import pandas as pd
 
-# 读取原始数据，遇到格式错误行自动跳过，保证流程不中断
-books_data = pd.read_csv(
-    "data/books_data.csv",
-    engine="python",
-    quotechar='"',
-    escapechar='\\',
-    on_bad_lines='skip'  # pandas >=1.3
-)
-ratings = pd.read_csv("data/Books_rating.csv", engine="python", quotechar='"', escapechar='\\', on_bad_lines='skip')
-
-# 只保留有用字段
-books_cols = [
-    "Title", "description", "authors", "image", "publisher", "publishedDate", "categories"
-]
-books_data = books_data[books_cols]
-
-# 只保留 Title, Id, review/score 字段用于合并
-ratings_cols = ["Title", "Id", "review/score"]
-ratings = ratings[ratings_cols]
-
-# 去重
-ratings = ratings.drop_duplicates(subset=["Title"])
-
-# 合并，左连接，保留 books_data 所有行
-merged = books_data.merge(ratings, on="Title", how="left")
-
-# 重命名字段
-merged = merged.rename(columns={
-    "Id": "isbn10",
-    "Title": "title",
-    "authors": "authors",
-    "description": "description",
-    "image": "image",
-    "publisher": "publisher",
-    "publishedDate": "publishedDate",
-    "categories": "categories",
-    "review/score": "average_rating"
-})
-
-# 生成 isbn13（如有更复杂规则可补充，这里仅占位）
-merged["isbn13"] = None  # 可后续补充isbn13生成逻辑
-
-# 保存新表，强制所有字段加引号，防止description等字段被截断
-merged.to_csv("data/books_basic_info.csv", index=False, quoting=csv.QUOTE_ALL, quotechar='"', escapechar='\\')
-print("已生成 data/books_basic_info.csv，包含基础书籍信息字段。")
+logger = logging.getLogger(__name__)
+
+
+def run(
+    books_path: Path = Path("data/books_data.csv"),
+    ratings_path: Path = Path("data/Books_rating.csv"),
+    output_path: Path = Path("data/books_basic_info.csv"),
+) -> None:
+    """Build books basic info from raw data. Callable from Pipeline."""
+    books_data = pd.read_csv(
+        str(books_path),
+        engine="python",
+        quotechar='"',
+        escapechar='\\',
+        on_bad_lines='skip',
+    )
+    ratings = pd.read_csv(
+        str(ratings_path),
+        engine="python",
+        quotechar='"',
+        escapechar='\\',
+        on_bad_lines='skip',
+    )
+
+    books_cols = ["Title", "description", "authors", "image", "publisher", "publishedDate", "categories"]
+    books_data = books_data[books_cols]
+    ratings = ratings[["Title", "Id", "review/score"]].drop_duplicates(subset=["Title"])
+    merged = books_data.merge(ratings, on="Title", how="left")
+    merged = merged.rename(columns={
+        "Id": "isbn10", "Title": "title", "authors": "authors", "description": "description",
+        "image": "image", "publisher": "publisher", "publishedDate": "publishedDate",
+        "categories": "categories", "review/score": "average_rating"
+    })
+    merged["isbn13"] = None
+
+    merged.to_csv(str(output_path), index=False, quoting=csv.QUOTE_ALL, quotechar='"', escapechar='\\')
+    logger.info("Saved %s", output_path)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    run()

scripts/data/clean_data.py

@@ -226,23 +226,21 @@ def analyze_data_quality(df: pd.DataFrame, text_columns: list) -> dict:
     return stats
 
 
-def main():
-    parser = argparse.ArgumentParser(description="Clean text data in books dataset")
-    parser.add_argument("--input", type=Path, default=Path("data/books_processed.csv"))
-    parser.add_argument("--output", type=Path, default=None)
-    parser.add_argument("--dry-run", action="store_true", help="Analyze without saving")
-    parser.add_argument("--backup", action="store_true", help="Create backup before overwriting")
-    args = parser.parse_args()
-    
-    if args.output is None:
-        args.output = args.input  # Overwrite by default
-    
-    if not args.input.exists():
-        raise FileNotFoundError(f"Input file not found: {args.input}")
+def run(
+    backup: bool = False,
+    input_path: Optional[Path] = None,
+    output_path: Optional[Path] = None,
+    dry_run: bool = False,
+) -> None:
+    """Clean text data. Callable from Pipeline."""
+    input_path = input_path or Path("data/books_processed.csv")
+    output_path = output_path or input_path
+
+    if not input_path.exists():
+        raise FileNotFoundError(f"Input file not found: {input_path}")
     
-    # Load data
-    logger.info(f"Loading data from {args.input}")
-    df = pd.read_csv(args.input)
+    logger.info(f"Loading data from {input_path}")
+    df = pd.read_csv(input_path)
     logger.info(f"Loaded {len(df):,} records")
     
     # Define columns to clean
@@ -261,31 +259,42 @@ def main():
     for col, s in stats_before.items():
         logger.info(f"  {col}: {s['has_html']} HTML, {s['has_url']} URLs, avg_len={s['avg_length']:.0f}")
     
-    if args.dry_run:
+    if dry_run:
         logger.info("\n[DRY RUN] No changes will be saved")
         return
-    
-    # Clean
+
     logger.info("\n🧹 Cleaning data...")
     df = clean_dataframe(df, text_columns, max_lengths)
-    
-    # Analyze after
+
     logger.info("\n📊 Data quality AFTER cleaning:")
     stats_after = analyze_data_quality(df, text_columns)
     for col, s in stats_after.items():
         logger.info(f"  {col}: {s['has_html']} HTML, {s['has_url']} URLs, avg_len={s['avg_length']:.0f}")
-    
-    # Backup if requested
-    if args.backup and args.output.exists():
-        backup_path = args.output.with_suffix('.csv.bak')
+
+    if backup and output_path.exists():
+        backup_path = output_path.with_suffix('.csv.bak')
         logger.info(f"Creating backup: {backup_path}")
-        args.output.rename(backup_path)
-    
-    # Save
-    logger.info(f"\n💾 Saving to {args.output}")
-    df.to_csv(args.output, index=False)
+        output_path.rename(backup_path)
+
+    logger.info(f"\n💾 Saving to {output_path}")
+    df.to_csv(output_path, index=False)
     logger.info("✅ Done!")
 
 
+def main():
+    parser = argparse.ArgumentParser(description="Clean text data in books dataset")
+    parser.add_argument("--input", type=Path, default=Path("data/books_processed.csv"))
+    parser.add_argument("--output", type=Path, default=None)
+    parser.add_argument("--dry-run", action="store_true", help="Analyze without saving")
+    parser.add_argument("--backup", action="store_true", help="Create backup before overwriting")
+    args = parser.parse_args()
+    run(
+        backup=args.backup,
+        input_path=args.input,
+        output_path=args.output or args.input,
+        dry_run=args.dry_run,
+    )
+
+
 if __name__ == "__main__":
     main()

scripts/data/generate_emotions.py

@@ -76,72 +76,32 @@ def scores_to_vector(scores: List[Dict[str, float]]) -> Dict[str, float]:
     return mapped
 
 
-def main():
-    ap = argparse.ArgumentParser(description="Generate emotion scores from descriptions")
-    ap.add_argument("--input", type=Path, default=Path("data/books_processed.csv"))
-    ap.add_argument("--output", type=Path, default=Path("data/books_processed.csv"))
-    ap.add_argument("--batch-size", type=int, default=16)
-    ap.add_argument("--max-rows", type=int, default=None, help="Optional cap for debugging")
-    ap.add_argument("--device", default=None, help="'mps' for Apple GPU, CUDA device id, or omit for CPU")
-    ap.add_argument("--checkpoint", type=int, default=5000, help="Rows between checkpoint writes")
-    ap.add_argument("--resume", action="store_true", help="Resume if output exists (skip rows with scores)")
-    args = ap.parse_args()
-
-    if not args.input.exists():
-        raise FileNotFoundError(f"Input file not found: {args.input}")
-
-    logger.info("Loading data from %s", args.input)
-    df = pd.read_csv(args.input)
+def run(
+    input_path: Path = Path("data/books_processed.csv"),
+    output_path: Path = Path("data/books_processed.csv"),
+    batch_size: int = 16,
+    device=None,
+) -> None:
+    """Generate emotion scores. Callable from Pipeline."""
+    if not input_path.exists():
+        raise FileNotFoundError(f"Input file not found: {input_path}")
+
+    logger.info("Loading data from %s", input_path)
+    df = pd.read_csv(input_path)
     if "description" not in df.columns:
         raise ValueError("Input CSV must have a 'description' column")
 
-    if args.max_rows:
-        df = df.head(args.max_rows)
-        logger.info("Truncated to %d rows for max_rows", len(df))
-
-    n = len(df)
-    # Normalize device arg
-    dev: str | int | None
-    if args.device is None:
-        dev = None
-    else:
-        if isinstance(args.device, str) and args.device.lower() == "mps":
-            dev = "mps"
-        else:
-            try:
-                dev = int(args.device)
-            except ValueError:
-                dev = None
-    model = load_model(dev)
-
-    # Prepare containers
     for col in TARGET_LABELS:
         if col not in df.columns:
             df[col] = 0.0
 
-    # Resume support: if output exists, and resume flag set, load scores
-    if args.resume and args.output.exists():
-        logger.info("Resume enabled: loading existing output from %s", args.output)
-        df_prev = pd.read_csv(args.output)
-        for col in TARGET_LABELS:
-            if col in df_prev.columns:
-                df[col] = df_prev[col]
-
+    model = load_model(device)
     texts = df["description"].fillna("").astype(str).tolist()
-    batch = args.batch_size
-    checkpoint = max(1, args.checkpoint)
-
-    logger.info("Scoring %d descriptions (batch=%d, checkpoint=%d)...", n, batch, checkpoint)
-    total_batches = (n + batch - 1) // batch
-    for bidx, start in enumerate(tqdm(range(0, n, batch), total=total_batches)):
-        end = min(start + batch, n)
-
-        # Skip already-computed rows when resuming (all scores > 0)
-        if args.resume:
-            existing = df.loc[start:end-1, TARGET_LABELS].values
-            if np.all(existing > 0):
-                continue
+    n = len(df)
 
+    logger.info("Scoring %d descriptions...", n)
+    for start in tqdm(range(0, n, batch_size)):
+        end = min(start + batch_size, n)
         chunk = texts[start:end]
         outputs = model(chunk, truncation=True, max_length=512, top_k=None)
         for i, out in enumerate(outputs):
@@ -150,13 +110,29 @@ def main():
             for col in TARGET_LABELS:
                 df.at[idx, col] = vec[col]
 
-        # periodic checkpoint write
-        if (start > 0) and ((start % checkpoint) == 0):
-            df.to_csv(args.output, index=False)
+    logger.info("Writing to %s", output_path)
+    df.to_csv(output_path, index=False)
+
 
-    logger.info("Writing to %s", args.output)
-    df.to_csv(args.output, index=False)
-    logger.info("Done. Example row: %s", df.head(1)[TARGET_LABELS].to_dict(orient="records"))
+def main():
+    ap = argparse.ArgumentParser(description="Generate emotion scores from descriptions")
+    ap.add_argument("--input", type=Path, default=Path("data/books_processed.csv"))
+    ap.add_argument("--output", type=Path, default=Path("data/books_processed.csv"))
+    ap.add_argument("--batch-size", type=int, default=16)
+    ap.add_argument("--max-rows", type=int, default=None, help="Optional cap for debugging")
+    ap.add_argument("--device", default=None, help="'mps' for Apple GPU, CUDA device id, or omit for CPU")
+    ap.add_argument("--checkpoint", type=int, default=5000, help="Rows between checkpoint writes")
+    ap.add_argument("--resume", action="store_true", help="Resume if output exists (skip rows with scores)")
+    args = ap.parse_args()
+    dev = None
+    if args.device:
+        dev = "mps" if str(args.device).lower() == "mps" else (int(args.device) if str(args.device).isdigit() else None)
+    run(
+        input_path=args.input,
+        output_path=args.output,
+        batch_size=args.batch_size,
+        device=dev,
+    )
 
 
 if __name__ == "__main__":

scripts/data/generate_tags.py

@@ -116,6 +116,30 @@ def compute_tags(corpus: List[str], top_n: int, max_features: int, min_df: int,
     return tags
 
 
+def run(
+    input_path: Path = Path("data/books_processed.csv"),
+    output_path: Path = Path("data/books_processed.csv"),
+    top_n: int = 8,
+    max_features: int = 60000,
+    min_df: int = 5,
+    max_df: float = 0.5,
+) -> None:
+    """Generate per-book tags. Callable from Pipeline."""
+    if not input_path.exists():
+        raise FileNotFoundError(f"Input file not found: {input_path}")
+
+    logger.info("Loading data from %s", input_path)
+    df = pd.read_csv(input_path)
+    if "description" not in df.columns:
+        raise ValueError("Input CSV must have a 'description' column")
+
+    corpus = [normalize_text(x) for x in df["description"].fillna("").astype(str).tolist()]
+    tags = compute_tags(corpus, top_n=top_n, max_features=max_features, min_df=min_df, max_df=max_df)
+    df["tags"] = tags
+    logger.info("Writing tagged data to %s", output_path)
+    df.to_csv(output_path, index=False)
+
+
 def main():
     parser = argparse.ArgumentParser(description="Generate per-book tags from descriptions")
     parser.add_argument("--input", type=Path, default=Path("data/books_processed.csv"))
@@ -125,29 +149,15 @@ def main():
     parser.add_argument("--min-df", type=int, default=5)
     parser.add_argument("--max-df", type=float, default=0.5)
     args = parser.parse_args()
-
-    if not args.input.exists():
-        raise FileNotFoundError(f"Input file not found: {args.input}")
-
-    logger.info("Loading data from %s", args.input)
-    df = pd.read_csv(args.input)
-    if "description" not in df.columns:
-        raise ValueError("Input CSV must have a 'description' column")
-
-    corpus = [normalize_text(x) for x in df["description"].fillna("").astype(str).tolist()]
-    tags = compute_tags(
-        corpus,
+    run(
+        input_path=args.input,
+        output_path=args.output,
         top_n=args.top_n,
         max_features=args.max_features,
         min_df=args.min_df,
         max_df=args.max_df,
     )
 
-    df["tags"] = tags
-    logger.info("Writing tagged data to %s", args.output)
-    df.to_csv(args.output, index=False)
-    logger.info("Done. Sample tags: %s", tags[0:3])
-
 
 if __name__ == "__main__":
     main()

scripts/data/split_rec_data.py

@@ -4,7 +4,7 @@
 
 划分策略: 时序划分 (Leave-Last-Out)
 - 每个用户的最后一次评分 → test
-- 每个用户的倒数第二次评分 → val  
+- 每个用户的倒数第二次评分 → val
 - 其余评分 → train
 
 只保留评分 >= 3 次的用户 (有足够历史)
@@ -15,128 +15,71 @@ import numpy as np
 from pathlib import Path
 from tqdm import tqdm
 import time
+import logging
 
-print('='*60)
-print('推荐系统数据划分')
-print('='*60)
-
-start_time = time.time()
-
-# 路径配置
-DATA_PATH = Path('data/raw/Books_rating.csv')
-OUTPUT_DIR = Path('data/rec')
-OUTPUT_DIR.mkdir(exist_ok=True)
-
-# ==================== 1. 加载数据 ====================
-print('\n[1/5] 加载原始评论数据...')
-# 原始列: Id (ISBN), User_id (用户), review/score, review/time, review/text
-df = pd.read_csv(DATA_PATH, usecols=['Id', 'User_id', 'review/score', 'review/time', 'review/text'])
-df.columns = ['isbn', 'user_id', 'rating', 'timestamp', 'review']
-
-print(f'  原始记录数: {len(df):,}')
-print(f'  用户数: {df["user_id"].nunique():,}')
-print(f'  书籍数: {df["isbn"].nunique():,}')
-
-# ==================== 2. 数据清洗 ====================
-print('\n[2/5] 数据清洗...')
-
-# 去除重复评分 (同一用户对同一本书)
-df = df.drop_duplicates(subset=['user_id', 'isbn'], keep='last')
-print(f'  去重后: {len(df):,}')
-
-# 去除缺失值
-df = df.dropna(subset=['rating', 'timestamp'])
-print(f'  去除缺失后: {len(df):,}')
-
-# 过滤低质量评分 (可选: 只保留 rating > 0)
-df = df[df['rating'] > 0]
-print(f'  过滤低质量后: {len(df):,}')
-
-# ==================== 3. 用户筛选 ====================
-print('\n[3/5] 筛选活跃用户...')
-
-# 统计每个用户的评分数
-user_counts = df.groupby('user_id').size()
-print(f'  评分分布:')
-print(f'    1次: {(user_counts == 1).sum():,}')
-print(f'    2次: {(user_counts == 2).sum():,}')
-print(f'    3-5次: {((user_counts >= 3) & (user_counts <= 5)).sum():,}')
-print(f'    5-10次: {((user_counts > 5) & (user_counts <= 10)).sum():,}')
-print(f'    10+次: {(user_counts > 10).sum():,}')
-
-# 只保留评分 >= 3 次的用户 (需要 1 train + 1 val + 1 test)
-active_users = user_counts[user_counts >= 3].index
-df = df[df['user_id'].isin(active_users)]
-print(f'  活跃用户 (>=3次): {len(active_users):,}')
-print(f'  筛选后记录数: {len(df):,}')
-
-# ==================== 4. 时序划分 ====================
-print('\n[4/5] 时序划分 (Leave-Last-Out)...')
-
-# 按用户和时间排序
-df = df.sort_values(['user_id', 'timestamp'])
-
-train_list = []
-val_list = []
-test_list = []
-
-for user_id, group in tqdm(df.groupby('user_id'), desc='  划分用户'):
-    # 按时间排序
-    group = group.sort_values('timestamp')
-    n = len(group)
-    
-    # 最后一条 → test
-    test_list.append(group.iloc[-1])
-    
-    # 倒数第二条 → val
-    val_list.append(group.iloc[-2])
-    
-    # 其余 → train
-    train_list.extend(group.iloc[:-2].to_dict('records'))
-
-# 转换为 DataFrame
-train_df = pd.DataFrame(train_list)
-val_df = pd.DataFrame(val_list)
-test_df = pd.DataFrame(test_list)
-
-print(f'  训练集: {len(train_df):,} ({len(train_df)/len(df)*100:.1f}%)')
-print(f'  验证集: {len(val_df):,} ({len(val_df)/len(df)*100:.1f}%)')
-print(f'  测试集: {len(test_df):,} ({len(test_df)/len(df)*100:.1f}%)')
-
-# ==================== 5. 保存数据 ====================
-print('\n[5/5] 保存数据...')
-
-train_df.to_csv(OUTPUT_DIR / 'train.csv', index=False)
-val_df.to_csv(OUTPUT_DIR / 'val.csv', index=False)
-test_df.to_csv(OUTPUT_DIR / 'test.csv', index=False)
-
-# 保存用户列表 (用于后续评估)
-active_users_df = pd.DataFrame({'user_id': active_users})
-active_users_df.to_csv(OUTPUT_DIR / 'active_users.csv', index=False)
-
-# 保存统计信息
-stats = {
-    'total_records': len(df),
-    'train_records': len(train_df),
-    'val_records': len(val_df),
-    'test_records': len(test_df),
-    'active_users': len(active_users),
-    'books': df['isbn'].nunique(),
-}
-
-with open(OUTPUT_DIR / 'stats.txt', 'w') as f:
-    for k, v in stats.items():
-        f.write(f'{k}: {v:,}\n')
-
-elapsed = time.time() - start_time
-
-print('\n' + '='*60)
-print('✅ 数据划分完成!')
-print('='*60)
-print(f'输出目录: {OUTPUT_DIR}')
-print(f'  - train.csv: {len(train_df):,} 条')
-print(f'  - val.csv: {len(val_df):,} 条')
-print(f'  - test.csv: {len(test_df):,} 条')
-print(f'  - active_users.csv: {len(active_users):,} 用户')
-print(f'执行时间: {elapsed:.1f}秒')
-print('='*60)
+logger = logging.getLogger(__name__)
+
+DATA_PATH = Path("data/raw/Books_rating.csv")
+OUTPUT_DIR = Path("data/rec")
+
+
+def run(
+    data_path: Path = DATA_PATH,
+    output_dir: Path = OUTPUT_DIR,
+) -> None:
+    """Split train/val/test with Leave-Last-Out. Callable from Pipeline."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    start_time = time.time()
+
+    logger.info("Loading raw ratings...")
+    df = pd.read_csv(data_path, usecols=['Id', 'User_id', 'review/score', 'review/time', 'review/text'])
+    df.columns = ['isbn', 'user_id', 'rating', 'timestamp', 'review']
+    logger.info(f"  Records: {len(df):,}, Users: {df['user_id'].nunique():,}, Items: {df['isbn'].nunique():,}")
+
+    logger.info("Cleaning data...")
+    df = df.drop_duplicates(subset=['user_id', 'isbn'], keep='last')
+    df = df.dropna(subset=['rating', 'timestamp'])
+    df = df[df['rating'] > 0]
+
+    logger.info("Filtering active users (>=3 interactions)...")
+    user_counts = df.groupby('user_id').size()
+    active_users = user_counts[user_counts >= 3].index
+    df = df[df['user_id'].isin(active_users)]
+    logger.info(f"  Active users: {len(active_users):,}, Records: {len(df):,}")
+
+    logger.info("Splitting train/val/test (Leave-Last-Out)...")
+    df = df.sort_values(['user_id', 'timestamp'])
+
+    train_list = []
+    val_list = []
+    test_list = []
+
+    for user_id, group in tqdm(df.groupby('user_id'), desc="  Splitting"):
+        group = group.sort_values('timestamp')
+        test_list.append(group.iloc[-1])
+        val_list.append(group.iloc[-2])
+        train_list.extend(group.iloc[:-2].to_dict('records'))
+
+    train_df = pd.DataFrame(train_list)
+    val_df = pd.DataFrame(val_list)
+    test_df = pd.DataFrame(test_list)
+
+    logger.info(f"  Train: {len(train_df):,}, Val: {len(val_df):,}, Test: {len(test_df):,}")
+
+    train_df.to_csv(output_dir / 'train.csv', index=False)
+    val_df.to_csv(output_dir / 'val.csv', index=False)
+    test_df.to_csv(output_dir / 'test.csv', index=False)
+    pd.DataFrame({'user_id': active_users}).to_csv(output_dir / 'active_users.csv', index=False)
+
+    with open(output_dir / 'stats.txt', 'w') as f:
+        for k, v in [('total_records', len(df)), ('train_records', len(train_df)),
+                     ('val_records', len(val_df)), ('test_records', len(test_df)),
+                     ('active_users', len(active_users)), ('books', df['isbn'].nunique())]:
+            f.write(f'{k}: {v:,}\n')
+
+    logger.info("Split complete in %.1fs", time.time() - start_time)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+    run()

scripts/model/build_recall_models.py

@@ -1,76 +1,61 @@
 #!/usr/bin/env python3
 """
-Build Traditional Recall Models (ItemCF, UserCF, Swing, Popularity, Item2Vec)
+Entry script: Build recall models (ItemCF, UserCF, Swing, Popularity, Item2Vec).
 
-Trains collaborative filtering, embedding-based, and popularity recall models.
-These are CPU-friendly and provide strong baselines.
+All training logic lives in src/recall/*.fit(). This script only loads data,
+imports models, and calls fit().
 
 Usage:
     python scripts/model/build_recall_models.py
 
-Input:
-    - data/rec/train.csv
-
-Output:
-    - data/model/recall/itemcf.pkl   (~1.4 GB)
-    - data/model/recall/usercf.pkl   (~70 MB)
-    - data/model/recall/swing.pkl
-    - data/model/recall/popularity.pkl
-    - data/model/recall/item2vec.pkl
-
-Algorithms:
-    - ItemCF: Co-rating similarity with direction weight (forward=1.0, backward=0.7)
-    - UserCF: User similarity (Jaccard + activity penalty)
-    - Swing: User-pair overlap weighting for substitute relationships
-    - Popularity: Rating count with time decay
-    - Item2Vec: Word2Vec (Skip-gram) on user interaction sequences
+Input:  data/rec/train.csv (columns: user_id, isbn, rating, timestamp)
+Output: data/model/recall/*.pkl, data/recall_models.db (ItemCF)
 """
 
 import sys
-import os
-sys.path.append(os.getcwd())
+from pathlib import Path
+
+# Run from project root
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
 
 import pandas as pd
 import logging
+
 from src.recall.itemcf import ItemCF
 from src.recall.usercf import UserCF
 from src.recall.swing import Swing
 from src.recall.popularity import PopularityRecall
 from src.recall.item2vec import Item2Vec
 
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+TRAIN_PATH = PROJECT_ROOT / "data" / "rec" / "train.csv"
+
+
 def main():
-    logger.info("Loading training data...")
-    df = pd.read_csv('data/rec/train.csv')
-    
-    # 1. ItemCF (force retrain — direction weight updated)
+    logger.info("Loading training data from %s...", TRAIN_PATH)
+    df = pd.read_csv(TRAIN_PATH)
+    logger.info("Loaded %d records.", len(df))
+
     logger.info("--- Training ItemCF ---")
-    itemcf = ItemCF()
-    itemcf.fit(df)
-    
-    # 2. UserCF
+    ItemCF().fit(df)
+
     logger.info("--- Training UserCF ---")
-    usercf = UserCF()
-    usercf.fit(df)
-    
-    # 3. Swing
+    UserCF().fit(df)
+
     logger.info("--- Training Swing ---")
-    swing = Swing()
-    swing.fit(df)
+    Swing().fit(df)
 
-    # 4. Popularity
     logger.info("--- Training Popularity ---")
-    pop = PopularityRecall()
-    pop.fit(df)
-    
-    # 5. Item2Vec
+    PopularityRecall().fit(df)
+
     logger.info("--- Training Item2Vec ---")
-    item2vec = Item2Vec()
-    item2vec.fit(df)
+    Item2Vec().fit(df)
+
+    logger.info("Recall models built and saved successfully.")
 
-    logger.info("Recall models built and saved successfully!")
 
 if __name__ == "__main__":
     main()

scripts/model/train_sasrec.py

@@ -1,204 +1,44 @@
 #!/usr/bin/env python3
 """
-Train SASRec Self-Attentive Sequential Recommendation Model
+Entry script: Train SASRec sequential recommendation model.
 
-A Transformer-based model for sequential recommendation.
-Predicts the next item based on user's historical interaction sequence.
+All training logic lives in SASRecRecall.fit(). This script loads data
+and calls fit().
 
 Usage:
     python scripts/model/train_sasrec.py
 
-Input:
-    - data/rec/user_sequences.pkl
-    - data/rec/item_map.pkl
-
-Output:
-    - data/model/rec/sasrec_model.pth  (model weights)
-    - data/rec/user_seq_emb.pkl        (user sequence embeddings)
-
-Architecture:
-    - Self-Attention layers (Transformer encoder)
-    - Positional embeddings
-    - BCE loss with negative sampling
-
-Recommended:
-    - GPU: 30 epochs, ~20 minutes
-    - The user embeddings are used as features in LGBMRanker and as an independent recall channel
+Input:  data/rec/train.csv
+Output: data/model/rec/sasrec_model.pth
+        data/rec/user_seq_emb.pkl, item_map.pkl, user_sequences.pkl
 """
 
 import sys
-import os
-sys.path.append(os.getcwd())
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
 
-import torch
-import torch.nn as nn
-import torch.optim as optim
-from torch.utils.data import Dataset, DataLoader
-import pickle
-import numpy as np
+import pandas as pd
 import logging
-from tqdm import tqdm
-from pathlib import Path
-from src.model.sasrec import SASRec
 
-logging.basicConfig(level=logging.INFO)
+from src.recall.sasrec_recall import SASRecRecall
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
-class SeqDataset(Dataset):
-    def __init__(self, seqs_dict, num_items, max_len=50):
-        self.seqs = []
-        self.num_items = num_items
-        
-        # Prepare (seq_in, target) pairs
-        for u, s in seqs_dict.items():
-            if len(s) < 2:
-                continue
-            
-            # Pad
-            seq_processed = [0] * max_len
-            seq_len = min(len(s), max_len)
-            seq_processed[-seq_len:] = s[-seq_len:]
-            
-            self.seqs.append(seq_processed)
-            
-        self.seqs = torch.LongTensor(self.seqs)
-        
-    def __len__(self):
-        return len(self.seqs)
-        
-    def __getitem__(self, idx):
-        seq = self.seqs[idx]
-        
-        # Determine pos/neg for training
-        # Target: seq shifted right (positives)
-        pos = np.zeros_like(seq)
-        pos[:-1] = seq[1:]
-        
-        # Negatives: random sample
-        neg = np.random.randint(1, self.num_items + 1, size=len(seq))
-        
-        return seq, torch.LongTensor(pos), torch.LongTensor(neg)
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+TRAIN_PATH = PROJECT_ROOT / "data" / "rec" / "train.csv"
+
+
+def main():
+    logger.info("Loading training data from %s...", TRAIN_PATH)
+    df = pd.read_csv(TRAIN_PATH)
+    logger.info("Loaded %d records.", len(df))
+
+    model = SASRecRecall()
+    model.fit(df)
+    logger.info("SASRec training complete.")
 
-def train_sasrec():
-    max_len = 50
-    hidden_dim = 64
-    batch_size = 128
-    epochs = 30 # Increased from 3
-    lr = 1e-4   # Aligned with optimizer
-    
-    data_dir = Path('data/rec')
-    logger.info("Loading sequences...")
-    with open(data_dir / 'user_sequences.pkl', 'rb') as f:
-        seqs_dict = pickle.load(f)
-        
-    with open(data_dir / 'item_map.pkl', 'rb') as f:
-        item_map = pickle.load(f)
-    num_items = len(item_map)
-    
-    dataset = SeqDataset(seqs_dict, num_items, max_len)
-    dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
-    
-    # Check for MPS (Mac GPU) or CUDA
-    if torch.cuda.is_available():
-        device = torch.device('cuda')
-    elif torch.backends.mps.is_available():
-        device = torch.device('mps')
-    else:
-        device = torch.device('cpu')
-        
-    logger.info(f"Training on {device}")
-    
-    model = SASRec(num_items, max_len, hidden_dim).to(device)
-    optimizer = optim.Adam(model.parameters(), lr=1e-4) # Reduced LR
-    
-    # BCE Loss for Pos/Neg
-    criterion = nn.BCEWithLogitsLoss()
-    
-    model.train()
-    for epoch in range(epochs):
-        total_loss = 0
-        pbar = tqdm(dataloader, desc=f"Epoch {epoch+1}/{epochs}")
-        for seq, pos, neg in pbar:
-            seq = seq.to(device)
-            pos = pos.to(device)
-            neg = neg.to(device)
-            
-            # Forward pass to get seq embeddings: [B, L, H]
-            seq_emb = model(seq) # [B, L, H]
-            
-            # Mask padding (0) in targets
-            mask = (pos != 0)
-            
-            # Get Item Embeddings for Pos and Neg
-            pos_emb = model.item_emb(pos) # [B, L, H]
-            neg_emb = model.item_emb(neg) # [B, L, H]
-            
-            # Calculate logits
-            pos_logits = (seq_emb * pos_emb).sum(dim=-1)
-            neg_logits = (seq_emb * neg_emb).sum(dim=-1)
-            
-            pos_logits = pos_logits[mask]
-            neg_logits = neg_logits[mask]
-            
-            pos_labels = torch.ones_like(pos_logits)
-            neg_labels = torch.zeros_like(neg_logits)
-            
-            loss = criterion(pos_logits, pos_labels) + criterion(neg_logits, neg_labels)
-            
-            optimizer.zero_grad()
-            loss.backward()
-            
-            # Clip Gradient
-            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
-            
-            optimizer.step()
-            
-            total_loss += loss.item()
-            pbar.set_postfix({'loss': total_loss / (pbar.n + 1)})
-            
-    # Save Model
-    torch.save(model.state_dict(), data_dir / '../model/rec/sasrec_model.pth')
-    
-    # Save User Embeddings (Last hidden state)
-    logger.info("Extracting User Sequence Embeddings...")
-    model.eval()
-    user_emb_dict = {}
-    
-    # Create evaluation loader (no shuffle, keep user order)
-    # We need to map back to user_ids, so iterate dict directly
-    
-    # Batch processing for inference
-    all_users = list(seqs_dict.keys())
-    
-    with torch.no_grad():
-        for i in tqdm(range(0, len(all_users), batch_size)):
-            batch_users = all_users[i : i+batch_size]
-            batch_seqs = []
-            for u in batch_users:
-                s = seqs_dict[u]
-                # Same padding logic
-                seq_processed = [0] * max_len
-                seq_len = min(len(s), max_len)
-                if seq_len > 0:
-                    seq_processed[-seq_len:] = s[-seq_len:]
-                batch_seqs.append(seq_processed)
-                
-            input_tensor = torch.LongTensor(batch_seqs).to(device)
-            
-            # Initial forward
-            # Note: During inference, we use the FULL sequence to predict the FUTURE (Test Item)
-            # So input is the full available history
-            
-            output = model(input_tensor) # [B, L, H]
-            last_state = output[:, -1, :].cpu().numpy() # [B, H]
-            
-            for j, u in enumerate(batch_users):
-                user_emb_dict[u] = last_state[j]
-                
-    with open(data_dir / 'user_seq_emb.pkl', 'wb') as f:
-        pickle.dump(user_emb_dict, f)
-        
-    logger.info("User Seq Embeddings saved.")
 
 if __name__ == "__main__":
-    train_sasrec()
+    main()

scripts/model/train_youtube_dnn.py

@@ -1,239 +1,45 @@
 #!/usr/bin/env python3
 """
-Train YoutubeDNN Two-Tower Model for Candidate Retrieval
+Entry script: Train YoutubeDNN Two-Tower recall model.
 
-A deep learning recall model using separate user and item towers.
-Trained with in-batch negative sampling for efficient learning.
+All training logic lives in YoutubeDNNRecall.fit(). This script loads data
+and calls fit().
 
 Usage:
     python scripts/model/train_youtube_dnn.py
 
-Input:
-    - data/rec/user_sequences.pkl
-    - data/rec/item_map.pkl
-    - data/books_processed.csv (for category features)
-
-Output:
-    - data/model/recall/youtube_dnn.pt      (model weights)
-    - data/model/recall/youtube_dnn_meta.pkl (config + mappings)
+Input:  data/rec/train.csv
+Output: data/model/recall/youtube_dnn.pt, youtube_dnn_meta.pkl
+        data/rec/item_map.pkl, user_sequences.pkl
+"""
 
-Architecture:
-    - User Tower: Embedding(history) -> Mean Pooling -> MLP
-    - Item Tower: Embedding(item) + Embedding(category) -> MLP
-    - Training: Contrastive loss with in-batch negatives
+import sys
+from pathlib import Path
 
-Recommended:
-    - GPU: 10-50 epochs, ~30 minutes
-    - CPU: 3-5 epochs for testing only
-"""
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
 
-import numpy as np
 import pandas as pd
-import torch
-import torch.nn as nn
-import torch.optim as optim
-from torch.utils.data import Dataset, DataLoader
-import pickle
-from pathlib import Path
-from tqdm import tqdm
-import sys
-import os
+import logging
+
+from src.recall.embedding import YoutubeDNNRecall
 
-# Add src to path
-sys.path.append(os.path.abspath('.'))
-from src.recall.youtube_dnn import YoutubeDNN
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
 
-# Configuration
-BATCH_SIZE = 512
-EPOCHS = 10
-LR = 0.001
-EMBED_DIM = 64
-MAX_HISTORY = 20
-DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-if torch.backends.mps.is_available():
-    DEVICE = torch.device('mps')
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+TRAIN_PATH = PROJECT_ROOT / "data" / "rec" / "train.csv"
+BOOKS_PATH = PROJECT_ROOT / "data" / "books_processed.csv"
 
-print(f"Using device: {DEVICE}")
 
-def load_data():
-    print("Loading data...")
-    # Load mappings
-    with open('data/rec/item_map.pkl', 'rb') as f:
-        item_map = pickle.load(f)
-    isbn_to_id = item_map
-    id_to_isbn = {v: k for k, v in item_map.items()}
-    
-    # Load sequences
-    with open('data/rec/user_sequences.pkl', 'rb') as f:
-        user_seqs = pickle.load(f)
-        
-    # Load book features for category mapping
-    books_df = pd.read_csv('data/books_processed.csv', usecols=['isbn13', 'simple_categories'])
-    books_df['isbn'] = books_df['isbn13'].astype(str)
-    
-    # Create category map
-    # Categories are often strings like 'Fiction', 'Juvenile Fiction'. We take the first one.
-    cate_map = {'<PAD>': 0, '<UNK>': 1}
-    item_to_cate = {}
-    
-    print("Building category map...")
-    for _, row in books_df.iterrows():
-        isbn = row['isbn']
-        if isbn in isbn_to_id:
-            iid = isbn_to_id[isbn]
-            cates = str(row['simple_categories']).split(';')
-            main_cate = cates[0].strip() if cates else 'Unknown'
-            
-            if main_cate not in cate_map:
-                cate_map[main_cate] = len(cate_map)
-            
-            item_to_cate[iid] = cate_map[main_cate]
-            
-    # Default category for unknown items
-    default_cate = cate_map.get('Unknown', 1)
-    
-    return user_seqs, item_to_cate, len(item_map)+1, len(cate_map), default_cate
+def main():
+    logger.info("Loading training data from %s...", TRAIN_PATH)
+    df = pd.read_csv(TRAIN_PATH)
+    logger.info("Loaded %d records.", len(df))
 
-class RetrievalDataset(Dataset):
-    def __init__(self, user_seqs, item_to_cate, default_cate, max_history=20):
-        self.samples = []
-        self.item_to_cate = item_to_cate
-        self.default_cate = default_cate
-        self.max_history = max_history
-        
-        print("Generating training samples...")
-        # Leave-Last-Out:
-        # Last item -> Test
-        # 2nd Last -> Val
-        # Rest -> Train
-        # So we use items 0 to N-3 for training history generation
-        
-        for user, seq in tqdm(user_seqs.items()):
-            if len(seq) < 3:
-                continue
-            
-            # Use data up to the split point for training
-            # Valid Train Set: seq[:-2]
-            train_seq = seq[:-2]
-            
-            # Generate sliding window samples
-            # minimum history length = 1
-            for i in range(1, len(train_seq)):
-                target = train_seq[i]
-                history = train_seq[:i]
-                
-                # Truncate history
-                if len(history) > max_history:
-                    history = history[-max_history:]
-                
-                self.samples.append((history, target))
-                
-        print(f"Total training samples: {len(self.samples)}")
-        
-    def __len__(self):
-        return len(self.samples)
-        
-    def __getitem__(self, idx):
-        history, target = self.samples[idx]
-        
-        # Padding history
-        padded_hist = np.zeros(self.max_history, dtype=np.int64)
-        length = min(len(history), self.max_history)
-        if length > 0:
-            padded_hist[:length] = history[-length:]
-            
-        target_cate = self.item_to_cate.get(target, self.default_cate)
-        
-        return torch.LongTensor(padded_hist), torch.tensor(target, dtype=torch.long), torch.tensor(target_cate, dtype=torch.long)
+    model = YoutubeDNNRecall()
+    model.fit(df, books_path=BOOKS_PATH)
+    logger.info("YoutubeDNN training complete.")
 
-def train():
-    user_seqs, item_to_cate, vocab_size, cate_vocab_size, default_cate = load_data()
-    
-    dataset = RetrievalDataset(user_seqs, item_to_cate, default_cate, MAX_HISTORY)
-    dataloader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=True, num_workers=0) # mp issue on mac sometimes
-    
-    # Model Setup
-    user_config = {
-        'vocab_size': vocab_size,
-        'embed_dim': EMBED_DIM,
-        'history_len': MAX_HISTORY
-    }
-    item_config = {
-        'vocab_size': vocab_size,
-        'embed_dim': EMBED_DIM,
-        'cate_vocab_size': cate_vocab_size,
-        'cate_embed_dim': 32
-    }
-    model_config = {
-        'hidden_dims': [128, 64],
-        'dropout': 0.1
-    }
-    
-    model = YoutubeDNN(user_config, item_config, model_config).to(DEVICE)
-    optimizer = optim.Adam(model.parameters(), lr=LR)
-    criterion = nn.CrossEntropyLoss() # For In-Batch Negatives
-    
-    print("Start Training...")
-    model.train()
-    
-    for epoch in range(EPOCHS):
-        total_loss = 0
-        steps = 0
-        
-        pbar = tqdm(dataloader)
-        for history, target_item, target_cate in pbar:
-            history = history.to(DEVICE)
-            target_item = target_item.to(DEVICE)
-            target_cate = target_cate.to(DEVICE)
-            
-            optimizer.zero_grad()
-            
-            # Get Vectors
-            user_vec = model.user_tower(history) # (B, D)
-            item_vec = model.item_tower(target_item, target_cate) # (B, D)
-            
-            # Normalize
-            user_vec = nn.functional.normalize(user_vec, p=2, dim=1)
-            item_vec = nn.functional.normalize(item_vec, p=2, dim=1)
-            
-            # In-Batch Negatives
-            # logits[i][j] = user_i dot item_j
-            # We want logits[i][i] to be high
-            logits = torch.matmul(user_vec, item_vec.t()) # (B, B)
-            
-            # Temperature scaling (optional, helps convergence)
-            logits = logits / 0.1 
-            
-            labels = torch.arange(len(user_vec)).to(DEVICE)
-            
-            loss = criterion(logits, labels)
-            loss.backward()
-            optimizer.step()
-            
-            total_loss += loss.item()
-            steps += 1
-            
-            pbar.set_description(f"Epoch {epoch+1} Loss: {total_loss/steps:.4f}")
-            
-        print(f"Epoch {epoch+1} finished. Avg Loss: {total_loss/steps:.4f}")
-        
-    # Save Model
-    save_path = Path('data/model/recall')
-    save_path.mkdir(parents=True, exist_ok=True)
-    
-    torch.save(model.state_dict(), save_path / 'youtube_dnn.pt')
-    
-    # Save metadata
-    meta = {
-        'user_config': user_config,
-        'item_config': item_config,
-        'model_config': model_config,
-        'item_to_cate': item_to_cate
-    }
-    with open(save_path / 'youtube_dnn_meta.pkl', 'wb') as f:
-        pickle.dump(meta, f)
-        
-    print(f"Model saved to {save_path}")
 
-if __name__ == '__main__':
-    train()
+if __name__ == "__main__":
+    main()

scripts/run_pipeline.py

@@ -2,8 +2,8 @@
 """
 Unified Data Pipeline Runner
 
-Executes the complete data processing pipeline in correct order.
-Supports partial runs and validation between stages.
+Orchestrates Data Cleaning -> Training -> Evaluation using direct Python imports.
+No subprocess calls. All logic invoked via Module.run() or src classes.
 
 Usage:
     python scripts/run_pipeline.py                    # Full pipeline
@@ -13,156 +13,208 @@ Usage:
 """
 
 import argparse
-import subprocess
+import logging
 import sys
 import time
 from pathlib import Path
 
-PROJECT_ROOT = Path(__file__).parent.parent
-
-
-def run_script(script_path: str, description: str, args: list = None):
-    """Run a Python script and handle errors."""
-    print(f"\n{'='*60}")
-    print(f"▶️  {description}")
-    print(f"   Script: {script_path}")
-    print("=" * 60)
-    
-    cmd = [sys.executable, script_path]
-    if args:
-        cmd.extend(args)
-    
-    start = time.time()
-    result = subprocess.run(cmd, cwd=PROJECT_ROOT)
-    elapsed = time.time() - start
-    
-    if result.returncode != 0:
-        print(f"\n❌ FAILED: {description} (exit code: {result.returncode})")
-        sys.exit(1)
-    
-    print(f"✅ Completed in {elapsed:.1f}s")
-    return True
+# Ensure project root is on path
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger("pipeline")
+
+
+class Pipeline:
+    """
+    Manages the full data pipeline: Data Cleaning -> Training -> Evaluation.
+
+    All stages use direct Python imports; no subprocess.
+    """
+
+    def __init__(
+        self,
+        project_root: Path = PROJECT_ROOT,
+        device: str | None = None,
+        skip_models: bool = False,
+        skip_index: bool = False,
+        stacking: bool = False,
+    ):
+        self.project_root = Path(project_root)
+        self.data_dir = self.project_root / "data"
+        self.rec_dir = self.data_dir / "rec"
+        self.model_dir = self.data_dir / "model"
+        self.device = device
+        self.skip_models = skip_models
+        self.skip_index = skip_index
+        self.stacking = stacking
+
+    def _run_step(self, name: str, fn, *args, **kwargs):
+        """Run a step with timing log."""
+        logger.info("▶ %s", name)
+        start = time.time()
+        fn(*args, **kwargs)
+        logger.info("  ✓ Done in %.1fs", time.time() - start)
+
+    def run_data_cleaning(self, stage: str = "all") -> None:
+        """Stage 1: Book data processing."""
+        if stage not in ("all", "books"):
+            return
+
+        from scripts.data.clean_data import run as clean_run
+        self._run_step("Clean text data", clean_run, backup=True)
+
+        from scripts.data.build_books_basic_info import run as build_run
+        raw_dir = self.data_dir / "raw"
+        self._run_step("Build books basic info", build_run,
+            books_path=raw_dir / "books_data.csv",
+            ratings_path=raw_dir / "Books_rating.csv",
+            output_path=self.data_dir / "books_basic_info.csv",
+        )
+
+        from scripts.data.generate_emotions import run as emotions_run
+        self._run_step("Generate emotion scores", emotions_run, device=self.device)
+
+        from scripts.data.generate_tags import run as tags_run
+        self._run_step("Generate tags", tags_run)
+
+        from scripts.data.chunk_reviews import chunk_reviews
+        self._run_step("Chunk reviews", chunk_reviews,
+            str(self.data_dir / "review_highlights.txt"),
+            str(self.data_dir / "review_chunks.jsonl"),
+        )
+
+    def run_rec_preparation(self, stage: str = "all") -> None:
+        """Stage 2: RecSys data preparation."""
+        if stage not in ("all", "rec"):
+            return
+
+        from scripts.data.split_rec_data import run as split_run
+        self._run_step("Split train/val/test", split_run,
+            data_path=self.data_dir / "raw" / "Books_rating.csv",
+            output_dir=self.rec_dir,
+        )
+
+        from scripts.data.build_sequences import build_sequences
+        self._run_step("Build user sequences", build_sequences, str(self.rec_dir))
+
+    def run_index_building(self, stage: str = "all") -> None:
+        """Stage 3: Index building."""
+        if stage not in ("all", "index") or self.skip_index:
+            return
+
+        from scripts.init_sqlite_db import init_sqlite_db
+        self._run_step("Build SQLite metadata (books.db)", init_sqlite_db, str(self.data_dir))
+
+        from scripts.data.init_dual_index import init_chunk_index
+        self._run_step("Build chunk vector index", init_chunk_index)
+
+    def run_training(self, stage: str = "all") -> None:
+        """Stage 4: Model training via src imports."""
+        if stage not in ("all", "models") or self.skip_models:
+            return
+
+        train_path = self.rec_dir / "train.csv"
+        if not train_path.exists():
+            logger.warning("train.csv not found, skipping model training")
+            return
+
+        import pandas as pd
+        df = pd.read_csv(train_path)
+
+        from src.recall.itemcf import ItemCF
+        self._run_step("Train ItemCF", lambda: ItemCF().fit(df))
+
+        from src.recall.usercf import UserCF
+        self._run_step("Train UserCF", lambda: UserCF().fit(df))
+
+        from src.recall.swing import Swing
+        self._run_step("Train Swing", lambda: Swing().fit(df))
+
+        from src.recall.popularity import PopularityRecall
+        self._run_step("Train Popularity", lambda: PopularityRecall().fit(df))
+
+        from src.recall.item2vec import Item2Vec
+        self._run_step("Train Item2Vec", lambda: Item2Vec().fit(df))
+
+        from src.recall.embedding import YoutubeDNNRecall
+        self._run_step("Train YoutubeDNN", lambda: YoutubeDNNRecall().fit(
+            df, books_path=self.data_dir / "books_processed.csv"
+        ))
+
+        from src.recall.sasrec_recall import SASRecRecall
+        self._run_step("Train SASRec", lambda: SASRecRecall().fit(df))
+
+        from scripts.model.train_ranker import train_ranker, train_stacking
+        self._run_step("Train Ranker", train_stacking if self.stacking else train_ranker)
+
+    def run_evaluation(self) -> None:
+        """Stage 5: Validation."""
+        def _validate():
+            from scripts.data.validate_data import (
+                validate_raw, validate_processed, validate_rec,
+                validate_index, validate_models,
+            )
+            validate_raw()
+            validate_processed()
+            validate_rec()
+            validate_index()
+            validate_models()
+
+        self._run_step("Validate pipeline", _validate)
+
+    def run(self, stage: str = "all") -> None:
+        """Execute full pipeline: Data Cleaning -> Training -> Evaluation."""
+        logger.info("=" * 60)
+        logger.info("Pipeline: Data Cleaning -> Training -> Evaluation")
+        logger.info("=" * 60)
+
+        start_total = time.time()
+
+        self.run_data_cleaning(stage)
+        self.run_rec_preparation(stage)
+        self.run_index_building(stage)
+        self.run_training(stage)
+        self.run_evaluation()
+
+        elapsed = time.time() - start_total
+        logger.info("=" * 60)
+        logger.info("Pipeline completed in %.1f min", elapsed / 60)
+        logger.info("=" * 60)
 
 
 def main():
-    parser = argparse.ArgumentParser(description="Run data pipeline")
-    parser.add_argument("--stage", choices=[
-        "all", "books", "rec", "index", "models"
-    ], default="all", help="Which stage to run")
+    parser = argparse.ArgumentParser(description="Run data pipeline (no subprocess)")
+    parser.add_argument(
+        "--stage",
+        choices=["all", "books", "rec", "index", "models"],
+        default="all",
+        help="Which stage to run",
+    )
     parser.add_argument("--skip-models", action="store_true", help="Skip model training")
     parser.add_argument("--skip-index", action="store_true", help="Skip index building")
     parser.add_argument("--validate-only", action="store_true", help="Only run validation")
-    parser.add_argument("--device", default=None, help="Device for ML models (cpu/cuda/mps)")
-    parser.add_argument("--stacking", action="store_true", help="Enable Stacking model training (LGBM + XGB + Meta)")
+    parser.add_argument("--device", default=None, help="Device for ML (cpu/cuda/mps)")
+    parser.add_argument("--stacking", action="store_true", help="Enable stacking ranker")
     args = parser.parse_args()
-    
-    print("=" * 60)
-    print("🚀 DATA PIPELINE RUNNER")
-    print("=" * 60)
-    
+
     if args.validate_only:
-        run_script("scripts/data/validate_data.py", "Validating all data")
+        logger.info("Validation only")
+        Pipeline().run_evaluation()
         return
-    
-    start_total = time.time()
-    
-    # ==========================================================================
-    # Stage 1: Book Data Processing
-    # ==========================================================================
-    if args.stage in ["all", "books"]:
-        run_script(
-            "scripts/data/clean_data.py",
-            "Cleaning text data (HTML, encoding, whitespace)",
-            args=["--backup"]
-        )
-        
-        run_script(
-            "scripts/data/build_books_basic_info.py",
-            "Building books basic info"
-        )
-        
-        device_args = ["--device", args.device] if args.device else []
-        run_script(
-            "scripts/data/generate_emotions.py",
-            "Generating emotion scores",
-            args=device_args
-        )
-        
-        run_script(
-            "scripts/data/generate_tags.py",
-            "Generating tags"
-        )
-        
-        run_script(
-            "scripts/data/chunk_reviews.py",
-            "Chunking reviews for Small-to-Big"
-        )
-    
-    # ==========================================================================
-    # Stage 2: RecSys Data Preparation
-    # ==========================================================================
-    if args.stage in ["all", "rec"]:
-        run_script(
-            "scripts/data/split_rec_data.py",
-            "Splitting train/val/test data"
-        )
-        
-        run_script(
-            "scripts/data/build_sequences.py",
-            "Building user sequences"
-        )
-    
-    # ==========================================================================
-    # Stage 3: Index Building
-    # ==========================================================================
-    if args.stage in ["all", "index"] and not args.skip_index:
-        run_script(
-            "scripts/init_sqlite_db.py",
-            "Building SQLite metadata (books.db)"
-        )
-        run_script(
-            "scripts/data/init_dual_index.py",
-            "Building chunk vector index"
-        )
-    
-    # ==========================================================================
-    # Stage 4: Model Training
-    # ==========================================================================
-    if args.stage in ["all", "models"] and not args.skip_models:
-        run_script(
-            "scripts/model/build_recall_models.py",
-            "Building ItemCF/UserCF models"
-        )
-        
-        run_script(
-            "scripts/model/train_youtube_dnn.py",
-            "Training YoutubeDNN (requires GPU)"
-        )
-        
-        run_script(
-            "scripts/model/train_sasrec.py",
-            "Training SASRec (requires GPU)"
-        )
-        
-        ranker_args = ["--stacking"] if args.stacking else []
-        run_script(
-            "scripts/model/train_ranker.py",
-            "Training LGBMRanker (Stacking: {})".format("ON" if args.stacking else "OFF"),
-            args=ranker_args
-        )
-    
-    # ==========================================================================
-    # Final Validation
-    # ==========================================================================
-    run_script(
-        "scripts/data/validate_data.py",
-        "Final validation"
+
+    pipeline = Pipeline(
+        device=args.device,
+        skip_models=args.skip_models,
+        skip_index=args.skip_index,
+        stacking=args.stacking,
     )
-    
-    elapsed_total = time.time() - start_total
-    print("\n" + "=" * 60)
-    print(f"🎉 PIPELINE COMPLETED in {elapsed_total/60:.1f} minutes")
-    print("=" * 60)
+    pipeline.run(stage=args.stage)
 
 
 if __name__ == "__main__":

src/core/model_loader.py

@@ -13,7 +13,7 @@ Usage:
 import os
 import logging
 from pathlib import Path
-from huggingface_hub import hf_hub_download, snapshot_download
+from huggingface_hub import snapshot_download
 from src.config import DATA_DIR
 
 logger = logging.getLogger(__name__)

src/data/__init__.py

@@ -0,0 +1,5 @@
+"""Data access layer for book recommendation system."""
+
+from src.data.repository import DataRepository, data_repository
+
+__all__ = ["DataRepository", "data_repository"]

src/data/repository.py

@@ -0,0 +1,94 @@
+"""
+Unified Data Repository for book recommendation system.
+
+Centralizes all core data access: books metadata, user history, etc.
+Replaces scattered pandas.read_csv and pickle.load calls across services.
+"""
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+import sqlite3
+
+from src.config import DATA_DIR
+from src.core.metadata_store import metadata_store
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+# Core data file paths
+BOOKS_DB_PATH = DATA_DIR / "books.db"
+BOOKS_PROCESSED_CSV = DATA_DIR / "books_processed.csv"
+RECALL_MODELS_DB = DATA_DIR / "recall_models.db"
+
+
+class DataRepository:
+    """
+    Singleton data access layer. Manages loading of books_processed.csv,
+    books.db, recall_models.db (user_history), etc.
+    """
+
+    _instance: Optional["DataRepository"] = None
+
+    def __new__(cls) -> "DataRepository":
+        if cls._instance is None:
+            cls._instance = super(DataRepository, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self) -> None:
+        if getattr(self, "_initialized", False):
+            return
+        self._initialized = True
+        self._recall_conn: Optional[sqlite3.Connection] = None
+        logger.info("DataRepository: Initialized (singleton)")
+
+    def _get_recall_connection(self) -> Optional[sqlite3.Connection]:
+        """Lazy SQLite connection for recall_models.db."""
+        if self._recall_conn is None:
+            if not RECALL_MODELS_DB.exists():
+                logger.warning(f"recall_models.db not found at {RECALL_MODELS_DB}")
+                return None
+            try:
+                self._recall_conn = sqlite3.connect(
+                    str(RECALL_MODELS_DB), check_same_thread=False
+                )
+            except sqlite3.Error as e:
+                logger.error(f"DataRepository: Failed to connect to recall DB: {e}")
+        return self._recall_conn
+
+    def get_book_metadata(self, isbn: str) -> Optional[Dict[str, Any]]:
+        """
+        Get book metadata by ISBN.
+
+        Uses MetadataStore (books.db) as primary source. Returns None if not found.
+        """
+        meta = metadata_store.get_book_metadata(str(isbn))
+        return meta if meta else None
+
+    def get_user_history(self, user_id: str) -> List[str]:
+        """
+        Get user's interaction history (ISBNs) from recall_models.db.
+
+        Used by recommendation algorithms (ItemCF, etc.). Returns empty list if
+        DB unavailable or user has no history.
+        """
+        conn = self._get_recall_connection()
+        if not conn:
+            return []
+        try:
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT isbn FROM user_history WHERE user_id = ?", (user_id,)
+            )
+            return [row[0] for row in cursor.fetchall()]
+        except sqlite3.Error as e:
+            logger.error(f"DataRepository: get_user_history failed: {e}")
+            return []
+
+    def get_all_categories(self) -> List[str]:
+        """Get unique book categories. Delegates to MetadataStore."""
+        return metadata_store.get_all_categories()
+
+
+# Global singleton instance
+data_repository = DataRepository()

src/init_db.py

@@ -1,7 +1,6 @@
 import os
 import shutil
 import sys
-import torch
 from pathlib import Path
 
 # Add project root to Python path
@@ -21,20 +20,10 @@ def init_db():
     # FIX: Disable Tokenizers Parallelism to prevent deadlocks on macOS
     os.environ["TOKENIZERS_PARALLELISM"] = "false"
 
-    # Force CPU for data ingestion to avoid MPS (Metal) async hangs during long processing
-    # We only need speed for inference, reliability is key for building the DB.
+    # Force CPU for data ingestion to avoid MPS (Metal) async hangs during long processing.
+    # Reliability is key for building the DB; GPU acceleration is only needed for inference.
     device = "cpu"
     print("🐢  Forcing CPU for stable database ingestion (prevents macOS Freezes).")
-    
-    # if torch.backends.mps.is_available():
-    #     device = "mps"
-    #     print("⚡️  MacOS GPU (MPS) Detected! switching to GPU acceleration.")
-    # elif torch.cuda.is_available():
-    #     device = "cuda"
-    #     print("⚡️  NVIDIA GPU (CUDA) Detected!")
-    # else:
-    #     device = "cpu"
-    #     print("🐢  No GPU detected, running on CPU (this might be slow).")
 
     # 1. Clear existing DB if any (to avoid duplicates/corruption)
     if CHROMA_DB_DIR.exists():

src/main.py

@@ -15,8 +15,6 @@ from src.user.profile_store import (
     update_book_rating, update_reading_status, update_book_comment,
     get_favorites_with_metadata, get_reading_stats
 )
-from src.marketing.persona import build_persona
-from src.marketing.highlights import generate_highlights
 from src.api.chat import router as chat_router # ✨ NEW
 from src.services.chat_service import chat_service # ✨ NEW
 from src.services.recommend_service import RecommendationService # ✨ NEW
@@ -236,9 +234,6 @@ async def favorites_list(user_id: str):
         favorites_meta = get_favorites_with_metadata(user_id)
         # ENGINEERING IMPROVEMENT: Zero-RAM Lookup
         from src.core.metadata_store import metadata_store
-        
-        results = []
-        # Lazy load fetcher (Handled inside utils now)
         from src.utils import enrich_book_metadata
         
         results = []

src/marketing/persona.py

@@ -1,15 +1,17 @@
 from collections import Counter
-from typing import Dict, List, Any
-import pandas as pd
+from typing import Dict, List, Any, Optional
 
 from src.utils import setup_logger
 
 logger = setup_logger(__name__)
 
 
-def build_persona(fav_isbns: List[str], books: pd.DataFrame) -> Dict[str, Any]:
-    """Aggregate a simple persona from favorites: top authors and categories."""
-    if not isinstance(books, pd.DataFrame) or books.empty or not fav_isbns:
+def build_persona(fav_isbns: List[str], books: Optional[Any] = None) -> Dict[str, Any]:
+    """
+    Aggregate a simple persona from favorites: top authors and categories.
+    Uses MetadataStore for lookups; the books param is legacy and unused.
+    """
+    if not fav_isbns:
         return {
             "summary": "No profile yet. Start by adding your favorite books to see personalized recommendations.",
             "top_authors": [],

src/marketing/personalized_highlight.py

@@ -1,5 +1,4 @@
 import json
-import pandas as pd
 from src.marketing.persona import build_persona
 from src.marketing.highlights import generate_highlights
 

src/marketing/verify_p3.py

@@ -7,7 +7,7 @@ import torch
 from transformers import AutoModelForCausalLM, AutoTokenizer
 from peft import PeftModel
 from modelscope import snapshot_download
-from guardrails import ContentGuardrail
+from src.marketing.guardrails import ContentGuardrail
 
 # Config
 BASE_MODEL_ID = "qwen/Qwen2-7B-Instruct"

src/recall/embedding.py

@@ -5,16 +5,63 @@ V2.7: Replaced torch.matmul brute-force search with Faiss IndexFlatIP
 for SIMD-accelerated inner-product retrieval.
 """
 
-import torch
-import numpy as np
 import pickle
 import logging
-import faiss
 from pathlib import Path
+from typing import Optional
+
+import faiss
+import numpy as np
+import pandas as pd
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torch.utils.data import Dataset, DataLoader
+from tqdm import tqdm
+
+from src.recall.sequence_utils import build_sequences_from_df
 from src.recall.youtube_dnn import YoutubeDNN
 
 logger = logging.getLogger(__name__)
 
+
+class _RetrievalDataset(Dataset):
+    """Internal dataset for YoutubeDNN training (history -> target)."""
+
+    def __init__(self, user_seqs: dict, item_to_cate: dict, default_cate: int, max_history: int):
+        self.samples: list[tuple[list[int], int, int]] = []
+        self.item_to_cate = item_to_cate
+        self.default_cate = default_cate
+        self.max_history = max_history
+
+        for user, seq in user_seqs.items():
+            if len(seq) < 3:
+                continue
+            train_seq = seq[:-2]
+            for i in range(1, len(train_seq)):
+                target = train_seq[i]
+                history = train_seq[:i]
+                if len(history) > max_history:
+                    history = history[-max_history:]
+                target_cate = item_to_cate.get(target, default_cate)
+                self.samples.append((history, target, target_cate))
+
+    def __len__(self) -> int:
+        return len(self.samples)
+
+    def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+        history, target, target_cate = self.samples[idx]
+        padded = np.zeros(self.max_history, dtype=np.int64)
+        length = min(len(history), self.max_history)
+        if length > 0:
+            padded[:length] = history[-length:]
+        return (
+            torch.LongTensor(padded),
+            torch.tensor(target, dtype=torch.long),
+            torch.tensor(target_cate, dtype=torch.long),
+        )
+
+
 class YoutubeDNNRecall:
     def __init__(self, data_dir='data/rec', model_dir='data/model/recall'):
         self.data_dir = Path(data_dir)
@@ -33,7 +80,117 @@ class YoutubeDNNRecall:
         self.id_to_item = {}
         self.meta = None
 
-    def load(self):
+    def fit(
+        self,
+        df: pd.DataFrame,
+        books_path: Optional[Path] = None,
+        epochs: int = 10,
+        batch_size: int = 512,
+        lr: float = 0.001,
+        embed_dim: int = 64,
+        max_history: int = 20,
+    ) -> "YoutubeDNNRecall":
+        """
+        Train YoutubeDNN from interaction DataFrame. Builds sequences internally.
+
+        Args:
+            df: [user_id, isbn, timestamp] (timestamp optional)
+            books_path: Path to books_processed.csv for categories. If None, uses default.
+            epochs, batch_size, lr: Training hyperparameters.
+        """
+        logger.info("Building sequences from DataFrame...")
+        user_seqs, item_map = build_sequences_from_df(df, max_len=50)
+
+        self.item_map = item_map
+        self.id_to_item = {v: k for k, v in item_map.items()}
+        vocab_size = len(item_map) + 1
+
+        # Category map
+        cate_map: dict[str, int] = {"<PAD>": 0, "<UNK>": 1}
+        item_to_cate: dict[int, int] = {}
+        default_cate = 1
+
+        books_path = Path(books_path) if books_path else self.data_dir.parent / "books_processed.csv"
+        if books_path.exists():
+            books_df = pd.read_csv(books_path, usecols=["isbn13", "simple_categories"])
+            books_df["isbn"] = books_df["isbn13"].astype(str)
+            for _, row in books_df.iterrows():
+                isbn = str(row["isbn"])
+                if isbn in item_map:
+                    iid = item_map[isbn]
+                    cates = str(row["simple_categories"]).split(";")
+                    main_cate = cates[0].strip() if cates else "Unknown"
+                    if main_cate not in cate_map:
+                        cate_map[main_cate] = len(cate_map)
+                    item_to_cate[iid] = cate_map[main_cate]
+            default_cate = cate_map.get("Unknown", 1)
+
+        for iid in range(1, vocab_size):
+            if iid not in item_to_cate:
+                item_to_cate[iid] = default_cate
+
+        cate_vocab_size = len(cate_map)
+
+        user_config = {"vocab_size": vocab_size, "embed_dim": embed_dim, "history_len": max_history}
+        item_config = {
+            "vocab_size": vocab_size,
+            "embed_dim": embed_dim,
+            "cate_vocab_size": cate_vocab_size,
+            "cate_embed_dim": 32,
+        }
+        model_config = {"hidden_dims": [128, 64], "dropout": 0.1}
+
+        dataset = _RetrievalDataset(user_seqs, item_to_cate, default_cate, max_history)
+        dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True, num_workers=0)
+
+        self.model = YoutubeDNN(user_config, item_config, model_config).to(self.device)
+        optimizer = optim.Adam(self.model.parameters(), lr=lr)
+        criterion = nn.CrossEntropyLoss()
+
+        logger.info("Training YoutubeDNN...")
+        self.model.train()
+        for epoch in range(epochs):
+            total_loss = 0.0
+            steps = 0
+            for history, target_item, target_cate in tqdm(dataloader, desc=f"Epoch {epoch+1}"):
+                history = history.to(self.device)
+                target_item = target_item.to(self.device)
+                target_cate = target_cate.to(self.device)
+                optimizer.zero_grad()
+                user_vec = self.model.user_tower(history)
+                item_vec = self.model.item_tower(target_item, target_cate)
+                user_vec = nn.functional.normalize(user_vec, p=2, dim=1)
+                item_vec = nn.functional.normalize(item_vec, p=2, dim=1)
+                logits = torch.matmul(user_vec, item_vec.t()) / 0.1
+                labels = torch.arange(len(user_vec)).to(self.device)
+                loss = criterion(logits, labels)
+                loss.backward()
+                optimizer.step()
+                total_loss += loss.item()
+                steps += 1
+            logger.info(f"Epoch {epoch+1} Loss: {total_loss/steps:.4f}")
+
+        self.save_dir.mkdir(parents=True, exist_ok=True)
+        torch.save(self.model.state_dict(), self.model_dir / "youtube_dnn.pt")
+        self.meta = {
+            "user_config": user_config,
+            "item_config": item_config,
+            "model_config": model_config,
+            "item_to_cate": item_to_cate,
+        }
+        with open(self.model_dir / "youtube_dnn_meta.pkl", "wb") as f:
+            pickle.dump(self.meta, f)
+
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+        with open(self.data_dir / "item_map.pkl", "wb") as f:
+            pickle.dump(self.item_map, f)
+        with open(self.data_dir / "user_sequences.pkl", "wb") as f:
+            pickle.dump(user_seqs, f)
+
+        logger.info(f"YoutubeDNN saved to {self.model_dir}")
+        return self
+
+    def load(self) -> bool:
         try:
             logger.info("Loading YoutubeDNN model...")
             # Load metadata

src/recall/fusion.py

@@ -1,5 +1,7 @@
 import logging
 from collections import defaultdict
+from typing import Optional
+
 from src.recall.itemcf import ItemCF
 from src.recall.usercf import UserCF
 from src.recall.popularity import PopularityRecall
@@ -10,8 +12,43 @@ from src.recall.sasrec_recall import SASRecRecall
 
 logger = logging.getLogger(__name__)
 
+# Default: only the 3 most effective channels enabled. Others available but off.
+DEFAULT_CHANNEL_CONFIG = {
+    "itemcf": {"enabled": True, "weight": 1.0},
+    "sasrec": {"enabled": True, "weight": 1.0},
+    "youtube_dnn": {"enabled": True, "weight": 1.0},
+    "usercf": {"enabled": False, "weight": 1.0},
+    "swing": {"enabled": False, "weight": 1.0},
+    "item2vec": {"enabled": False, "weight": 0.8},
+    "popularity": {"enabled": False, "weight": 0.5},
+}
+
+
+def _merge_config(default: dict, override: Optional[dict]) -> dict:
+    """Deep-merge override into default (shallow per channel)."""
+    merged = {k: dict(v) for k, v in default.items()}
+    if override:
+        for ch, cfg in override.items():
+            if ch in merged:
+                merged[ch].update(cfg)
+            else:
+                merged[ch] = dict(cfg)
+    return merged
+
+
 class RecallFusion:
-    def __init__(self, data_dir='data/rec', model_dir='data/model/recall'):
+    def __init__(
+        self,
+        data_dir: str = "data/rec",
+        model_dir: str = "data/model/recall",
+        channel_config: Optional[dict] = None,
+        rrf_k: int = 60,
+    ):
+        self.data_dir = data_dir
+        self.model_dir = model_dir
+        self.channel_config = _merge_config(DEFAULT_CHANNEL_CONFIG, channel_config)
+        self.rrf_k = rrf_k
+
         self.itemcf = ItemCF(data_dir, model_dir)
         self.usercf = UserCF(data_dir, model_dir)
         self.popularity = PopularityRecall(data_dir, model_dir)
@@ -21,8 +58,8 @@ class RecallFusion:
         self.sasrec = SASRecRecall(data_dir, model_dir)
 
         self.models_loaded = False
-        
-    def load_models(self):
+
+    def load_models(self) -> None:
         if self.models_loaded:
             return
 
@@ -35,58 +72,57 @@ class RecallFusion:
         self.item2vec.load()
         self.sasrec.load()
         self.models_loaded = True
-        
-    def get_recall_items(self, user_id, history_items=None, k=100):
+
+    def get_recall_items(self, user_id: str, history_items=None, k: int = 100):
         """
-        Multi-channel recall fusion using RRF
+        Multi-channel recall fusion using RRF. Channels and weights controlled by config.
         """
         if not self.models_loaded:
             self.load_models()
-            
+
         candidates = defaultdict(float)
-        
-        # 1. YoutubeDNN (High weight for potential semantic match)
-        dnn_recs = self.youtube_dnn.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, dnn_recs, weight=0.1)
-        
-        # 2. ItemCF
-        icf_recs = self.itemcf.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, icf_recs, weight=1.0)
-        
-        # 3. UserCF
-        ucf_recs = self.usercf.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, ucf_recs, weight=1.0)
-        
-        # 4. Swing
-        swing_recs = self.swing.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, swing_recs, weight=1.0)
-
-        # 5. SASRec Embedding
-        sas_recs = self.sasrec.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, sas_recs, weight=1.0)
-
-        # 6. Item2Vec
-        i2v_recs = self.item2vec.recommend(user_id, history_items, top_k=k)
-        self._add_to_candidates(candidates, i2v_recs, weight=0.8)
-
-        # 7. Popularity (Filler)
-        pop_recs = self.popularity.recommend(user_id, top_k=k)
-        self._add_to_candidates(candidates, pop_recs, weight=0.5)
-        
-        # Sort by RRF score
+        cfg = self.channel_config
+
+        if cfg.get("youtube_dnn", {}).get("enabled", False):
+            recs = self.youtube_dnn.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["youtube_dnn"]["weight"])
+
+        if cfg.get("itemcf", {}).get("enabled", False):
+            recs = self.itemcf.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["itemcf"]["weight"])
+
+        if cfg.get("usercf", {}).get("enabled", False):
+            recs = self.usercf.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["usercf"]["weight"])
+
+        if cfg.get("swing", {}).get("enabled", False):
+            recs = self.swing.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["swing"]["weight"])
+
+        if cfg.get("sasrec", {}).get("enabled", False):
+            recs = self.sasrec.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["sasrec"]["weight"])
+
+        if cfg.get("item2vec", {}).get("enabled", False):
+            recs = self.item2vec.recommend(user_id, history_items, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["item2vec"]["weight"])
+
+        if cfg.get("popularity", {}).get("enabled", False):
+            recs = self.popularity.recommend(user_id, top_k=k)
+            self._add_to_candidates(candidates, recs, cfg["popularity"]["weight"])
+
         sorted_cands = sorted(candidates.items(), key=lambda x: x[1], reverse=True)
         return sorted_cands[:k]
-    
-    def _add_to_candidates(self, candidates, recs, weight=1.0, rrf_k=60):
+
+    def _add_to_candidates(self, candidates, recs, weight: float) -> None:
         """
-        Add recommendations to candidate pool using RRF
-        score += weight * (1 / (k + rank))
+        Add recommendations to candidate pool using RRF.
+        score += weight * (1 / (rrf_k + rank + 1))
         """
         if not recs:
             return
-            
         for rank, (item, score) in enumerate(recs):
-            rrf_score = weight * (1.0 / (rrf_k + rank + 1))
+            rrf_score = weight * (1.0 / (self.rrf_k + rank + 1))
             candidates[item] += rrf_score
 
 if __name__ == "__main__":

src/recall/itemcf.py

@@ -1,37 +1,150 @@
-import pickle
 import math
-import numpy as np
-import pandas as pd
-from tqdm import tqdm
+import sqlite3
 from collections import defaultdict
 from pathlib import Path
+from typing import Optional
+
+import pandas as pd
+from tqdm import tqdm
+
 import logging
 
 logger = logging.getLogger(__name__)
 
+# Direction weights for asymmetric co-occurrence (CHANGELOG: forward=1.0, backward=0.7)
+FORWARD_WEIGHT = 1.0
+BACKWARD_WEIGHT = 0.7
+
+
 class ItemCF:
     """
     Item-based Collaborative Filtering.
-    
-    ENGINEERING IMPROVEMENT:
-    Transitioned from loading a 7GB+ in-memory similarity matrix (pickle) to an
-    indexed SQLite database (`recall_models.db`). Candidate generation is now
-    offloaded to highly efficient SQL aggregations.
-    
-    This change ensures zero-RAM loading for the similarity matrix while maintaining
-    100% mathematical parity with the original Python implementation.
+
+    Co-occurrence similarity with direction weight: when user reads item A then B,
+    sim(A,B) += 1.0 (forward), sim(B,A) += 0.7 (backward). This captures temporal
+    "read-after" patterns.
+
+    Persists to SQLite (recall_models.db) for zero-RAM inference.
     """
-    def __init__(self, data_dir='data/rec', save_dir='data/model/recall'):
+
+    def __init__(self, data_dir: str = "data/rec", save_dir: str = "data/model/recall"):
         self.data_dir = Path(data_dir)
         self.save_dir = Path(save_dir)
-        self.db_path = Path("data/recall_models.db")
-        self.conn = None
-        
-    def load(self):
+        self.save_dir.mkdir(parents=True, exist_ok=True)
+        self.db_path = self.data_dir.parent / "recall_models.db"
+        self.conn: Optional[sqlite3.Connection] = None
+
+    def fit(self, df: pd.DataFrame, top_k_sim: int = 200) -> "ItemCF":
+        """
+        Build co-occurrence similarity matrix with direction weight, then persist to SQLite.
+
+        Args:
+            df: DataFrame with columns [user_id, isbn, rating, timestamp].
+                If timestamp is missing, assumes row order per user.
+            top_k_sim: Keep only top-k similar items per item to reduce size.
+        """
+        logger.info("Building ItemCF similarity matrix (direction-weighted co-occurrence)...")
+
+        # 1. Build per-user chronologically ordered item sequences
+        user_seqs: dict[str, list[tuple[str, float]]] = defaultdict(list)
+        if "timestamp" in df.columns:
+            for _, row in tqdm(df.iterrows(), total=len(df), desc="Building user sequences"):
+                user_seqs[row["user_id"]].append((str(row["isbn"]), float(row["timestamp"])))
+            for uid in user_seqs:
+                user_seqs[uid] = [x[0] for x in sorted(user_seqs[uid], key=lambda t: t[1])]
+        else:
+            for _, row in tqdm(df.iterrows(), total=len(df), desc="Building user sequences"):
+                user_seqs[row["user_id"]].append(str(row["isbn"]))
+
+        user_hist = {u: list(items) for u, items in user_seqs.items()}
+
+        # 2. Count users per item (for cosine normalization)
+        item_users: dict[str, set[str]] = defaultdict(set)
+        for user_id, items in user_hist.items():
+            for item in items:
+                item_users[item].add(user_id)
+        item_counts = {k: len(v) for k, v in item_users.items()}
+
+        # 3. Build item-item co-occurrence with direction weight
+        sim: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
+        for user_id, items in tqdm(user_hist.items(), desc="Computing co-occurrence"):
+            for i in range(len(items)):
+                item_i = items[i]
+                for j in range(i + 1, len(items)):
+                    item_j = items[j]
+                    # Forward: i before j -> sim(i,j) += 1.0
+                    sim[item_i][item_j] += FORWARD_WEIGHT
+                    # Backward: j after i -> sim(j,i) += 0.7
+                    sim[item_j][item_i] += BACKWARD_WEIGHT
+
+        # 4. Normalize by sqrt(|N_i| * |N_j|) (cosine-style)
+        logger.info("Normalizing ItemCF matrix...")
+        final_sim: dict[str, dict[str, float]] = {}
+        for item_i, related in tqdm(sim.items(), desc="Normalizing"):
+            ni = item_counts.get(item_i, 1)
+            pruned = sorted(related.items(), key=lambda x: x[1], reverse=True)[:top_k_sim]
+            final_sim[item_i] = {}
+            for item_j, raw_score in pruned:
+                nj = item_counts.get(item_j, 1)
+                norm = math.sqrt(ni * nj)
+                if norm > 0:
+                    final_sim[item_i][item_j] = raw_score / norm
+
+        self._sim_matrix = final_sim
+        self._user_hist = user_hist
+        self.save()
+        logger.info(f"ItemCF built: {len(final_sim)} items, saved to {self.db_path}")
+        return self
+
+    def save(self) -> None:
+        """Persist similarity matrix and user history to SQLite."""
+        if not hasattr(self, "_sim_matrix") or not hasattr(self, "_user_hist"):
+            logger.warning("ItemCF.save: No fitted model to save.")
+            return
+
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        conn = sqlite3.connect(str(self.db_path))
+        cursor = conn.cursor()
+
+        cursor.execute("DROP TABLE IF EXISTS item_similarity")
+        cursor.execute("""
+            CREATE TABLE item_similarity (item1 TEXT, item2 TEXT, score REAL)
+        """)
+        cursor.execute("DROP TABLE IF EXISTS user_history")
+        cursor.execute("""
+            CREATE TABLE user_history (user_id TEXT, isbn TEXT)
+        """)
+
+        batch = []
+        for item1, related in tqdm(self._sim_matrix.items(), desc="Writing item_similarity"):
+            for item2, score in related.items():
+                batch.append((item1, item2, score))
+                if len(batch) >= 100000:
+                    cursor.executemany("INSERT INTO item_similarity VALUES (?, ?, ?)", batch)
+                    batch = []
+        if batch:
+            cursor.executemany("INSERT INTO item_similarity VALUES (?, ?, ?)", batch)
+
+        batch = []
+        for user_id, isbns in tqdm(self._user_hist.items(), desc="Writing user_history"):
+            for isbn in isbns:
+                batch.append((user_id, isbn))
+                if len(batch) >= 100000:
+                    cursor.executemany("INSERT INTO user_history VALUES (?, ?)", batch)
+                    batch = []
+        if batch:
+            cursor.executemany("INSERT INTO user_history VALUES (?, ?)", batch)
+
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_item1 ON item_similarity(item1)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_user ON user_history(user_id)")
+        conn.commit()
+        conn.close()
+        logger.info(f"ItemCF saved to {self.db_path}")
+
+    def load(self) -> bool:
         if self.db_path.exists():
-            import sqlite3
             try:
-                self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
+                self.conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
                 logger.info(f"ItemCF: Connected to SQLite {self.db_path}")
                 return True
             except Exception as e:
@@ -86,8 +199,6 @@ class ItemCF:
             logger.error(f"ItemCF Query Error: {e}")
             return []
 
-    def save(self): pass # Migration is done via script
-    def fit(self, df): pass # Training should be done separately
 
 if __name__ == "__main__":
     # Test run

src/recall/popularity.py

@@ -1,5 +1,4 @@
 import pandas as pd
-from collections import defaultdict
 import pickle
 from pathlib import Path
 import logging

src/recall/sasrec_recall.py

@@ -10,13 +10,52 @@ for SIMD-accelerated approximate nearest neighbor search.
 
 import pickle
 import logging
-import numpy as np
-import faiss
 from pathlib import Path
+from typing import Optional
+
+import faiss
+import numpy as np
+import pandas as pd
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torch.utils.data import Dataset, DataLoader
+from tqdm import tqdm
+
+from src.model.sasrec import SASRec
+from src.recall.sequence_utils import build_sequences_from_df
 
 logger = logging.getLogger(__name__)
 
 
+class _SeqDataset(Dataset):
+    """Internal dataset for SASRec training (seq, pos, neg)."""
+
+    def __init__(self, seqs_dict: dict, num_items: int, max_len: int):
+        self.seqs: list[list[int]] = []
+        self.num_items = num_items
+        self.max_len = max_len
+
+        for seq in seqs_dict.values():
+            if len(seq) < 2:
+                continue
+            padded = [0] * max_len
+            seq_len = min(len(seq), max_len)
+            padded[-seq_len:] = seq[-seq_len:]
+            self.seqs.append(padded)
+        self.seqs = torch.LongTensor(self.seqs)
+
+    def __len__(self) -> int:
+        return len(self.seqs)
+
+    def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+        seq = self.seqs[idx]
+        pos = np.zeros_like(seq.numpy())
+        pos[:-1] = seq.numpy()[1:]
+        neg = np.random.randint(1, self.num_items + 1, size=len(seq))
+        return seq, torch.LongTensor(pos), torch.LongTensor(neg)
+
+
 class SASRecRecall:
     def __init__(self, data_dir='data/rec', model_dir='data/model/recall'):
         self.data_dir = Path(data_dir)
@@ -30,7 +69,123 @@ class SASRecRecall:
         self.faiss_index = None  # Faiss IndexFlatIP for fast inner-product search
         self.loaded = False
 
-    def load(self):
+    def fit(
+        self,
+        df: pd.DataFrame,
+        max_len: int = 50,
+        hidden_dim: int = 64,
+        epochs: int = 30,
+        batch_size: int = 128,
+        lr: float = 1e-4,
+    ) -> "SASRecRecall":
+        """
+        Train SASRec from interaction DataFrame. Builds sequences internally.
+
+        Args:
+            df: [user_id, isbn, timestamp] (timestamp optional)
+            max_len, hidden_dim, epochs, batch_size, lr: Training hyperparameters.
+        """
+        logger.info("Building sequences from DataFrame...")
+        user_seqs, item_map = build_sequences_from_df(df, max_len=max_len)
+
+        self.item_map = item_map
+        self.id_to_item = {v: k for k, v in item_map.items()}
+        num_items = len(item_map)
+
+        if torch.cuda.is_available():
+            device = torch.device("cuda")
+        elif torch.backends.mps.is_available():
+            device = torch.device("mps")
+        else:
+            device = torch.device("cpu")
+
+        logger.info(f"Training SASRec on {device}...")
+        dataset = _SeqDataset(user_seqs, num_items, max_len)
+        dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
+
+        model = SASRec(num_items, max_len, hidden_dim).to(device)
+        optimizer = optim.Adam(model.parameters(), lr=lr)
+        criterion = nn.BCEWithLogitsLoss()
+
+        model.train()
+        for epoch in range(epochs):
+            total_loss = 0.0
+            pbar = tqdm(dataloader, desc=f"Epoch {epoch+1}/{epochs}")
+            for seq, pos, neg in pbar:
+                seq, pos, neg = seq.to(device), pos.to(device), neg.to(device)
+                seq_emb = model(seq)
+                mask = pos != 0
+                pos_emb = model.item_emb(pos)
+                neg_emb = model.item_emb(neg)
+                pos_logits = (seq_emb * pos_emb).sum(dim=-1)[mask]
+                neg_logits = (seq_emb * neg_emb).sum(dim=-1)[mask]
+                pos_labels = torch.ones_like(pos_logits)
+                neg_labels = torch.zeros_like(neg_logits)
+                loss = criterion(pos_logits, pos_labels) + criterion(neg_logits, neg_labels)
+                optimizer.zero_grad()
+                loss.backward()
+                torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+                optimizer.step()
+                total_loss += loss.item()
+                pbar.set_postfix(loss=total_loss / (pbar.n + 1))
+
+        # Save model
+        sasrec_dir = self.model_dir.parent / "rec"
+        sasrec_dir.mkdir(parents=True, exist_ok=True)
+        torch.save(model.state_dict(), sasrec_dir / "sasrec_model.pth")
+
+        # Extract user embeddings
+        logger.info("Extracting user sequence embeddings...")
+        model.eval()
+        user_emb_dict: dict = {}
+        all_users = list(user_seqs.keys())
+
+        with torch.no_grad():
+            for i in tqdm(range(0, len(all_users), batch_size), desc="Embedding users"):
+                batch_users = all_users[i : i + batch_size]
+                batch_seqs = []
+                for u in batch_users:
+                    s = user_seqs[u]
+                    padded = [0] * max_len
+                    seq_len = min(len(s), max_len)
+                    if seq_len > 0:
+                        padded[-seq_len:] = s[-seq_len:]
+                    batch_seqs.append(padded)
+                input_tensor = torch.LongTensor(batch_seqs).to(device)
+                output = model(input_tensor)
+                last_state = output[:, -1, :].cpu().numpy()
+                for j, u in enumerate(batch_users):
+                    user_emb_dict[u] = last_state[j]
+
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+        with open(self.data_dir / "user_seq_emb.pkl", "wb") as f:
+            pickle.dump(user_emb_dict, f)
+        with open(self.data_dir / "item_map.pkl", "wb") as f:
+            pickle.dump(self.item_map, f)
+        with open(self.data_dir / "user_sequences.pkl", "wb") as f:
+            pickle.dump(user_seqs, f)
+
+        self.user_seq_emb = user_emb_dict
+        self.user_hist = {
+            u: set(self.id_to_item[idx] for idx in seq if idx in self.id_to_item)
+            for u, seq in user_seqs.items()
+        }
+        self.item_emb = model.item_emb.weight.detach().cpu().numpy()
+        self._build_faiss_index()
+        self.loaded = True
+
+        logger.info(f"SASRec saved to {sasrec_dir}")
+        return self
+
+    def _build_faiss_index(self) -> None:
+        """Build Faiss index from item embeddings."""
+        if self.item_emb is None:
+            return
+        dim = self.item_emb.shape[1]
+        self.faiss_index = faiss.IndexFlatIP(dim)
+        self.faiss_index.add(np.ascontiguousarray(self.item_emb.astype(np.float32)))
+
+    def load(self) -> bool:
         try:
             logger.info("Loading SASRec recall embeddings...")
 

src/recall/sequence_utils.py

@@ -0,0 +1,48 @@
+"""
+Shared utilities for building user sequences from interaction DataFrames.
+Used by SASRec and YoutubeDNN training.
+"""
+
+from typing import Tuple
+
+import pandas as pd
+from tqdm import tqdm
+
+
+def build_sequences_from_df(
+    df: pd.DataFrame, max_len: int = 50
+) -> Tuple[dict[str, list[int]], dict[str, int]]:
+    """
+    Build user sequences and item map from interaction DataFrame.
+
+    Args:
+        df: DataFrame with columns [user_id, isbn] and optionally [timestamp].
+        max_len: Maximum sequence length (truncate from the left).
+
+    Returns:
+        user_seqs: Dict[user_id, list of item_ids] (1-indexed, 0 is padding)
+        item_map: Dict[isbn, item_id]
+    """
+    items = df["isbn"].astype(str).unique()
+    item_map = {isbn: i + 1 for i, isbn in enumerate(items)}
+
+    user_history: dict[str, list[tuple[str, float]]] = {}
+    has_ts = "timestamp" in df.columns
+
+    for _, row in tqdm(df.iterrows(), total=len(df), desc="Building sequences"):
+        u = str(row["user_id"])
+        isbn = str(row["isbn"])
+        ts = float(row["timestamp"]) if has_ts else 0.0
+        if u not in user_history:
+            user_history[u] = []
+        user_history[u].append((isbn, ts))
+
+    user_seqs: dict[str, list[int]] = {}
+    for u, pairs in user_history.items():
+        if has_ts:
+            pairs.sort(key=lambda x: x[1])
+        item_ids = [item_map.get(isbn, 0) for isbn, _ in pairs]
+        item_ids = [x for x in item_ids if x != 0]
+        user_seqs[u] = item_ids[-max_len:]
+
+    return user_seqs, item_map

src/recommender.py

@@ -1,13 +1,9 @@
-import pandas as pd
 from typing import List, Dict, Any
-from src.etl import load_books_data
 from src.vector_db import VectorDB
 from src.config import TOP_K_INITIAL, TOP_K_FINAL, DATA_DIR
 from src.cache import CacheManager
 
-from src.utils import setup_logger, summarize_description
-from src.cover_fetcher import fetch_book_cover
-from src.marketing.personalized_highlight import get_persona_and_highlights
+from src.utils import setup_logger
 from src.core.metadata_store import metadata_store
 
 logger = setup_logger(__name__)

src/services/chat_service.py

@@ -1,22 +1,22 @@
 from typing import Generator, Optional, Dict, Any, List
-import pandas as pd
 from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, BaseMessage
 
 from src.core.llm import LLMFactory
-from src.etl import load_books_data
+from src.data.repository import data_repository
 from src.marketing.persona import build_persona
 from src.user.profile_store import list_favorites
 from src.utils import setup_logger
 
 logger = setup_logger(__name__)
 
+
 class ChatService:
     """
     Service for RAG-based chat interaction.
     Currently focused on 'Chat with Book' (Single Item Context).
+    Uses DataRepository for all book metadata lookups.
     """
     _instance = None
-    _books_df = None
     _history: Dict[str, List[BaseMessage]] = {}
 
     def __new__(cls):
@@ -25,25 +25,11 @@ class ChatService:
         return cls._instance
 
     def __init__(self):
-        # Data is now loaded lazily via _ensure_data
         pass
 
-    def _ensure_data(self):
-        if self._books_df is None:
-            logger.info("ChatService: Lazy-loading books data for context retrieval...")
-            self._books_df = load_books_data()
-
     def _get_book_context(self, isbn: str) -> Optional[Dict[str, Any]]:
-        """Retrieve full context for a specific book by ISBN."""
-        self._ensure_data()
-        # Handle string/int types for ISBN
-        try:
-            row = self._books_df[self._books_df["isbn13"].astype(str) == str(isbn)]
-            if row.empty:
-                return None
-            return row.iloc[0].to_dict()
-        except Exception:
-            return None
+        """Retrieve full context for a specific book by ISBN via DataRepository."""
+        return data_repository.get_book_metadata(str(isbn))
 
     def _format_book_info(self, book: Dict[str, Any]) -> str:
         """Format book metadata into a readable context string."""
@@ -97,8 +83,7 @@ class ChatService:
         """
         Stream chat response for a specific book.
         """
-        self._ensure_data()
-        # 1. Fetch Context
+        # 1. Fetch Context via DataRepository
         book = self._get_book_context(isbn)
         if not book:
             yield "I'm sorry, I couldn't find the details for this book."
@@ -106,7 +91,7 @@ class ChatService:
 
         # 2. Build Persona (User Profile)
         favs = list_favorites(user_id)
-        persona_data = build_persona(favs, self._books_df)
+        persona_data = build_persona(favs)
         user_persona = persona_data.get("summary", "General Reader")
 
         # 3. Construct Prompt with History
@@ -158,15 +143,11 @@ class ChatService:
             yield f"Error generating response: {str(e)}. Please check your API Key."
 
     def add_book_to_context(self, book_data: Dict[str, Any]):
-        """Dynamically add a new book to the ChatService context."""
-        self._ensure_data()
-        try:
-            if self._books_df is not None:
-                new_row_df = pd.DataFrame([book_data])
-                self._books_df = pd.concat([self._books_df, new_row_df], ignore_index=True)
-                logger.info(f"ChatService: Added book {book_data.get('isbn13')} to context.")
-        except Exception as e:
-            logger.error(f"ChatService: Failed to add book to context: {e}")
+        """
+        Called when a new book is added to the system. Book is already in MetadataStore
+        via recommender.add_new_book, so no in-memory cache to update. No-op for now.
+        """
+        logger.info(f"ChatService: Book {book_data.get('isbn13')} added; context served from MetadataStore.")
 
 def get_chat_service():
     """Helper for lazy access to the ChatService singleton."""

src/vector_db.py

@@ -1,10 +1,7 @@
-import gc
 from typing import List, Any
 # Using community version to avoid 'BaseBlobParser' version conflict in langchain-chroma/core
 from langchain_community.vectorstores import Chroma
 from langchain_huggingface import HuggingFaceEmbeddings
-from langchain_community.document_loaders import TextLoader
-from langchain_text_splitters import CharacterTextSplitter
 from src.config import REVIEW_HIGHLIGHTS_TXT, CHROMA_DB_DIR, EMBEDDING_MODEL
 from src.utils import setup_logger
 from src.core.metadata_store import metadata_store