在 Django 中构建动态任务和徽章评估系统

模型

任务

class Quest(models.Model):
    name = models.CharField(max_length=255)
    description = models.TextField(blank=True)
    criteria = models.JSONField()  # Store criteria as JSON
    reward_points = models.IntegerField(default=0)

    def __str__(self):
        return self.name

徽章

class Badge(models.Model):
    name = models.CharField(max_length=255)
    description = models.TextField(blank=True)
    criteria = models.JSONField()  # Store criteria as JSON
    reward_points = models.IntegerField(default=0)

    def __str__(self):
        return self.name

UserQuestProgress

class UserQuestProgress(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    quest = models.ForeignKey(Quest, on_delete=models.CASCADE)
    current_value = models.IntegerField(default=0)
    target_value = models.IntegerField()
    completed = models.BooleanField(default=False)
    completed_at = models.DateTimeField(null=True, blank=True)

    def save(self, *args, **kwargs):
        if self.current_value >= self.target_value:
            self.completed = True
            if not self.completed_at:
                self.completed_at = timezone.now()
        super().save(*args, **kwargs)

UserBadgeProgress

class UserBadgeProgress(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    badge = models.ForeignKey(Badge, on_delete=models.CASCADE)
    current_value = models.IntegerField(default=0)
    target_value = models.IntegerField()
    completed = models.BooleanField(default=False)
    completed_at = models.DateTimeField(null=True, blank=True)

    def save(self, *args, **kwargs):
        if self.current_value >= self.target_value:
            self.completed = True
            if not self.completed_at:
                self.completed_at = timezone.now()
        super().save(*args, **kwargs)

我正在开发一个社交媒体平台，用户可以根据他们的活动获得奖励（任务和徽章）。我的目标是创建一个灵活的系统，可以动态评估这些奖励，而无需每次引入新任务或徽章时都硬编码新条件。

到目前为止我们所做的：

1. 任务和徽章模型| ||：我们为 、 Quest 、 Badge 和 UserQuestProgress 定义了模型，用于跟踪用户的进度。 UserBadgeProgress 标准 JSON
2. ：我们使用 JSON 来定义每个任务或任务的标准徽章。例如： 评估函数

   {
       "Post": {
           "reaction_type_within_timeframe": ["dislike", 30, 50]
       }
   }
   

3. ：我们有一个 处理标准 JSON 来计算用户进度的函数： evaluate_quest Celery 任务

   def evaluate_quest(user, criteria):
       # Evaluate criteria and calculate progress
   

4. ：当用户导航到任务或徽章页面时，触发 Celery 任务来评估其进度并更新 或 UserQuestProgress UserBadgeProgress 当前挑战：

硬编码场景

• ：当前的实现仍然需要我们在 函数中硬编码新场景。我们的目标是动态处理各种复杂的情况，但正在努力应对未知的未来场景。 evaluate_quest 示例场景
• ： 首次发布状态更新或任何其他帖子
  • ： 您已完成您的所有个人资料信息字段

    {
        "Post": {
            "first_post": true
        }
    }
    

  • : 对您关注者的帖子给出 50 条喜欢和/或喜爱的反应

    {
        "Profile": {
            "complete_profile": true
        }
    }
    

  • : 单个帖子收到 100 多个喜欢和 50 多条评论

    {
        "Reaction": {
            "reaction_type": ["like", "love"],
            "count": 50,
            "on_followers_posts": true
        }
    }
    

  • :|| |用户首先对 50 多个朋友的帖子做出反应 ：

    {
        "Post": {
            "likes": 100,
            "comments": 50,
            "single_post": true
        }
    }
    

  • 工作流程： 当用户导航到任务或徽章页面时，Celery 任务启动。

    {
        "Reaction": {
            "first_reaction": true,
            "on_friends_posts": true,
            "count": 50
        }
    }
    

该任务使用

1. 来确保有一个进度实例。
2. get_or_create(user, quest|badge) 函数被调用来计算当前值和目标值。
3. evaluate_quest 和
4. 模型已覆盖 UserQuestProgress 如果当前值满足则标记完成的方法或超过目标值。 UserBadgeProgress 保存后，如果任务或徽章已完成， save 信号会通过 WebSocket 通知用户。
5. 当前 post_save 功能：

关注点： evaluate_quest 当前实现仍然需要我们在

def evaluate_quest(user, criteria):
    criteria = json.loads(criteria)
    progress = 0
    total_conditions = 0

    for model_name, conditions in criteria.items():
        Model = apps.get_model('your_app_name', model_name)  # Dynamically get the model
        queryset = Model.objects.filter(user=user)

        for field, condition in conditions.items():
            if field == 'time_range':
                start_time, end_time, target = condition
                queryset = queryset.filter(
                    created_at__time__gte=start_time,
                    created_at__time__lte=end_time
                )
                current_value = queryset.count()

            elif field == 'reaction_type_within_timeframe':
                reaction_type, timeframe, target = condition
                time_delta = timedelta(minutes=timeframe)
                queryset = queryset.annotate(
                    reactions_within_timeframe=Count(
                        Case(
                            When(
                                reactions__reaction_type=reaction_type,
                                reactions__created_at__lte=ExpressionWrapper(
                                    F('created_at') + time_delta,
                                    output_field=DurationField()
                                ),
                                then='reactions'
                            )
                        )
                    )
                )
                current_value = queryset.filter(reactions_within_timeframe__gte=target).count()

            elif field == 'consecutive_days':
                target_days = condition
                dates = queryset.values_list('created_at', flat=True)
                dates = sorted(dates)
                current_streak = 0
                max_streak = 0

                for i in range(1, len(dates)):
                    if (dates[i] - dates[i - 1]).days == 1:
                        current_streak += 1
                    else:
                        max_streak = max(max_streak, current_streak)
                        current_streak = 0

                max_streak = max(max_streak, current_streak)
                current_value = max_streak
                target = target_days

            elif field == 'linked_social_networks':
                target_value = condition
                social_fields = [
                    'facebook_username', 'twitter_username', 'instagram_username',
                    'twitch_username', 'google_username', 'youtube_username',
                    'patreon_username', 'discord_username', 'deviantart_username',
                    'behance_username', 'dribble_username', 'artstation_username'
                ]
                current_value = sum(bool(getattr(queryset.first(), field)) for field in social_fields)
                target = target_value

            else:
                value, target = condition
                if isinstance(value, str):
                    queryset = queryset.filter(**{field: value})
                elif isinstance(value, dict) and 'range' in value:
                    start, end = value['range']
                    queryset = queryset.filter(**{f"{field}__gte": start, f"{field}__lte": end})
                current_value = queryset.count()

            # Calculate progress for this condition
            if target > 0:
                progress += current_value / target
            total_conditions += 1

    # Average progress across all conditions
    if total_conditions > 0:
        progress = progress / total_conditions

    return progress

函数中对特定条件进行硬编码。

• 我们需要一种更加动态的方法，可以处理未知的未来场景，而不需要更改代码。 evaluate_quest 请求帮助：
• 我们怎样才能使| ||函数更加动态地处理各种复杂的条件，而无需对每个新场景进行硬编码？

Django 中是否有任何现有的库或模式可以帮助实现这种灵活性？

• 我们需要处理的复杂条件的示例：|| |首次发布状态更新。 evaluate_quest 填写所有个人资料信息字段。
• 在关注者的帖子上给予 50 个点赞和/或喜爱反应。
• 收到 100 多个点赞和 50 多个
  • 评论单个帖子。
  • 首先对 50 多个朋友的帖子做出反应。
  • 当前工作流程：
  • 用户导航到任务或徽章页面，触发 Celery 任务。

该任务评估所有任务或徽章用户使用

调用

1. 函数来计算进度。
2. 进度根据 get_or_create(user, quest|badge) 函数的响应进行更新。
3. evaluate_quest 中的
4. 方法。如果当前值达到或超过目标值， evaluate_quest 自动标记完成。
5. A save 如果任务或徽章已完成，则信号通过 WebSocket 通知用户。 UserQuestProgress 制作此系统的任何见解或建议更加强大和动态将不胜感激。谢谢你！ UserBadgeProgress automatically marks completion if the current value meets or exceeds the target value.
6. A post_save signal notifies the user via WebSocket if a quest or badge is completed.

Any insights or suggestions to make this system more robust and dynamic would be greatly appreciated. Thank you!

好的，以下是构建动态任务和徽章评估系统的方法，该系统解决了的担忧并最大程度地减少了 Django 中的硬编码逻辑：

1. 基于规则引擎的标准

与其直接在 evaluate_quest 函数中对条件进行硬编码，不如使用基于规则的引擎来实现更大的灵活性。可以使用 Python 库（如 rules ）或开发自己的简单引擎。

• 定义规则： 为常见场景创建可重用的规则函数。

```python from rules import predicate, Rule

@predicate def has_posted(user): return user.posts.exists()

@predicate def profile_complete(user): # 检查个人资料字段 pass

@predicate def reaction_count_on_followers_posts(user, reaction_type, min_count): return user.reactions_given.filter( reaction_type=reaction_type, post__author__in=user.following.all() ).count() >= min_count

# ... 为其他条件添加更多规则

# 定义规则 first_post_rule = Rule(has_posted) complete_profile_rule = Rule(profile_complete) likes_on_followers_posts_rule = Rule(reaction_count_on_followers_posts) # ... 更多规则 ```

• 将规则存储在标准中： 在 criteria JSON 字段中存储规则名称和必要参数。

json { "rules": [ {"name": "first_post"}, {"name": "complete_profile"}, {"name": "reaction_count_on_followers_posts", "args": ["like", 50]} ] }

• 评估规则： 修改 evaluate_quest 以动态获取和应用规则。

```python def evaluate_quest(user, criteria): criteria = json.loads(criteria) progress = 0 total_rules = len(criteria['rules'])

   for rule_data in criteria['rules']:
       rule_name = rule_data['name']
       args = rule_data.get('args', [])

       # 通过名称动态获取规则函数
       rule_function = globals().get(rule_name + '_rule') 
       if rule_function:
           if rule_function.test(user, *args): 
               progress += 1

   return progress / total_rules if total_rules else 0

```

2. 自定义管理器或查询集

为常用操作创建自定义管理器或查询集，以使代码更清晰、更可重用。

```python class PostQuerySet(models.QuerySet): def within_timeframe(self, timeframe): return self.filter(created_at__gte=timezone.now() - timedelta(minutes=timeframe))

   def with_reaction_type(self, reaction_type):
       return self.filter(reactions__reaction_type=reaction_type)

class PostManager(models.Manager): def get_queryset(self): return PostQuerySet(self.model, using=self._db)

   # ... 添加更多方法以简化常见查询

class Post(models.Model): # ... objects = PostManager() ```

3. 信号和任务队列

• 信号： 继续使用信号（例如 post_save 、 m2m_changed ）来检测触发任务或徽章评估的用户操作。这可以防止对数据库进行不必要的查询。
• 任务队列（Celery）： 使用 Celery 等任务队列在后台异步处理评估。这将提高性能并提供更好的用户体验。

4. 改进的标准结构

考虑使用更结构化的格式来定义标准，以实现更好的可扩展性。例如，可以使用类似 JSON Schema 的方法来定义可用的条件、它们的预期数据类型以及对 evaluate_quest 函数的任何验证。

示例场景实现

让我们看看如何使用基于规则的方法实现“对关注者的帖子给予 50 个点赞”场景：

# rules.py 或类似文件
@predicate
def reaction_count_on_followers_posts(user, reaction_type, min_count):
   return user.reactions_given.filter(
       reaction_type=reaction_type,
       post__author__in=user.following.all()
   ).count() >= min_count

likes_on_followers_posts_rule = Rule(reaction_count_on_followers_posts)

# models.py 中的任务/徽章标准
{
   "rules": [
       {"name": "reaction_count_on_followers_posts", "args": ["like", 50]} 
   ]
}

# evaluate_quest 函数
# ...（如前所述）

通过此设置，可以在不修改代码库的情况下轻松定义新规则和条件，从而使的任务和徽章系统更加灵活和可维护。

其他提示

• 彻底测试： 为规则评估逻辑实施全面的单元测试，以确保在添加新功能时系统的正确性。
• 缓存： 考虑缓存任务和徽章进度，尤其是当计算成本很高时。
• 管理员界面： 如果需要，请创建一个用户友好的管理员界面，以供非技术用户定义新任务、徽章及其标准。

通过遵循这些原则，可以构建一个健壮、灵活和可扩展的 Django 任务和徽章评估系统，该系统可以适应不断变化的需求，而无需持续的硬编码修改。