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




def skipUnlessDB(engine):
    """
    This decorator makes a test skip unless the current connection uses the
    specified engine.
    """

    actual_engine = settings.DATABASES["default"]["ENGINE"]
    if engine not in actual_engine:
        return unittest.skip("Unsupported connection engine: %s (expected %s)" % (actual_engine, engine))
    return lambda func: func


@override_settings(
    MIDDLEWARE=(
        "django.contrib.sessions.middleware.SessionMiddleware",
        "django.contrib.auth.middleware.AuthenticationMiddleware",
        "django.middleware.locale.LocaleMiddleware",
        "django.middleware.common.CommonMiddleware",
    ),
    ROOT_URLCONF="ratings.tests.urls",
)
class RatingsTestCase(TestCase):
    fixtures = ["ratings_testdata.json"]


    rated_model = Food