starpivot/apps/services/modules/echomeet/model.go

package echomeet

import (
	"fmt"
	"path/filepath"
	"runtime"
	"yunyan/comm"
	"yunyan/lego/core"
	"yunyan/lego/core/cbase"
	"yunyan/lego/sys/mysql"
	"yunyan/lego/sys/postgres"
	"yunyan/pb"

	"gorm.io/gorm/clause"
)

// templatePrivateIdStart 模板 id 分段阈值（唯一的归属判定依据）：
//   - id <  此值 → 公共模板，存公共数据库（Postgres），可有多语言版本；
//   - id >= 此值 → 用户私有模板，存业务服务数据库（MySQL），单条、无多语言版本。
//
// 由建表时对私有模板表设置自增起点保证两段 id 不重叠（见 modelComp.Init）。
const templatePrivateIdStart = 10000

// 代理模型
type modelComp struct {
	cbase.ModuleCompBase
	module *Echomeet
}

// 组件初始化接口
func (this *modelComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, opt core.IModuleOptions) (err error) {
	this.ModuleCompBase.Init(service, module, comp, opt)
	this.module = module.(*Echomeet)
	if err = mysql.CreateTable(comm.TableEchomeetTemplate, &pb.DBEchoMeetTemplate{}); err == nil {
		// 私有模板 id 从分段起点(10000)开始递增，确保与公共模板(公共库，id<10000)id 段不重叠，
		// 从而 gettemplateforid 能仅凭 id 把请求路由到正确的库（仅空表生效）。
		mysql.AutoIncrementStart(comm.TableEchomeetTemplate, "id", templatePrivateIdStart)
	}
	if err = mysql.CreateTable(comm.TableEchomeetRecord, &pb.DBEchoMeetRecord{}); err != nil {
		return
	}
	this.ensureRecordFulltextIndex()
	return
}

// 会议纪要全文检索索引的名字与列。
//
// ⚠️ 列的顺序和个数**必须与查询里的 MATCH(...) 完全一致**，MySQL 才会用上这个索引；
// 少一列多一列都会报 "Can't find FULLTEXT index matching the column list"。
// 改这里就要同步改 modules/mcp/tool_meeting.go 的 fulltextColumns。
const (
	recordFulltextIndex   = "ft_echomeet_record_content"
	recordFulltextColumns = "title,summary,overview,remark"
)

// ensureRecordFulltextIndex 给会议记录建 ngram 全文索引（幂等，失败不阻断启动）。
//
// 为什么要它：MCP 的 search_meeting_notes 原来用 `LIKE %kw%` 逐个关键词 AND，
// 中文场景下召回极差——用户问「关于降本增效的会议」，纪要里写的是「成本优化」就一条都匹配不到，
// 而且每个关键词都是全表扫描。换成全文索引后按相关度排序，多关键词是 OR + 打分，
// 召回和排序都合理得多。
//
// ⚠️ 用 ngram 解析器，因为默认解析器按空格切词，中文整段会被当成一个巨型 token，等于没索引。
// 服务器上 `ngram_token_size=2`，所以**单字关键词检索不到**（工具那边对单字回退用 LIKE）。
//
// ⚠️ 失败只记日志：索引建不出来（权限不足、旧版 MySQL 不支持 ngram）时，
// 检索侧会自动退回 LIKE，功能降级但不能因此让 home 服务起不来。
func (this *modelComp) ensureRecordFulltextIndex() {
	var n int64
	if err := mysql.Table("information_schema.STATISTICS").
		Where("TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?",
			comm.TableEchomeetRecord, recordFulltextIndex).
		Count(&n).Error; err != nil {
		this.module.Warnf("检查会议全文索引失败，跳过建索引: %v", err)
		return
	}
	if n > 0 {
		return
	}
	sql := fmt.Sprintf("CREATE FULLTEXT INDEX %s ON `%s` (%s) WITH PARSER ngram",
		recordFulltextIndex, comm.TableEchomeetRecord, recordFulltextColumns)
	if err := mysql.Exec(sql).Error; err != nil {
		this.module.Warnf("创建会议全文索引失败（检索将退回 LIKE）: %v", err)
		return
	}
	this.module.Infof("会议全文索引已创建: %s(%s)", recordFulltextIndex, recordFulltextColumns)
}

func (this *modelComp) getuser(uid string) (user *pb.DBUser, err error) {
	user = &pb.DBUser{}
	err = mysql.FindOne(comm.TableUser, user, "uid=?", uid)
	return
}

func (this *modelComp) updateuser(user *pb.DBUser) (err error) {
	err = mysql.Save(comm.TableUser, user)
	return
}

// d读取模版
func (this *modelComp) getcommtemplates(language string) (models []*pb.DBEchoMeetTemplate, err error) {
	models = make([]*pb.DBEchoMeetTemplate, 0)
	if language == "" {
		err = postgres.Find(comm.TableEchomeetTemplate, &models, "source = ?", "public")
		return
	}
	if err = postgres.Find(comm.TableEchomeetTemplate, &models, "source = ? AND language = ?", "public", language); err != nil {
		return
	}
	if len(models) > 0 {
		return // 精确命中
	}
	// 回退：该区域 locale 无专属模板时，取同基础语言变体（裸 base 或区域变体，如 es-MX→es/es-ES），
	// 按 tid 去重保留 id 最小一条，避免同一模板出现多条。
	if base := baseLang(language); base != "" && base != language {
		var variants []*pb.DBEchoMeetTemplate
		if err = postgres.Find(comm.TableEchomeetTemplate, &variants, "source = ? AND (language = ? OR language LIKE ?)", "public", base, base+"-%"); err != nil {
			return
		}
		models = pickBaseLangVariants(variants, base)
	}
	return
}

// pickBaseLangVariants 从候选模板里挑出与 base 同基础语言的，按 tid 去重保留 id 最小一条。
// 用于区域 locale（如 es-MX）无专属模板时回退到同基础语言那套（es / es-ES）做列表，避免同 tid 重复。
func pickBaseLangVariants(candidates []*pb.DBEchoMeetTemplate, base string) []*pb.DBEchoMeetTemplate {
	byTid := make(map[string]*pb.DBEchoMeetTemplate)
	for _, m := range candidates {
		if baseLang(m.Language) != base {
			continue
		}
		if cur, ok := byTid[m.Tid]; !ok || m.Id < cur.Id {
			byTid[m.Tid] = m
		}
	}
	out := make([]*pb.DBEchoMeetTemplate, 0, len(byTid))
	for _, m := range byTid {
		out = append(out, m)
	}
	return out
}

// 读取模版
func (this *modelComp) gettemplates(uid string, language string) (models []*pb.DBEchoMeetTemplate, err error) {
	models = make([]*pb.DBEchoMeetTemplate, 0)
	if language == "" {
		err = mysql.Find(comm.TableEchomeetTemplate, &models, "source = ?", uid)
	} else {
		err = mysql.Find(comm.TableEchomeetTemplate, &models, "source = ? AND language = ?", uid, language)
	}
	return
}

/*
gettemplateforid 按主键读取模板，仅凭 id 段判定归属并路由到对应库：
  - id <  templatePrivateIdStart：公共模板 → 公共数据库（Postgres）
  - id >= templatePrivateIdStart：用户私有模板 → 业务服务数据库（MySQL）

私有模板单条、无多语言版本，按主键直接取即可。
*/
func (this *modelComp) gettemplateforid(id uint64) (template *pb.DBEchoMeetTemplate, err error) {
	template = &pb.DBEchoMeetTemplate{}
	if id >= templatePrivateIdStart {
		err = mysql.FindOne(comm.TableEchomeetTemplate, template, "id=?", id)
		return
	}
	err = postgres.FindOne(comm.TableEchomeetTemplate, template, "id=?", id)
	return
}

// resolvetemplate 私有模板（id>=起点）单条无多语言，按主键直取；
// 公共模板 tid 非空按 tid+language 取对应语言版本，否则按主键。
func (this *modelComp) resolvetemplate(tid string, templateid uint64, language string) (template *pb.DBEchoMeetTemplate, err error) {
	if templateid >= templatePrivateIdStart {
		return this.gettemplateforid(templateid)
	}
	if tid != "" {
		return this.gettemplatefortid(tid, language)
	}
	return this.gettemplateforid(templateid)
}

// gettemplatefortid 按 tid + language 查找模板（用于公共模板）。
// 精确匹配不到时，回退到同一基础语言的任一变体（如 es-MX→es-ES、zh-TW→zh-CN），
// 避免区域 locale 没有专属模板时解析失败；FindOne→gorm First 默认按主键 id 升序，取最小一条保证结果稳定。
func (this *modelComp) gettemplatefortid(tid string, language string) (template *pb.DBEchoMeetTemplate, err error) {
	template = &pb.DBEchoMeetTemplate{}
	if err = postgres.FindOne(comm.TableEchomeetTemplate, template, "tid=? AND language=?", tid, language); err == nil {
		return
	}
	if err = mysql.FindOne(comm.TableEchomeetTemplate, template, "tid=? AND language=?", tid, language); err == nil {
		return
	}
	// 回退：同基础语言（裸 base 或其任一区域变体，如 es / es-ES 都能兜底 es-MX）
	if base := baseLang(language); base != "" && base != language {
		if err = postgres.FindOne(comm.TableEchomeetTemplate, template, "tid=? AND (language=? OR language LIKE ?)", tid, base, base+"-%"); err == nil {
			return
		}
		err = mysql.FindOne(comm.TableEchomeetTemplate, template, "tid=? AND (language=? OR language LIKE ?)", tid, base, base+"-%")
	}
	return
}

func (this *modelComp) addtemplates(template *pb.DBEchoMeetTemplate) (err error) {
	err = mysql.Insert(comm.TableEchomeetTemplate, template)
	return
}

func (this *modelComp) updatetemplates(template *pb.DBEchoMeetTemplate) (err error) {
	err = mysql.Save(comm.TableEchomeetTemplate, template)
	return
}

func (this *modelComp) deltemplates(id []uint64) (err error) {
	err = mysql.Delete(comm.TableEchomeetTemplate, "id IN ?", id)
	return
}

func (this *modelComp) getrecordsforuid(uid string) (models []*pb.DBEchoMeetRecord, err error) {
	models = make([]*pb.DBEchoMeetRecord, 0)
	err = mysql.Find(comm.TableEchomeetRecord, &models, "uid=?", uid)
	return
}

// 查询用户记录的精简信息（不含 original、translate、summary 大字段）
func (this *modelComp) getrecordbriefforuid(uid string) (models []*pb.DBEchoMeetRecord, err error) {
	models = make([]*pb.DBEchoMeetRecord, 0)
	err = mysql.Table(comm.TableEchomeetRecord).
		Omit("original", "translate", "summary").
		Where("uid = ?", uid).
		Find(&models).Error
	return
}
func (this *modelComp) getrecord(id uint64) (model *pb.DBEchoMeetRecord, err error) {
	model = &pb.DBEchoMeetRecord{}
	err = mysql.FindOne(comm.TableEchomeetRecord, model, "id=?", id)
	return
}
func (this *modelComp) getrecordByTaskId(taskId string) (model *pb.DBEchoMeetRecord, err error) {
	model = &pb.DBEchoMeetRecord{}
	err = mysql.FindOne(comm.TableEchomeetRecord, model, "taskid=?", taskId)
	return
}

// getstucktranscribing 捞一批「卡在转写中」的记录：已经拿到第三方 taskid，
// 但距上次主动查询超过阈值。
//
// ⚠️ lastquerytime=0 也要算进来——那表示**从来没有人查过**（客户端一次都没拉过，
// 比如它的轮询循环崩了）。只按 `lastquerytime<?` 过滤会把这批最需要兜底的漏掉。
//
// ⚠️ NULL 也要单独判。SQL 里 `NULL=0` 和 `NULL<x` 都不成立（结果是 NULL 不是
// true），漏掉 `IS NULL` 的话这类行对兜底**永远不可见**，卡死了也捞不出来。
// gorm 写入的新记录是 0 不是 NULL，但历史行/手工插入的不保证。
func (this *modelComp) getstucktranscribing(before int64, limit int) (models []*pb.DBEchoMeetRecord, err error) {
	models = make([]*pb.DBEchoMeetRecord, 0)
	err = mysql.Table(comm.TableEchomeetRecord).
		Where("state=? AND taskid<>'' AND (lastquerytime IS NULL OR lastquerytime=0 OR lastquerytime<?)",
			int32(pb.DBEchoMeetRecordState_Transcribing), before).
		Order("id").Limit(limit).Find(&models).Error
	return
}

// getstucksummarizing 捞一批停在「待总结/总结中」超过阈值的记录。
// 锚点用 starttime（StartTask 时写入），转写完成后它不会再变，正好当「这单开始多久了」。
func (this *modelComp) getstucksummarizing(before int64, limit int) (models []*pb.DBEchoMeetRecord, err error) {
	models = make([]*pb.DBEchoMeetRecord, 0)
	err = mysql.Table(comm.TableEchomeetRecord).
		Where("state IN ? AND starttime>0 AND starttime<?",
			[]int32{
				int32(pb.DBEchoMeetRecordState_AwaitSummarizing),
				int32(pb.DBEchoMeetRecordState_Summarizing),
			}, before).
		Order("id").Limit(limit).Find(&models).Error
	return
}

func (this *modelComp) getrecords(ids []uint64) (models []*pb.DBEchoMeetRecord, err error) {
	models = make([]*pb.DBEchoMeetRecord, 0)
	err = mysql.Find(comm.TableEchomeetRecord, &models, "id IN ?", ids)
	return
}
func (this *modelComp) addrecord(model *pb.DBEchoMeetRecord) (err error) {
	err = mysql.Insert(comm.TableEchomeetRecord, model)
	return
}
func (this *modelComp) saverecords(record []*pb.DBEchoMeetRecord) (err error) {
	caller := saveCaller()
	for _, r := range record {
		this.module.Infof("saverecord(batch) [from %s] id:%d state:%s taskid:%q logid:%q",
			caller, r.Id, r.State.String(), r.Taskid, r.Logid)
	}
	err = mysql.Save(comm.TableEchomeetRecord, record)
	if err != nil {
		this.module.Errorf("saverecord(batch) [from %s] 失败 err:%v", caller, err)
	}
	return
}
func (this *modelComp) saverecord(record *pb.DBEchoMeetRecord) (err error) {
	caller := saveCaller()
	// 重点观测 taskid：保存时打印调用方+本次写入的 taskid/logid/state。
	// taskid 由「有」变「空」时，对照这行的调用方即可定位是哪段代码覆盖掉的。
	this.module.Infof("saverecord [from %s] id:%d state:%s taskid:%q logid:%q seconds:%d size:%d",
		caller, record.Id, record.State.String(), record.Taskid, record.Logid, record.Seconds, record.Size)
	if record.Taskid == "" && (record.State == pb.DBEchoMeetRecordState_Transcribing ||
		record.State == pb.DBEchoMeetRecordState_AwaitSummarizing ||
		record.State == pb.DBEchoMeetRecordState_Summarizing) {
		this.module.Warnf("saverecord [from %s] id:%d 异常：state=%s 却写入空 taskid（疑似被覆盖）",
			caller, record.Id, record.State.String())
	}
	err = mysql.Save(comm.TableEchomeetRecord, record)
	if err != nil {
		this.module.Errorf("saverecord [from %s] id:%d 失败 err:%v", caller, record.Id, err)
	}
	return
}

// saveCaller 返回调用 saverecord/saverecords 的上一层代码位置（file:line），
// 用于在日志里标明本次写库是哪条业务路径触发的。
func saveCaller() string {
	if _, file, line, ok := runtime.Caller(2); ok {
		return fmt.Sprintf("%s:%d", filepath.Base(file), line)
	}
	return "?"
}
func (this *modelComp) delrecords(ids []uint64) (err error) {
	err = mysql.Delete(comm.TableEchomeetRecord, "id IN ?", ids)
	return
}

func (this *modelComp) getStatistics(uid string) (statistics *pb.DBUserStatistics, err error) {
	statistics = &pb.DBUserStatistics{Uid: uid}
	mysql.FindOne(comm.TableUserStatistics, statistics, "uid=?", uid)
	return
}

func (this *modelComp) updateStatistics(statistics *pb.DBUserStatistics) (err error) {
	err = mysql.Table(comm.TableUserStatistics).
		Clauses(clause.OnConflict{UpdateAll: true}).
		Create(statistics).Error
	return
}

func (this *modelComp) addUseLog(log *pb.DBUserUseLog) (err error) {
	err = mysql.Insert(comm.TableUserUseLog, log)
	return
}