starpivot/apps/services/modules/memory/report.go

package memory

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"yunyan/comm"
	"yunyan/lego/base"
	"yunyan/lego/core"
	"yunyan/lego/core/cbase"
	"yunyan/lego/sys/cron"
	redissys "yunyan/lego/sys/redis"
	"yunyan/pb"
)

/*
周期报告：cron 建单入队 → N 个 worker BRPop → SQL 算 stat_json → LLM 组织播报文案。

三条要点，全是踩过或推演出来的：

 1. **金额与完成数走 SQL，LLM 只组织文案**。数字算错比话说得干巴严重得多。
 2. **锁与队列 key 必须过 redissys.RKey 加应用前缀**。测试机上多个应用共用一个
    Redis 实例，锁 key 不带前缀会让 A 应用抢到的锁挡住 B 应用整周不生成报告，
    而且安安静静什么都不报。
 3. **漏跑比重复更可能**。当前 compose 是单副本，"多副本重复生成"是前瞻性风险；
    真会发生的是 00:30 跑到一半容器重启、cron 不补跑，报告永远停在 pending。
    所以 memory_getreport 里有一条超时重新入队的补偿（见 requeueIfStale）。
*/

type reportComp struct {
	cbase.ModuleCompBase
	module  *Memory
	service base.IRPCXService
	options *Options
	ctx     context.Context
	cancel  context.CancelFunc
}

func (this *reportComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, opt core.IModuleOptions) error {
	this.ModuleCompBase.Init(service, module, comp, opt)
	this.module = module.(*Memory)
	this.service = service.(base.IRPCXService)
	this.options = opt.(*Options)
	this.ctx, this.cancel = context.WithCancel(context.Background())
	return nil
}

func (this *reportComp) Start() error {
	if err := this.ModuleCompBase.Start(); err != nil {
		return err
	}
	for i := 0; i < this.options.MaxReportProcess; i++ {
		go this.consume(i)
	}
	if _, err := cron.AddFunc(this.options.ReportCronWeek, func() { this.tick(comm.MemoryPeriodWeek) }); err != nil {
		this.module.Errorf("注册周报 cron 失败 spec:%s err:%v", this.options.ReportCronWeek, err)
	}
	if _, err := cron.AddFunc(this.options.ReportCronMonth, func() { this.tick(comm.MemoryPeriodMonth) }); err != nil {
		this.module.Errorf("注册月报 cron 失败 spec:%s err:%v", this.options.ReportCronMonth, err)
	}
	this.module.Infof("周期报告已启动 workers:%d week:%q month:%q",
		this.options.MaxReportProcess, this.options.ReportCronWeek, this.options.ReportCronMonth)
	return nil
}

func (this *reportComp) Destroy() error {
	if this.cancel != nil {
		this.cancel()
	}
	return this.ModuleCompBase.Destroy()
}

// ===== key（一律经 RKey 加应用前缀）=====

func (this *reportComp) awaitKey() string {
	return redissys.RKey(fmt.Sprintf("%s:%s", this.service.GetTag(), QueueReportAwait))
}

func (this *reportComp) lockKey(ptype, pkey string) string {
	return redissys.RKey(fmt.Sprintf("%s:%s:%s:%s", this.service.GetTag(), LockReportPrefix, ptype, pkey))
}

// ===== cron =====

// tick 到点建单入队。多副本时用 SET NX EX 抢锁，只有一个实例真的干活。
func (this *reportComp) tick(ptype string) {
	now := time.Now()
	var start, end time.Time
	var pkey string
	switch ptype {
	case comm.MemoryPeriodWeek:
		// 周一 00:30 跑的是「上周」，往前退一天落到上周日再取那一周
		prev := now.AddDate(0, 0, -1)
		start, end = comm.MemoryWeekRange(prev)
		pkey = comm.MemoryWeekKey(prev)
	case comm.MemoryPeriodMonth:
		prev := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location()).AddDate(0, 0, -1)
		start, end = comm.MemoryMonthRange(prev)
		pkey = comm.MemoryMonthKey(prev)
	default:
		return
	}

	// 锁存 2 小时：足够覆盖一次生成，又不会长到把下一个周期也挡掉
	ok, err := redissys.Conn().SetNX(this.ctx, this.lockKey(ptype, pkey), "1", 2*time.Hour).Result()
	if err != nil {
		this.module.Errorf("报告 cron 抢锁失败 %s/%s err:%v", ptype, pkey, err)
		return
	}
	if !ok {
		this.module.Infof("报告 cron %s/%s 已被其它实例领走，跳过", ptype, pkey)
		return
	}

	sd, ed := comm.FormatMemoryDate(start), comm.FormatMemoryDate(end)
	uids, err := this.module.model.uidsWithItemsBetween(sd, ed, this.options.ReportMinItems)
	if err != nil {
		this.module.Errorf("报告 cron 扫描用户失败 %s/%s err:%v", ptype, pkey, err)
		return
	}
	this.module.Infof("报告 cron %s/%s 范围 %s~%s 命中 %d 个用户", ptype, pkey, sd, ed, len(uids))

	for _, uid := range uids {
		rec := &pb.DBMemoryReport{
			Uid: uid, PeriodType: ptype, PeriodKey: pkey,
			PeriodStart: sd, PeriodEnd: ed,
			State: pb.MemoryReportState_MemoryReportState_Pending,
		}
		if err := this.module.model.upsertReport(rec); err != nil {
			this.module.Errorf("报告建单失败 uid:%s %s/%s err:%v", uid, ptype, pkey, err)
			continue
		}
		if rec.State == pb.MemoryReportState_MemoryReportState_Done && rec.Confirmed {
			continue // upsert 里对已确认的直接返回原记录，不重算
		}
		this.enqueue(rec.Id)
	}
}

func (this *reportComp) enqueue(id uint64) {
	if err := redissys.Conn().LPush(this.ctx, this.awaitKey(), strconv.FormatUint(id, 10)).Err(); err != nil {
		this.module.Errorf("报告入队失败 id:%d err:%v", id, err)
	}
}

// requeueIfStale 补偿：报告卡在 pending/processing 超过阈值就重新入队。
//
// 由 memory_getreport 驱动（同 echomeet 用 PollTranscribe 兜「转写回调丢了」的思路）。
// 没有这条，容器在 00:30 重启一次就够让那一周的报告永远停在 pending —— 而且不报任何错。
func (this *reportComp) requeueIfStale(rec *pb.DBMemoryReport) bool {
	if rec == nil {
		return false
	}
	if rec.State != pb.MemoryReportState_MemoryReportState_Pending &&
		rec.State != pb.MemoryReportState_MemoryReportState_Processing {
		return false
	}
	if time.Now().Unix()-rec.UpdateTime < this.options.ReportStaleSec {
		return false
	}
	this.module.Warnf("报告 id:%d 卡在 state:%d 超过 %ds，重新入队", rec.Id, rec.State, this.options.ReportStaleSec)
	rec.State = pb.MemoryReportState_MemoryReportState_Pending
	if err := this.module.model.saveReport(rec); err != nil {
		this.module.Errorf("报告 id:%d 重置状态失败 err:%v", rec.Id, err)
		return false
	}
	this.enqueue(rec.Id)
	return true
}

// ===== worker =====

func (this *reportComp) consume(idx int) {
	key := this.awaitKey()
	for {
		select {
		case <-this.ctx.Done():
			return
		default:
		}
		res, err := redissys.Conn().BRPop(this.ctx, 5*time.Second, key).Result()
		if err != nil {
			continue // 超时是常态，不刷日志
		}
		if len(res) < 2 {
			continue
		}
		id, e := strconv.ParseUint(res[1], 10, 64)
		if e != nil {
			continue
		}
		this.process(id)
	}
}

func (this *reportComp) process(id uint64) {
	defer func() {
		if r := recover(); r != nil {
			this.module.Errorf("报告生成 panic id:%d err:%v", id, r)
		}
	}()

	rec, err := this.module.model.getReport(id)
	if err != nil || rec == nil {
		this.module.Errorf("报告生成取记录失败 id:%d err:%v", id, err)
		return
	}
	if rec.State == pb.MemoryReportState_MemoryReportState_Done {
		return
	}
	rec.State = pb.MemoryReportState_MemoryReportState_Processing
	_ = this.module.model.saveReport(rec)

	// 1) 统计走 SQL，确定性的
	st, err := this.module.computeStats(rec.Uid, rec.PeriodStart, rec.PeriodEnd)
	if err != nil {
		this.fail(rec, "统计失败: "+err.Error())
		return
	}
	if !st.hasContent() {
		// 到这儿说明入队后数据被删光了。空报告没有播报价值，直接标完成并确认掉，
		// 免得它挂在「待确认」里天天弹。
		rec.StatJson = toJSON(st)
		rec.Summary = ""
		rec.State = pb.MemoryReportState_MemoryReportState_Done
		rec.Confirmed = true
		rec.ConfirmTime = time.Now().Unix()
		_ = this.module.model.saveReport(rec)
		return
	}
	rec.StatJson = toJSON(st)

	// 2) LLM 只把 stat_json 组织成一段适合朗读的话
	if this.module.echo != nil {
		ctx, cancel := context.WithTimeout(this.ctx, 60*time.Second)
		text, svcId, e := this.module.echo.ChatLLM(ctx, "", this.options.ReportPrompt, rec.StatJson)
		cancel()
		rec.LlmSvcId = svcId
		if e != nil {
			// 文案生成失败**不算整体失败**：统计数据是完整的，客户端拿 stat_json
			// 一样能显示，只是少一段朗读文案。宁可少一句话，不可让报告整个没有。
			this.module.Warnf("报告 id:%d 文案生成失败，仅保留统计 err:%v", rec.Id, e)
		} else {
			rec.Summary = text
		}
	}

	rec.State = pb.MemoryReportState_MemoryReportState_Done
	rec.ErrorMsg = ""
	if err := this.module.model.saveReport(rec); err != nil {
		this.module.Errorf("报告 id:%d 回写失败 err:%v", rec.Id, err)
		return
	}
	this.module.Infof("报告生成完成 id:%d uid:%s %s/%s svc:%s",
		rec.Id, rec.Uid, rec.PeriodType, rec.PeriodKey, rec.LlmSvcId)
}

func (this *reportComp) fail(rec *pb.DBMemoryReport, msg string) {
	rec.State = pb.MemoryReportState_MemoryReportState_Failed
	if len([]rune(msg)) > 400 {
		msg = string([]rune(msg)[:400])
	}
	rec.ErrorMsg = msg
	_ = this.module.model.saveReport(rec)
	this.module.Errorf("报告生成失败 id:%d uid:%s %s", rec.Id, rec.Uid, msg)
}

func toJSON(v interface{}) string {
	b, err := json.Marshal(v)
	if err != nil {
		return "{}"
	}
	return string(b)
}