如何在Django查询中检测Message关联Annotation的标签集冲突？

需求背景 #

需要为每个Message生成查询，标注其关联的已审核Annotation是否存在冲突。冲突判定条件：

• 所有已审核Annotation的text不一致，或urgency不一致
• 或label_set（标签的name与topic组合集合）不一致

模型定义 #

class Message(models.Model):
    text = models.CharField()

class Annotation(models.Model):
    message = models.ForeignKey(to=Message)
    user = models.ForeignKey(to=User)
    text = models.CharField()
    reviewed = models.BooleanField()
    urgency = models.ForeignKey(to=Urgency)

class Label(models.Model):
   annotation = models.ForeignKey(to=Annotation)  # 注：原代码外键关联错误，修正为关联Annotation
   topic = models.ForeignKey(to=Topic)
   name = models.CharField()

注：原Label模型外键应关联Annotation而非Message，否则无法对应到具体标注的标签集合，此处先修正该逻辑错误。

现有代码（已实现text和urgency的冲突检测） #

messages = Message.objects.filter([arbitrary filter]).annotate(
    status=Case(
        When(
            GreaterThan(
                Count('annotation__urgency', filter=Q(annotation__reviewed=True), distinct=True), Value(1)
            ) | 
            GreaterThan(
                Count('annotation__text', filter=Q(annotation__reviewed=True), distinct=True), Value(1)
            ),
            then=Value('Conflict')
        ),
        default=Value('Multiple')
    )
)

补充标签集冲突检测的实现方法 #

要检测label_set冲突，核心是判断同一Message下的已审核Annotation，它们的(name, topic)标签组合集合是否存在差异，以下是两种可行实现：

方法1：基于字符串聚合的子查询方案 #

通过为每个Annotation生成唯一的标签组合标识，统计同一Message下不同标识的数量，判断是否冲突：

from django.db.models import (
    Case, When, Value, Count, Q, Func, F, StringAgg
)

# 为每个已审核Annotation生成标签组合的唯一键（按topic_id和name排序，保证相同组合的键一致）
annotation_label_keys = Annotation.objects.filter(
    reviewed=True
).annotate(
    label_key=StringAgg(
        Func(F('label__topic__id'), Value('_'), F('label__name'), function='CONCAT'),
        separator='|',
        ordering=('label__topic__id', 'label__name')
    )
).values('message_id', 'label_key').distinct()

# 统计每个Message下不同标签组合集合的数量
message_label_conflict = annotation_label_keys.values('message_id').annotate(
    distinct_label_sets=Count('label_key', distinct=True)
)

# 整合到原查询中
messages = Message.objects.filter([arbitrary filter]).annotate(
    # 原text和urgency冲突检测
    text_urgency_conflict=Case(
        When(
            GreaterThan(Count('annotation__urgency', filter=Q(annotation__reviewed=True), distinct=True), 1) |
            GreaterThan(Count('annotation__text', filter=Q(annotation__reviewed=True), distinct=True), 1),
            then=Value(True),
            default=Value(False)
        )
    ),
    # 标签集冲突检测
    label_set_conflict=Case(
        When(
            message_id__in=message_label_conflict.filter(distinct_label_sets__gt=1).values('message_id'),
            then=Value(True),
            default=Value(False)
        )
    ),
    # 最终状态判断
    status=Case(
        When(Q(text_urgency_conflict=True) | Q(label_set_conflict=True), then=Value('Conflict')),
        default=Value('Multiple')
    )
)

方法2：基于数组聚合的原生SQL方案（适用于PostgreSQL） #

利用PostgreSQL的数组聚合功能，直接比较每个Annotation的标签组合数组是否一致：

from django.db.models import Func, Value, Subquery

class ArrayAggLabels(Func):
    function = 'ARRAY_AGG'
    template = "%(function)s(DISTINCT CONCAT(%(expressions)s)) ORDER BY CONCAT(%(expressions)s)"

messages = Message.objects.filter([arbitrary filter]).annotate(
    # 原text和urgency冲突检测
    text_urgency_conflict=Case(
        When(
            GreaterThan(Count('annotation__urgency', filter=Q(annotation__reviewed=True), distinct=True), 1) |
            GreaterThan(Count('annotation__text', filter=Q(annotation__reviewed=True), distinct=True), 1),
            then=Value(True),
            default=Value(False)
        )
    ),
    # 标签集冲突检测：对比当前Message下所有Annotation的标签数组是否与第一个一致
    label_set_conflict=Case(
        When(
            ArrayAggLabels(F('annotation__label__topic__id'), Value('_'), F('annotation__label__name'), filter=Q(annotation__reviewed=True)) !=
            ArrayAggLabels(F('annotation__label__topic__id'), Value('_'), F('annotation__label__name'), filter=Q(annotation__reviewed=True) & Q(annotation__id=Subquery(
                Annotation.objects.filter(message=F('pk'), reviewed=True).values('id')[:1]
            ))),
            then=Value(True),
            default=Value(False)
        )
    ),
    # 最终状态判断
    status=Case(
        When(Q(text_urgency_conflict=True) | Q(label_set_conflict=True), then=Value('Conflict')),
        default=Value('Multiple')
    )
)

关键注意点 #

• 必须修正Label模型的外键关联，否则无法正确关联到具体Annotation的标签集合。
• 生成标签组合标识时必须排序，避免因标签顺序不同导致误判冲突。
• 不同数据库对聚合函数的支持不同，StringAgg在MySQL、PostgreSQL中可用，SQLite需额外配置扩展。

内容的提问来源于stack exchange，提问作者itdub