starpivot/apps/services/modules/api/model_rebuildstats.go

package api

import (
	"yunyan/comm"
	"yunyan/lego/sys/mysql"
	"yunyan/pb"
	"fmt"
	"time"
)

// rebuildUserStatisticsFromLogs 从 useruselog + echomeet_record 全量重建 userstatistics。
//
// Phase 1 — useruselog（logtype=UserConsume）：
//   - addagentintegral < 0 → 翻译/AI 消耗，合并到 trademodel1_time/trademodel1_num
//   - addmeetsecond    < 0 → 实时会议（UsageType_Meet 路径）消耗
//
// Phase 2 — echomeet_record：
//
//	echomeet 启动任务时直接扣 meetintegral，未写 useruselog；
//	state > 0 且 != TranscribeFail(10002) 的记录表示积分已扣且未退款。
//	使用 ON DUPLICATE KEY UPDATE 累加到 Phase 1 已写入的行。
//
// 步骤：先 TRUNCATE，再两阶段 INSERT，保证幂等。
func (this *modelComp) rebuildUserStatisticsFromLogs() (rowsInserted int64, err error) {
	if err = mysql.Exec(fmt.Sprintf("TRUNCATE TABLE %s", comm.TableUserStatistics)).Error; err != nil {
		return
	}
	// Phase 1: 从 useruselog 聚合翻译消耗 + 实时会议消耗
	sql1 := fmt.Sprintf(
		`INSERT INTO %s (uid, meetnum, meettime, trademodel1_num, trademodel1_time)
		 SELECT
		     uid,
		     SUM(CASE WHEN addmeetsecond    < 0 THEN 1 ELSE 0 END),
		     COALESCE(SUM(CASE WHEN addmeetsecond    < 0 THEN -addmeetsecond    ELSE 0 END), 0),
		     SUM(CASE WHEN addagentintegral < 0 THEN 1 ELSE 0 END),
		     COALESCE(SUM(CASE WHEN addagentintegral < 0 THEN -addagentintegral ELSE 0 END), 0)
		 FROM %s
		 WHERE logtype = ?
		 GROUP BY uid`,
		comm.TableUserStatistics, comm.TableUserUseLog,
	)
	tx1 := mysql.Exec(sql1, int32(pb.UserLogType_UserConsume))
	if tx1.Error != nil {
		err = tx1.Error
		return
	}
	rowsInserted += tx1.RowsAffected

	// Phase 2: 从 echomeet_record 聚合文件转写会议消耗
	// state > 0: 任务已启动（积分已扣）
	// state != 10002 (TranscribeFail): 短音频失败时积分会退款，排除
	sql2 := fmt.Sprintf(
		`INSERT INTO %s (uid, meetnum, meettime)
		 SELECT uid, COUNT(*) AS meetnum, COALESCE(SUM(seconds), 0) AS meettime
		 FROM %s
		 WHERE state > 0 AND state != ? AND seconds > 0
		 GROUP BY uid
		 ON DUPLICATE KEY UPDATE
		     meetnum  = meetnum  + VALUES(meetnum),
		     meettime = meettime + VALUES(meettime)`,
		comm.TableUserStatistics, comm.TableEchomeetRecord,
	)
	tx2 := mysql.Exec(sql2, int32(pb.DBEchoMeetRecordState_TranscribeFail))
	if tx2.Error != nil {
		err = tx2.Error
		return
	}
	rowsInserted += tx2.RowsAffected
	return
}

// 产品激活/绑定统计重建 ----------------------------------------------------------

// productDeviceAgg userdevice 按 productid 聚合的中间结果
type productDeviceAgg struct {
	Productid uint32 `gorm:"column:productid"`
	Activated int64  `gorm:"column:activated"`
	Bound     int64  `gorm:"column:bound"`
}

// aggUserDeviceByProduct 按 productid 聚合 userdevice：
//
//	activated = 该产品的设备绑定记录行数（激活设备数）
//	bound     = 该产品去重后的非空 uid 数（绑定用户数）
//
// NULLIF(uid,”) 把空串转 NULL，使 COUNT(DISTINCT) 自动忽略未绑定用户的记录。
func (this *modelComp) aggUserDeviceByProduct() (rows []*productDeviceAgg, err error) {
	rows = make([]*productDeviceAgg, 0)
	err = mysql.Table(comm.TableUserdevice).
		Select("productid, COUNT(*) AS activated, COUNT(DISTINCT NULLIF(uid,'')) AS bound").
		Group("productid").
		Scan(&rows).Error
	return
}

// rebuildProductStats 全量重建 product_stat：
// 扫描整张 userdevice 表汇总每个产品的激活设备数 / 绑定用户数，
// 并为 product 表中尚无设备的产品补 0 行，保证“所有产品”都有一条统计；
// 最后 TRUNCATE 后整表改写。返回写入的行数。
func (this *modelComp) rebuildProductStats() (rows int, err error) {
	agg, err := this.aggUserDeviceByProduct()
	if err != nil {
		return
	}
	now := time.Now().Unix()
	statMap := make(map[uint32]*pb.DBProductStat, len(agg))
	for _, a := range agg {
		if a.Productid == 0 { // userdevice 中无产品归属的脏数据，仪表盘不会查询，跳过
			continue
		}
		statMap[a.Productid] = &pb.DBProductStat{
			Productid:      a.Productid,
			ActivatedCount: a.Activated,
			BoundUserCount: a.Bound,
			UpdateTime:     now,
		}
	}
	// product 表里尚无设备的产品补 0 行
	products, perr := this.products()
	if perr != nil {
		err = perr
		return
	}
	for _, p := range products {
		if _, ok := statMap[p.Id]; !ok {
			statMap[p.Id] = &pb.DBProductStat{Productid: p.Id, UpdateTime: now}
		}
	}
	list := make([]*pb.DBProductStat, 0, len(statMap))
	for _, v := range statMap {
		list = append(list, v)
	}
	if err = mysql.Exec("TRUNCATE TABLE " + comm.TableProductStat).Error; err != nil {
		return
	}
	rows = len(list)
	if len(list) == 0 {
		return
	}
	const batch = 500
	for i := 0; i < len(list); i += batch {
		j := i + batch
		if j > len(list) {
			j = len(list)
		}
		if err = mysql.Table(comm.TableProductStat).CreateInBatches(list[i:j], batch).Error; err != nil {
			return
		}
	}
	return
}