131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
"""Рекомендательная система для заданий (MVP-1)."""
|
||
from typing import Dict, List, Optional
|
||
|
||
import numpy as np
|
||
from sklearn.feature_extraction.text import TfidfVectorizer
|
||
from sklearn.metrics.pairwise import cosine_similarity
|
||
|
||
|
||
class RecommendationEngine:
|
||
"""Простая рекомендательная система на основе TF-IDF."""
|
||
|
||
def __init__(self):
|
||
self.vectorizer = TfidfVectorizer(max_features=100, stop_words="english")
|
||
self.task_vectors = None
|
||
self.tasks = []
|
||
|
||
def fit(self, tasks: List[Dict]):
|
||
"""
|
||
Обучить модель на исторических данных.
|
||
|
||
Args:
|
||
tasks: Список заданий с полями: title, description, category, completed
|
||
"""
|
||
self.tasks = tasks
|
||
|
||
# Создаем текстовые описания для векторизации
|
||
texts = []
|
||
for task in tasks:
|
||
text = f"{task.get('title', '')} {task.get('description', '')} {task.get('category', '')}"
|
||
texts.append(text)
|
||
|
||
if texts:
|
||
self.task_vectors = self.vectorizer.fit_transform(texts)
|
||
|
||
def recommend(
|
||
self,
|
||
preferences: List[str],
|
||
completed_tasks: Optional[List[str]] = None,
|
||
top_k: int = 5,
|
||
) -> List[Dict]:
|
||
"""
|
||
Рекомендовать задания на основе предпочтений.
|
||
|
||
Args:
|
||
preferences: Предпочтения пользователя
|
||
completed_tasks: Список уже выполненных заданий (для исключения)
|
||
top_k: Количество рекомендаций
|
||
|
||
Returns:
|
||
Список рекомендованных заданий
|
||
"""
|
||
if not self.tasks or self.task_vectors is None:
|
||
return []
|
||
|
||
# Векторизуем предпочтения
|
||
preferences_text = " ".join(preferences)
|
||
preference_vector = self.vectorizer.transform([preferences_text])
|
||
|
||
# Вычисляем схожесть
|
||
similarities = cosine_similarity(preference_vector, self.task_vectors)[0]
|
||
|
||
# Исключаем уже выполненные задания
|
||
if completed_tasks:
|
||
for i, task in enumerate(self.tasks):
|
||
if task.get("title") in completed_tasks or task.get("id") in completed_tasks:
|
||
similarities[i] = -1
|
||
|
||
# Получаем топ-K индексов
|
||
top_indices = np.argsort(similarities)[::-1][:top_k]
|
||
top_indices = [idx for idx in top_indices if similarities[idx] > 0]
|
||
|
||
return [self.tasks[idx] for idx in top_indices]
|
||
|
||
def recommend_by_category(
|
||
self,
|
||
category: str,
|
||
completed_tasks: Optional[List[str]] = None,
|
||
top_k: int = 3,
|
||
) -> List[Dict]:
|
||
"""
|
||
Рекомендовать задания по категории.
|
||
|
||
Args:
|
||
category: Категория заданий
|
||
completed_tasks: Выполненные задания
|
||
top_k: Количество рекомендаций
|
||
|
||
Returns:
|
||
Список рекомендованных заданий
|
||
"""
|
||
category_tasks = [task for task in self.tasks if task.get("category") == category]
|
||
|
||
if completed_tasks:
|
||
category_tasks = [
|
||
task
|
||
for task in category_tasks
|
||
if task.get("title") not in completed_tasks
|
||
and task.get("id") not in completed_tasks
|
||
]
|
||
|
||
# Сортируем по популярности (можно добавить поле rating)
|
||
return category_tasks[:top_k]
|
||
|
||
def get_popular_tasks(self, top_k: int = 10) -> List[Dict]:
|
||
"""
|
||
Получить популярные задания.
|
||
|
||
Args:
|
||
top_k: Количество заданий
|
||
|
||
Returns:
|
||
Список популярных заданий
|
||
"""
|
||
# Простая эвристика: задания, которые чаще выполняются
|
||
task_scores: Dict[str, float] = {}
|
||
|
||
for task in self.tasks:
|
||
task_id = task.get("id") or task.get("title")
|
||
if task.get("completed", False):
|
||
task_scores[task_id] = task_scores.get(task_id, 0) + 1
|
||
|
||
# Сортируем по популярности
|
||
sorted_tasks = sorted(
|
||
self.tasks,
|
||
key=lambda t: task_scores.get(t.get("id") or t.get("title"), 0),
|
||
reverse=True,
|
||
)
|
||
|
||
return sorted_tasks[:top_k]
|
||
|