我已经创建了一个考勤服务器，允许员工记录他们的出勤。有手动和自动记录出勤的方式。系统的SQL表如下所示。
| 工作人员|本地用户|设备|生物识别日志|打孔日志|
| - -----|- -----|- -----|- -----|- -----|
| 身份证|身份证|身份证|身份证|身份证|
| 姓名|姓名|位置|本地用户ID|员工ID|
| | 设备ID| API密钥|打孔|打孔|
| | 员工ID|||已禁用|
每当添加生物特征日志时，如果该日志中的本地用户具有人员外键，则会创建打孔日志。我觉得这有助于结合自动和手动日志。手动日志直接添加到打孔日志
当用户请求出席时。目前，这将获取活动的打孔日志的替代行，并将其合并为一行。

输出数据格式

| 考勤记录|
| - -----|
| 身份证|
| 打卡|
| 打卡|
| 员工ID|

使用代码

目前我使用原始的sql查询来获取数据
基本逻辑是

• 拆分为两个子查询->使用用户唯一的卷号打卡和打卡
• 使用卷号连接

select i.id, username, i.punched_on as punch_in,  min(punch_out) as punch_out
    from (select *,row_number() over (partition by user_id order by user_id) as r_index from attendance_punch_log) as i
    left join local_user on i.user_id = local_user.id
    left join (
    select  o.logged_on as punch_out, username as out_user
    from (select *, row_number() over (partition by user_id order by user_id) as l_index from attendance_punch_log) as o
    left join local_user on o.user_id = local_user.id) as o on username = out_user and punch_out > punch_in
    where mod(r_index, 2) = 1
    group by username , r_index
    order by punch_in, punch_out

Django模型如下

class BiometricsDevice(models.Model):
    location = models.ForeignKey(
        Location, on_delete=models.CASCADE, related_name="device"
    )
    description = models.CharField(max_length=120)
    api_key = models.CharField(max_length=20, null=False, blank=False, unique=True)
    last_sync = models.DateTimeField(null=True)
    sync_user = models.BooleanField(default=True)
    is_active = models.BooleanField(default=True)

    created_on = models.DateField(auto_now_add=True)
    created_by = models.ForeignKey(
        User,
        on_delete=models.CASCADE,
        related_name="created_biometrics",
    )

    def __str__(self):
        return f"{self.location.name} - {self.api_key}"

class LocalUser(models.Model):
    ref = models.PositiveIntegerField()
    name = models.CharField(max_length=20)
    device = models.ForeignKey(
        BiometricsDevice, on_delete=models.CASCADE, related_name="users"
    )
    # Signal on staff update -> Re-Run Biometrics Log to Punch Log Conversion
    staff = models.ForeignKey(
        Staff,
        on_delete=models.CASCADE,
        related_name="biometric_user",
        blank=True,
        null=True,
    )

    def __str__(self):
        return f"{self.name} - {self.device}"

    class Meta:
        unique_together = (["name", "device"], ["name", "ref", "device"])

class PunchLog(models.Model):
    staff = models.ForeignKey(
        Staff,
        on_delete=models.CASCADE,
        related_name="punch_log",
    )
    punched_on = models.DateTimeField()
    is_active = models.BooleanField(default=True)

    created_on = models.DateField(auto_now_add=True)
    created_by = models.ForeignKey(
        User,
        on_delete=models.CASCADE,
        related_name="created_punch_logs",
        null=True,
        blank=True,
    )

    def __str__(self):
        return f"{self.staff.name} - {self.punched_on}"

class BiometricsLog(models.Model):
    user = models.ForeignKey(
        LocalUser,
        on_delete=models.CASCADE,
        related_name="biometrics_log",
    )
    punched_on = models.DateTimeField()

    punch_log = models.OneToOneField(
        PunchLog,
        on_delete=models.CASCADE,
        related_name="biometrics_log",
        null=True,
        blank=True,
    )

    class Meta:
        unique_together = ["user", "punched_on"]

    def __str__(self):
        return f"{self.user.name} - {self.punched_on}"

创建查询集的代码如下

def get(self, request):
        punch_in = (
            BiometricsLog.objects.annotate(
                index=Window(
                    expression=RowNumber(),
                    order_by=F("user_id").asc(),
                    partition_by=[F("user_id")],
                ),
            )
            .annotate(evenindex=F("index") % 2)
            .filter(evenindex=1)
            .order_by("-punched_on")
        )
        punch_out = (
            BiometricsLog.objects.annotate(
                index=Window(
                    expression=RowNumber(),
                    order_by=F("user_id").asc(),
                    partition_by=[F("user_id")],
                ),
            )
            .annotate(evenindex=F("index") % 2)
            .filter(evenindex=0)
            .order_by("-punched_on")
        )
        # Odd Even index join

        queryset = punch_in.annotate(
            punch_in=F("punched_on"),
            punch_out=Subquery(
                punch_out.filter(
                    index__gt=OuterRef("index"),
                    user_id=OuterRef("user_id"),
                ).values("punched_on")[:1]
            ),
            staff=F("user__staff"),
        )

总结

• 最后我应该用什么方法来创建考勤表？
• 创建一个新表而不是查询集是否更好？