class RatedItem(RatedItemBase):
    object_id = models.IntegerField()
    content_type = models.ForeignKey(
        ContentType,
        related_name="rated_items",
        on_delete=models.CASCADE,
    )
    content_object = GenericForeignKey()

    @classmethod
    def lookup_kwargs(cls, instance):
        return {"object_id": instance.pk, "content_type": ContentType.objects.get_for_model(instance)}

    @classmethod
    def base_kwargs(cls, model_class):
        return {"content_type": ContentType.objects.get_for_model(model_class)} class Beverage(models.Model):
    name = models.CharField(max_length=50)

    ratings = Ratings(BeverageRating)

    def __str__(self):
        return self.name class BeverageRating(RatedItemBase):
    content_object = models.ForeignKey("Beverage", on_delete=models.CASCADE) class Food(models.Model):
    name = models.CharField(max_length=50)

    ratings = Ratings()

    def __str__(self):
        return self.name def calculate_similar_items(ratings_queryset, num=10):
    # get distinct items from the ratings queryset - this can be optimized
    field = ratings_queryset.model._meta.get_field("content_object")

    if is_gfk(field):
        rated_ctypes = ratings_queryset.values_list("content_type", flat=True).distinct()
        ctypes = ContentType.objects.filter(pk__in=rated_ctypes)
        for ctype in ctypes:
            ratings_subset = ratings_queryset.filter(content_type=ctype)
            rating_ids = ratings_subset.values_list("object_id")
            model_class = ctype.model_class()
            queryset = model_class._default_manager.filter(pk__in=rating_ids)
            _store_top_matches(ratings_queryset, queryset, num, True)
    else:
        rated_model = field.rel.to
        rating_ids = ratings_queryset.values_list("content_object__pk")
        queryset = rated_model._default_manager.filter(pk__in=rating_ids)
        _store_top_matches(ratings_queryset, queryset, num, False) def recommendations(ratings_queryset, people, person, similarity=sim_pearson_correlation):

    already_rated = ratings_queryset.filter(user=person).values_list("hashed")

    totals = {}
    sim_sums = {}

    for other in people:
        if other == person:
            continue

        sim = similarity(ratings_queryset, person, other)

        if sim <= 0:
            continue

        items = ratings_queryset.filter(user=other).exclude(hashed__in=already_rated)

        # now, score the items person hasn't rated yet
        for item in items:
            totals.setdefault(item.content_object, 0)
            totals[item.content_object] += item.score * sim

            sim_sums.setdefault(item.content_object, 0)
            sim_sums[item.content_object] += sim

    rankings = [(total / sim_sums[pk], pk) for pk, total in totals.items()]

    rankings.sort()
    rankings.reverse()
    return rankings def recommended_items(ratings_queryset, user):
    from ratings.models import SimilarItem

    scores = {}
    total_sim = {}

    for item in ratings_queryset.filter(user=user):
        similar_items = SimilarItem.objects.get_for_item(item.content_object)
        for similar_item in similar_items:

            actual = similar_item.similar_object
            lookup_kwargs = ratings_queryset.model.lookup_kwargs(actual)
            lookup_kwargs["user"] = user

            if ratings_queryset.filter(**lookup_kwargs):
                continue

            scores.setdefault(actual, 0)
            scores[actual] += similar_item.score * item.score

            total_sim.setdefault(actual, 0)
            total_sim[actual] += similar_item.score

    rankings = [(score / total_sim[item], item) for item, score in scores.items()]

    rankings.sort()
    rankings.reverse()
    return rankings def sim_euclidean_distance(ratings_queryset, factor_a, factor_b):
    rating_model = ratings_queryset.model

    if isinstance(factor_a, User):
        filter_field = "user_id"
        match_on = "hashed"
        lookup_a = factor_a.pk
        lookup_b = factor_b.pk
    else:
        filter_field = "hashed"
        match_on = "user_id"
        lookup_a = rating_model(content_object=factor_a).generate_hash()
        lookup_b = rating_model(content_object=factor_b).generate_hash()

    sql = """
    SELECT r1.score - r2.score AS diff
    FROM
        %(ratings_table)s AS r1
    INNER JOIN
        %(ratings_table)s AS r2
    ON r1.%(match_on)s = r2.%(match_on)s
    WHERE
        r1.%(filter_field)s = '%(lookup_a)s' AND
        r2.%(filter_field)s = '%(lookup_b)s'
        %(queryset_filter)s
    """

    rating_query = ratings_queryset.values_list("pk").query
    if query_has_where(rating_query):
        queryset_filter = ""
    else:
        q, p = query_as_sql(rating_query)
        rating_qs_sql = q % p
        queryset_filter = " AND r1.id IN (%s)" % rating_qs_sql

    params = {
        "ratings_table": rating_model._meta.db_table,
        "filter_field": filter_field,
        "match_on": match_on,
        "lookup_a": lookup_a,
        "lookup_b": lookup_b,
        "queryset_filter": queryset_filter,
    }

    cursor = connection.cursor()
    cursor.execute(sql % params)

    sum_of_squares = 0
    while True:
        result = cursor.fetchone()
        if result is None:
            break
        sum_of_squares += result[0] ** 2

    return 1 / (1 + sum_of_squares) def sim_pearson_correlation(ratings_queryset, factor_a, factor_b):
    rating_model = ratings_queryset.model

    if isinstance(factor_a, User):
        filter_field = "user_id"
        match_on = "hashed"
        lookup_a = factor_a.pk
        lookup_b = factor_b.pk
    else:
        filter_field = "hashed"
        match_on = "user_id"
        lookup_a = rating_model(content_object=factor_a).generate_hash()
        lookup_b = rating_model(content_object=factor_b).generate_hash()

    sql = """
    SELECT
        SUM(r1.score) AS r1_sum,
        SUM(r2.score) AS r2_sum,
        SUM(r1.score*r1.score) AS r1_square_sum,
        SUM(r2.score*r2.score) AS r2_square_sum,
        SUM(r1.score*r2.score) AS p_sum,
        COUNT(r1.id) AS sample_size
    FROM
        %(ratings_table)s AS r1
    INNER JOIN
        %(ratings_table)s AS r2
    ON r1.%(match_on)s = r2.%(match_on)s
    WHERE
        r1.%(filter_field)s = '%(lookup_a)s' AND
        r2.%(filter_field)s = '%(lookup_b)s'
        %(queryset_filter)s
    """

    rating_query = ratings_queryset.values_list("pk").query
    if query_has_where(rating_query):
        queryset_filter = ""
    else:
        q, p = query_as_sql(rating_query)
        rating_qs_sql = q % p
        queryset_filter = " AND r1.id IN (%s)" % rating_qs_sql

    params = {
        "ratings_table": rating_model._meta.db_table,
        "filter_field": filter_field,
        "match_on": match_on,
        "lookup_a": lookup_a,
        "lookup_b": lookup_b,
        "queryset_filter": queryset_filter,
    }

    cursor = connection.cursor()
    cursor.execute(sql % params)

    result = cursor.fetchone()

    if not result:
        return 0

    sum1, sum2, sum1_sq, sum2_sq, psum, sample_size = result

    if sum1 is None or sum2 is None or sample_size == 0:
        return 0

    num = psum - (sum1 * sum2 / sample_size)
    den = sqrt((sum1_sq - pow(sum1, 2) / sample_size) * (sum2_sq - pow(sum2, 2) / sample_size))

    if den == 0:
        return 0

    return num / den def top_matches(ratings_queryset, items, item, n=5, similarity=sim_pearson_correlation):
    scores = [(similarity(ratings_queryset, item, other), other) for other in items if other != item]
    scores.sort()
    scores.reverse()
    return scores[:n]
import unittest
from django.contrib.auth.models import AnonymousUser, User
from django.contrib.contenttypes.models import ContentType
from django.template import Context, Template
from django.test import TestCase
from django.test.utils import override_settings
from django.urls import reverse
from .. import utils as ratings_utils
from .. import views as ratings_views
from ..models import RatedItem
from ..tests.models import Beverage, BeverageRating, Food
from ..utils import (
    calculate_similar_items,
    recommendations,
    recommended_items,
    sim_euclidean_distance,
    sim_pearson_correlation,
    top_matches,
)
    from django.conf import settings
                self.item1.pk,
                2.5,
            ),
        )
        resp = self.client.post(test_url, {}, HTTP_X_REQUESTED_WITH="XMLHttpRequest")
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp["Content-Type"], "application/json")

        self.assertEqual(self.item1.ratings.cumulative_score(), 2.5)

        self.client.login(username="b", password="b")

        resp = self.client.post(test_url, {}, HTTP_X_REQUESTED_WITH="XMLHttpRequest")
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(self.item1.ratings.cumulative_score(), 5.0)

        resp = self.client.post(test_unrate_url, {}, HTTP_X_REQUESTED_WITH="XMLHttpRequest")
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(self.item1.ratings.cumulative_score(), 2.5)

        # hit it with a negative rating
        test_url = reverse(
            "ratings_rate_object",
            args=(
                ctype.pk,
                self.item1.pk,
                -1.5,
            ),
        )

        resp = self.client.post(test_url, {}, HTTP_X_REQUESTED_WITH="XMLHttpRequest")
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(self.item1.ratings.cumulative_score(), 1.0)

        # finally, test that we can hit it with a GET
        ratings_views.ALLOW_GET = True

        test_url = reverse(
            "ratings_rate_object",
            args=(
                ctype.pk,
                self.item1.pk,
                100.0,
            ),
        )

        # hit the endpoint with a not safe next url
        resp = self.client.get(test_url, {"next": "http://externalurl.com/"})
        self.assertEqual(resp.status_code, 400)

        resp = self.client.get(test_url, {"next": "/"})
        self.assertEqual(resp.status_code, 302)

        self.assertEqual(self.item1.ratings.cumulative_score(), 102.5)

        # if no redirect url is given, redirect to /
        resp = self.client.get(test_url)
        self.assertRedirects(resp, "/", fetch_redirect_response=False)

    def test_rated_item_model_unicode(self):
        self.john.username = "Иван"
        rating = self.item1.ratings.rate(self.john, 1)
        str(rating)


class CustomModelRatingsTestCase(RatingsTestCase):
    rated_model = Beverage
    rating_model = BeverageRating


class RecommendationsTestCase(TestCase):
    fixtures = ["ratings_testdata.json"]

    def setUp(self):
        super(RecommendationsTestCase, self).setUp()

        self.food_a = Food.objects.create(name="food_a")
        self.food_b = Food.objects.create(name="food_b")
        self.food_c = Food.objects.create(name="food_c")
        self.food_d = Food.objects.create(name="food_d")
        self.food_e = Food.objects.create(name="food_e")
        self.food_f = Food.objects.create(name="food_f")

        self.user_a = User.objects.create_user("user_a", "user_a")
        self.user_b = User.objects.create_user("user_b", "user_b")
        self.user_c = User.objects.create_user("user_c", "user_c")
        self.user_d = User.objects.create_user("user_d", "user_d")
        self.user_e = User.objects.create_user("user_e", "user_e")
        self.user_f = User.objects.create_user("user_f", "user_f")
        self.user_g = User.objects.create_user("user_g", "user_g")

        ratings_matrix = [
            # a    b    c    d    e    f
            [2.5, 3.5, 3.0, 3.5, 2.5, 3.0],
            [3.0, 3.5, 1.5, 5.0, 3.5, 3.0],
            [2.5, 3.0, None, 3.5, None, 4.0],
            [None, 3.5, 3.0, 4.0, 2.5, 4.5],
            [3.0, 4.0, 2.0, 3.0, 2.0, 3.0],
            [3.0, 4.0, None, 5.0, 3.5, 3.0],
            [None, 4.5, None, 4.0, 1.0, None],
        ]

        # x-axis
        self.foods = [self.food_a, self.food_b, self.food_c, self.food_d, self.food_e, self.food_f]

        # y-axis
        self.users = [self.user_a, self.user_b, self.user_c, self.user_d, self.user_e, self.user_f, self.user_g]

        for x, food in enumerate(self.foods):
            for y, user in enumerate(self.users):
                if ratings_matrix[y][x]:
                    food.ratings.rate(user, ratings_matrix[y][x])

    def test_simple(self):
        result = sim_euclidean_distance(RatedItem.objects.all(), self.user_a, self.user_b)
        self.assertEqual(str(result)[:5], "0.148")


        result = sim_pearson_correlation(RatedItem.objects.all(), self.user_a, self.user_b)