265 lines
9.2 KiB
Go
265 lines
9.2 KiB
Go
package dao
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/gogf/gf/v2/database/gdb"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"observer-server/biz/consts"
|
||
"observer-server/biz/model/entity"
|
||
"observer-server/common"
|
||
)
|
||
|
||
// AnnotateRecord 用户标注记录表 DAO：pending 即领取锁（partial unique index 同图仅一条），
|
||
// UNIQUE(phone_num, image_id) 一人一图仅一次；拒绝不删记录（低质统计与审计依据）。
|
||
type annotateRecordDao struct{}
|
||
|
||
var AnnotateRecord = &annotateRecordDao{}
|
||
|
||
func init() {
|
||
ctx := context.Background()
|
||
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS annotate_record (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
phone_num VARCHAR(20) NOT NULL,
|
||
task_id BIGINT NOT NULL,
|
||
dataset_id BIGINT NOT NULL DEFAULT 0,
|
||
image_id BIGINT NOT NULL,
|
||
labels_json JSONB,
|
||
status VARCHAR(20) NOT NULL DEFAULT 'pending',
|
||
created_at TIMESTAMP NOT NULL,
|
||
submitted_at TIMESTAMP,
|
||
reviewed_at TIMESTAMP,
|
||
UNIQUE (phone_num, image_id)
|
||
)`)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
// 存量表补列（领取时冗余数据集 id，管理端详情页按数据集过滤记录）
|
||
common.EnsureColumn(ctx, consts.TableAnnotateRecord, "dataset_id", "dataset_id BIGINT NOT NULL DEFAULT 0")
|
||
// 领取锁：同一张图同时只允许一条 pending（partial unique index，SQLite 原生支持）
|
||
if _, err := g.DB().Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_annotate_record_image_pending
|
||
ON annotate_record (image_id) WHERE status = 'pending'`); err != nil {
|
||
panic(err)
|
||
}
|
||
}
|
||
|
||
// InsertInTx 事务内插入领取记录（锁竞争由 partial unique index 兜底，冲突即失败回滚）
|
||
func (d *annotateRecordDao) InsertInTx(ctx context.Context, tx gdb.TX, m *entity.AnnotateRecord) error {
|
||
_, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).TX(tx).Data(g.Map{
|
||
"phone_num": m.PhoneNum,
|
||
"task_id": m.TaskId,
|
||
"dataset_id": m.DatasetId,
|
||
"image_id": m.ImageId,
|
||
"labels_json": common.NilIfEmpty(m.LabelsJson),
|
||
"status": m.Status,
|
||
"created_at": m.CreatedAt,
|
||
}).Insert()
|
||
return err
|
||
}
|
||
|
||
// DeleteExpiredPending 惰性释放过期领取锁（领取时调用；pending 未产生任何标注，直接删除）
|
||
func (d *annotateRecordDao) DeleteExpiredPending(ctx context.Context, before *gtime.Time) error {
|
||
_, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("status", consts.AnnotateRecordPending).
|
||
WhereLT("created_at", before).
|
||
Delete()
|
||
return err
|
||
}
|
||
|
||
// ListImageIdsByPhone 某用户全部记录的图片 id（领取时排除自己处理过的图，含全部状态）
|
||
func (d *annotateRecordDao) ListImageIdsByPhone(ctx context.Context, phone string) ([]int64, error) {
|
||
out, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("phone_num", phone).Fields("image_id").Array()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return gconvInt64Slice(out), nil
|
||
}
|
||
|
||
// ListPendingImageIds 当前全部领取锁的图片 id（领取时排除他人锁定的图）
|
||
func (d *annotateRecordDao) ListPendingImageIds(ctx context.Context) ([]int64, error) {
|
||
out, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("status", consts.AnnotateRecordPending).Fields("image_id").Array()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return gconvInt64Slice(out), nil
|
||
}
|
||
|
||
// ListPendingImageIdsByTask 某任务当前领取锁的图片 id（停用任务释放时保留这些图——
|
||
// 已领取未提交的可继续提交；过期锁由调用方先经 DeleteExpiredPending 惰性清理）
|
||
func (d *annotateRecordDao) ListPendingImageIdsByTask(ctx context.Context, taskId int64) ([]int64, error) {
|
||
out, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("task_id", taskId).
|
||
Where("status", consts.AnnotateRecordPending).Fields("image_id").Array()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return gconvInt64Slice(out), nil
|
||
}
|
||
|
||
// GetMyPending 某用户在某图上的领取记录（提交校验：必须存在且为 pending）
|
||
func (d *annotateRecordDao) GetMyPending(ctx context.Context, phone string, imageId int64) (*entity.AnnotateRecord, error) {
|
||
var one *entity.AnnotateRecord
|
||
err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("phone_num", phone).Where("image_id", imageId).
|
||
Where("status", consts.AnnotateRecordPending).Scan(&one)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return one, nil
|
||
}
|
||
|
||
// MarkSubmitted 提交：写快照 + 置 submitted
|
||
func (d *annotateRecordDao) MarkSubmitted(ctx context.Context, id int64, labelsJson string, submittedAt *gtime.Time) error {
|
||
_, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).Where("id", id).
|
||
Data(g.Map{"labels_json": labelsJson, "status": consts.AnnotateRecordSubmitted, "submitted_at": submittedAt}).
|
||
Update()
|
||
return err
|
||
}
|
||
|
||
// MarkReviewedByImages 按图片批量审核当前提交中的记录（通过→approved / 拒绝→rejected）：
|
||
// 只命中 status=submitted 的记录（历史已审核记录不动）
|
||
func (d *annotateRecordDao) MarkReviewedByImages(ctx context.Context, imageIds []int64, status string, reviewedAt *gtime.Time) error {
|
||
if len(imageIds) == 0 {
|
||
return nil
|
||
}
|
||
_, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
WhereIn("image_id", imageIds).
|
||
Where("status", consts.AnnotateRecordSubmitted).
|
||
Data(g.Map{"status": status, "reviewed_at": reviewedAt}).
|
||
Update()
|
||
return err
|
||
}
|
||
|
||
// ListSubmittedPhonesByImages 被审核图片对应的提交用户（审核后重算通过比例用，去重）
|
||
func (d *annotateRecordDao) ListSubmittedPhonesByImages(ctx context.Context, imageIds []int64) ([]string, error) {
|
||
if len(imageIds) == 0 {
|
||
return []string{}, nil
|
||
}
|
||
out, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
WhereIn("image_id", imageIds).
|
||
Where("status", consts.AnnotateRecordSubmitted).
|
||
Fields("DISTINCT phone_num").Array()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
phones := make([]string, 0, len(out))
|
||
for _, v := range out {
|
||
if s := v.String(); s != "" {
|
||
phones = append(phones, s)
|
||
}
|
||
}
|
||
return phones, nil
|
||
}
|
||
|
||
// StatusCount 按 task_id × status 聚合的任务记录数（任务列表进度展示）
|
||
type StatusCount struct {
|
||
TaskId int64 `orm:"task_id"`
|
||
Status string `orm:"status"`
|
||
Cnt int64 `orm:"cnt"`
|
||
}
|
||
|
||
// CountByTaskIds 多任务各状态记录数（一次 GROUP BY；IN 按 ≤100 分批由调用方保证）
|
||
func (d *annotateRecordDao) CountByTaskIds(ctx context.Context, taskIds []int64) (map[int64]map[string]int64, error) {
|
||
out := make(map[int64]map[string]int64)
|
||
if len(taskIds) == 0 {
|
||
return out, nil
|
||
}
|
||
var rows []StatusCount
|
||
err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
WhereIn("task_id", taskIds).
|
||
Fields("task_id, status, COUNT(*) AS cnt").
|
||
Group("task_id, status").Scan(&rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for _, r := range rows {
|
||
if out[r.TaskId] == nil {
|
||
out[r.TaskId] = make(map[string]int64)
|
||
}
|
||
out[r.TaskId][r.Status] = r.Cnt
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// UserStats 用户统计：累计提交数 + 审核通过/拒绝数（reviewedSince 非空时只统计其后的审核——
|
||
// 冻结重置基线后「重新累计」语义）
|
||
type UserStats struct {
|
||
SubmittedTotal int64
|
||
Approved int64
|
||
Rejected int64
|
||
}
|
||
|
||
// CountUserStats 用户统计查询
|
||
func (d *annotateRecordDao) CountUserStats(ctx context.Context, phone string, reviewedSince *gtime.Time) (*UserStats, error) {
|
||
stats := &UserStats{}
|
||
// 累计提交 = 已产生标注的全部记录（submitted/approved/rejected；pending 是未处理的锁不算）
|
||
n, err := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).
|
||
Where("phone_num", phone).
|
||
WhereIn("status", []string{consts.AnnotateRecordSubmitted, consts.AnnotateRecordApproved, consts.AnnotateRecordRejected}).
|
||
Count()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
stats.SubmittedTotal = int64(n)
|
||
base := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx).Where("phone_num", phone)
|
||
if reviewedSince != nil {
|
||
base = base.WhereGTE("reviewed_at", reviewedSince)
|
||
}
|
||
approved, err := base.Clone().Where("status", consts.AnnotateRecordApproved).Count()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
rejected, err := base.Where("status", consts.AnnotateRecordRejected).Count()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
stats.Approved = int64(approved)
|
||
stats.Rejected = int64(rejected)
|
||
return stats, nil
|
||
}
|
||
|
||
// PageByFilter 管理端记录分页（datasetId>0/phone 精确/taskId/status 过滤，id 倒序）
|
||
func (d *annotateRecordDao) PageByFilter(ctx context.Context, phone string, taskId, datasetId int64, status string, page, size int) ([]*entity.AnnotateRecord, int64, error) {
|
||
base := func() *gdb.Model {
|
||
m := g.DB().Model(consts.TableAnnotateRecord).Ctx(ctx)
|
||
if phone != "" {
|
||
m = m.Where("phone_num", phone)
|
||
}
|
||
if taskId > 0 {
|
||
m = m.Where("task_id", taskId)
|
||
}
|
||
if datasetId > 0 {
|
||
m = m.Where("dataset_id", datasetId)
|
||
}
|
||
if status != "" {
|
||
m = m.Where("status", status)
|
||
}
|
||
return m
|
||
}
|
||
total, err := base().Count()
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
var list []*entity.AnnotateRecord
|
||
err = base().OrderDesc("id").Limit((page-1)*size, size).Scan(&list)
|
||
if err != nil {
|
||
if common.IsNoRows(err) {
|
||
return []*entity.AnnotateRecord{}, int64(total), nil
|
||
}
|
||
return nil, 0, err
|
||
}
|
||
return list, int64(total), nil
|
||
}
|
||
|
||
func gconvInt64Slice(vals gdb.Array) []int64 {
|
||
out := make([]int64, 0, len(vals))
|
||
for _, v := range vals {
|
||
out = append(out, v.Int64())
|
||
}
|
||
return out
|
||
}
|