505 lines
16 KiB
Go
505 lines
16 KiB
Go
package report
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"dataengine/common/report/config"
|
||
"dataengine/common/report/ddlsync"
|
||
"dataengine/common/report/executor"
|
||
"dataengine/common/report/extract"
|
||
"dataengine/common/report/model"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/beans"
|
||
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
|
||
)
|
||
|
||
// ReportService 报表公共服务
|
||
// 对外暴露的统一接口
|
||
type ReportService struct {
|
||
configLoader *config.ConfigLoader
|
||
tableCreator *ddlsync.StatTableCreator
|
||
queryExecutor *executor.QueryExecutor
|
||
dailyExtractor *extract.DailyExtractor
|
||
}
|
||
|
||
var defaultService *ReportService
|
||
|
||
// GetService 获取报表服务单例
|
||
func GetService() *ReportService {
|
||
if defaultService == nil {
|
||
defaultService = &ReportService{
|
||
configLoader: config.GetLoader(),
|
||
tableCreator: ddlsync.NewStatTableCreator(),
|
||
queryExecutor: executor.NewQueryExecutor(),
|
||
dailyExtractor: extract.NewDailyExtractor(),
|
||
}
|
||
}
|
||
return defaultService
|
||
}
|
||
|
||
// ============================================================
|
||
// 核心接口 1: 自动创建统计宽表
|
||
// 首次抽取前调用
|
||
// ============================================================
|
||
|
||
// AutoCreateStatTable 根据配置自动创建统计宽表
|
||
// businessCode: 业务编码
|
||
// reportCode: 报表编码
|
||
func (s *ReportService) AutoCreateStatTable(ctx context.Context, businessCode, reportCode string) (*model.AutoCreateStatTableResp, error) {
|
||
// 初始化系统表
|
||
if err := initTables(ctx); err != nil {
|
||
return nil, fmt.Errorf("初始化系统表失败: %w", err)
|
||
}
|
||
|
||
resp, err := s.tableCreator.AutoCreateStatTable(ctx, businessCode, reportCode)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("AutoCreateStatTable 失败: %w", err)
|
||
}
|
||
|
||
return resp, nil
|
||
}
|
||
|
||
// ============================================================
|
||
// 核心接口 2: 按天抽取数据
|
||
// 业务层定时任务调用
|
||
// ============================================================
|
||
|
||
// ExtractDailyData 按天抽取数据
|
||
// businessCode: 业务编码
|
||
// reportCode: 报表编码
|
||
// statDate: 统计日期 yyyy-MM-dd
|
||
// executor: 执行人
|
||
func (s *ReportService) ExtractDailyData(ctx context.Context, businessCode, reportCode, statDate, executor string) (*model.ExtractDailyDataResp, error) {
|
||
// 1. 先确保统计宽表存在
|
||
report, err := s.configLoader.GetReport(ctx, businessCode, reportCode)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取报表配置失败: %w", err)
|
||
}
|
||
|
||
// 检查表是否存在
|
||
result, err := gfdb.DB(ctx).GetAll(ctx,
|
||
"SELECT EXISTS (SELECT 1 FROM pg_tables WHERE tablename = $1) AS exists",
|
||
strings.ToLower(report.StatTableName))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("检查统计宽表失败: %w", err)
|
||
}
|
||
tableExists := false
|
||
if len(result) > 0 {
|
||
tableExists = result[0]["exists"].Bool()
|
||
}
|
||
if !tableExists {
|
||
// 表不存在，先创建
|
||
if _, createErr := s.AutoCreateStatTable(ctx, businessCode, reportCode); createErr != nil {
|
||
return nil, fmt.Errorf("创建统计宽表失败: %w", createErr)
|
||
}
|
||
}
|
||
|
||
resp, err := s.dailyExtractor.ExtractDailyData(ctx, businessCode, reportCode, statDate, executor)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("ExtractDailyData 失败: %w", err)
|
||
}
|
||
|
||
// 清除缓存
|
||
s.configLoader.InvalidateCache(businessCode, reportCode)
|
||
|
||
return resp, nil
|
||
}
|
||
|
||
// ============================================================
|
||
// 核心接口 3: 用户选择查询（最核心）
|
||
// 前端用户选择条件 → 实时构建SQL → 返回报表数据
|
||
// ============================================================
|
||
|
||
// QueryReportByUserSelect 根据用户选择实时查询报表数据
|
||
// 不是自动生成报表，是用户在前端选择维度/指标/筛选/时间后实时查询展示
|
||
func (s *ReportService) QueryReportByUserSelect(ctx context.Context, req *model.UserSelectQueryReq) (*model.UserSelectQueryResp, error) {
|
||
// 参数校验
|
||
if req.BusinessCode == "" {
|
||
return nil, fmt.Errorf("businessCode 不能为空")
|
||
}
|
||
if req.ReportCode == "" {
|
||
return nil, fmt.Errorf("reportCode 不能为空")
|
||
}
|
||
|
||
resp, err := s.queryExecutor.QueryReportByUserSelect(ctx, req)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("QueryReportByUserSelect 失败: %w", err)
|
||
}
|
||
|
||
return resp, nil
|
||
}
|
||
|
||
// ============================================================
|
||
// 辅助接口
|
||
// ============================================================
|
||
|
||
// GetReportFields 获取报表可用字段（按维度/指标/筛选分类）
|
||
func (s *ReportService) GetReportFields(ctx context.Context, businessCode, reportCode string) (*model.GetReportFieldsResp, error) {
|
||
resp, err := s.configLoader.GetReportFields(ctx, businessCode, reportCode)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("GetReportFields 失败: %w", err)
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
// GetAllBusinesses 获取所有启用业务列表
|
||
func (s *ReportService) GetAllBusinesses(ctx context.Context) ([]model.BusinessConfig, error) {
|
||
return s.configLoader.GetAllBusinesses(ctx)
|
||
}
|
||
|
||
// GetAllReports 获取业务下所有报表列表
|
||
func (s *ReportService) GetAllReports(ctx context.Context, businessCode string) ([]model.ReportConfig, error) {
|
||
return s.configLoader.GetAllReports(ctx, businessCode)
|
||
}
|
||
|
||
// InvalidateCache 失效指定业务报表缓存
|
||
func (s *ReportService) InvalidateCache(businessCode, reportCode string) {
|
||
s.configLoader.InvalidateCache(businessCode, reportCode)
|
||
}
|
||
|
||
// InitSystemTables 初始化系统表
|
||
func (s *ReportService) InitSystemTables(ctx context.Context) error {
|
||
return initTables(ctx)
|
||
}
|
||
|
||
// ============================================================
|
||
// 配置 CRUD: 业务
|
||
// ============================================================
|
||
|
||
// SaveBusiness 保存业务配置（新增/修改合一）
|
||
func (s *ReportService) SaveBusiness(ctx context.Context, req *model.SaveBusinessReq) (*model.SaveResult, error) {
|
||
if err := initTables(ctx); err != nil {
|
||
return nil, fmt.Errorf("初始化系统表失败: %w", err)
|
||
}
|
||
|
||
biz := &model.BusinessConfig{
|
||
SQLBaseDO: beans.SQLBaseDO{Creator: req.Operator, Updater: req.Operator},
|
||
BusinessCode: req.BusinessCode,
|
||
BusinessName: req.BusinessName,
|
||
Description: req.Description,
|
||
Status: req.Status,
|
||
Config: req.Config,
|
||
}
|
||
|
||
if req.Status == "" {
|
||
biz.Status = model.StatusActive
|
||
}
|
||
if biz.Config == nil {
|
||
biz.Config = make(map[string]interface{})
|
||
}
|
||
|
||
if req.ID != nil && *req.ID > 0 {
|
||
// 更新
|
||
biz.Id = *req.ID
|
||
if err := s.configLoader.UpdateBusiness(ctx, biz); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: *req.ID, Message: "更新成功"}, nil
|
||
}
|
||
|
||
// 新增
|
||
id, err := s.configLoader.CreateBusiness(ctx, biz)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: id, Message: "创建成功"}, nil
|
||
}
|
||
|
||
// DeleteBusiness 删除业务配置
|
||
func (s *ReportService) DeleteBusiness(ctx context.Context, id int64) (*model.DeleteResult, error) {
|
||
biz, err := s.configLoader.GetBusinessByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.configLoader.DeleteBusiness(ctx, id, biz.BusinessCode); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.DeleteResult{Success: true, Message: "删除成功"}, nil
|
||
}
|
||
|
||
// GetBusiness 获取单个业务配置
|
||
func (s *ReportService) GetBusiness(ctx context.Context, id int64) (*model.BusinessConfig, error) {
|
||
return s.configLoader.GetBusinessByID(ctx, id)
|
||
}
|
||
|
||
// ============================================================
|
||
// 配置 CRUD: 报表
|
||
// ============================================================
|
||
|
||
// SaveReport 保存报表配置（新增/修改合一）
|
||
func (s *ReportService) SaveReport(ctx context.Context, req *model.SaveReportReq) (*model.SaveResult, error) {
|
||
if err := initTables(ctx); err != nil {
|
||
return nil, fmt.Errorf("初始化系统表失败: %w", err)
|
||
}
|
||
|
||
rpt := &model.ReportConfig{
|
||
SQLBaseDO: beans.SQLBaseDO{Creator: req.Operator, Updater: req.Operator},
|
||
BusinessCode: req.BusinessCode,
|
||
ReportCode: req.ReportCode,
|
||
ReportName: req.ReportName,
|
||
Description: req.Description,
|
||
Status: req.Status,
|
||
StatTableName: req.StatTableName,
|
||
StatTableComment: req.StatTableComment,
|
||
DateField: req.DateField,
|
||
PrimaryKeys: req.PrimaryKeys,
|
||
ConflictKeys: req.ConflictKeys,
|
||
Config: req.Config,
|
||
}
|
||
|
||
if req.Status == "" {
|
||
rpt.Status = model.StatusActive
|
||
}
|
||
if rpt.DateField == "" {
|
||
rpt.DateField = "stat_date"
|
||
}
|
||
if rpt.PrimaryKeys == nil {
|
||
rpt.PrimaryKeys = []string{"id"}
|
||
}
|
||
if rpt.ConflictKeys == nil {
|
||
rpt.ConflictKeys = []string{rpt.DateField}
|
||
}
|
||
if rpt.Config == nil {
|
||
rpt.Config = make(map[string]interface{})
|
||
}
|
||
|
||
if req.ID != nil && *req.ID > 0 {
|
||
rpt.Id = *req.ID
|
||
if err := s.configLoader.UpdateReport(ctx, rpt); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: *req.ID, Message: "更新成功"}, nil
|
||
}
|
||
|
||
id, err := s.configLoader.CreateReport(ctx, rpt)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: id, Message: "创建成功"}, nil
|
||
}
|
||
|
||
// DeleteReport 删除报表配置
|
||
func (s *ReportService) DeleteReport(ctx context.Context, id int64) (*model.DeleteResult, error) {
|
||
rpt, err := s.configLoader.GetReportByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.configLoader.DeleteReport(ctx, id, rpt.BusinessCode, rpt.ReportCode); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.DeleteResult{Success: true, Message: "删除成功"}, nil
|
||
}
|
||
|
||
// GetReport 获取单个报表配置
|
||
func (s *ReportService) GetReport(ctx context.Context, id int64) (*model.ReportConfig, error) {
|
||
return s.configLoader.GetReportByID(ctx, id)
|
||
}
|
||
|
||
// ============================================================
|
||
// 配置 CRUD: 字段
|
||
// ============================================================
|
||
|
||
// SaveField 保存字段配置（新增/修改合一）
|
||
func (s *ReportService) SaveField(ctx context.Context, req *model.SaveFieldReq) (*model.SaveResult, error) {
|
||
if err := initTables(ctx); err != nil {
|
||
return nil, fmt.Errorf("初始化系统表失败: %w", err)
|
||
}
|
||
|
||
field := &model.FieldConfig{
|
||
SQLBaseDO: beans.SQLBaseDO{Creator: req.Operator, Updater: req.Operator},
|
||
BusinessCode: req.BusinessCode,
|
||
ReportCode: req.ReportCode,
|
||
FieldCode: req.FieldCode,
|
||
FieldName: req.FieldName,
|
||
FieldType: req.FieldType,
|
||
DataType: req.DataType,
|
||
FieldRole: req.FieldRole,
|
||
IsAggregatable: req.IsAggregatable,
|
||
IsFilterable: req.IsFilterable,
|
||
IsQueryable: req.IsQueryable,
|
||
IsSortable: req.IsSortable,
|
||
DefaultAggregate: req.DefaultAggregate,
|
||
ValidAggregates: req.ValidAggregates,
|
||
FilterOperators: req.FilterOperators,
|
||
Expression: req.Expression,
|
||
ExpressionType: req.ExpressionType,
|
||
FormatPattern: req.FormatPattern,
|
||
Unit: req.Unit,
|
||
DictCode: req.DictCode,
|
||
SortOrder: req.SortOrder,
|
||
GroupName: req.GroupName,
|
||
Status: req.Status,
|
||
}
|
||
|
||
// 校验字段类型
|
||
validFieldTypes := map[string]bool{
|
||
model.FieldTypeString: true,
|
||
model.FieldTypeInt: true,
|
||
model.FieldTypeFloat: true,
|
||
model.FieldTypeDate: true,
|
||
model.FieldTypeDatetime: true,
|
||
model.FieldTypeJsonb: true,
|
||
}
|
||
if req.FieldType != "" && !validFieldTypes[req.FieldType] {
|
||
return nil, fmt.Errorf("不支持的字段类型: %s，仅支持 STRING/INT/FLOAT/DATE/DATETIME/JSONB", req.FieldType)
|
||
}
|
||
if req.DataType != "" && !validFieldTypes[req.DataType] {
|
||
return nil, fmt.Errorf("不支持的存储类型: %s，仅支持 STRING/INT/FLOAT/DATE/DATETIME/JSONB", req.DataType)
|
||
}
|
||
|
||
if req.Status == "" {
|
||
field.Status = model.StatusActive
|
||
}
|
||
if field.DataType == "" {
|
||
field.DataType = model.FieldTypeString
|
||
}
|
||
if field.ValidAggregates == nil {
|
||
field.ValidAggregates = []string{}
|
||
}
|
||
if field.FilterOperators == nil {
|
||
field.FilterOperators = []string{"=", "!=", ">", "<", ">=", "<=", "IN", "LIKE", "BETWEEN"}
|
||
}
|
||
|
||
// 新增时校验字段编码唯一性
|
||
if req.ID == nil || *req.ID == 0 {
|
||
existing, err := s.configLoader.GetReportFields(ctx, req.BusinessCode, req.ReportCode)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("校验字段编码失败: %w", err)
|
||
}
|
||
allFields := append(append(existing.Dimensions, existing.Indicators...), existing.Filters...)
|
||
for _, f := range allFields {
|
||
if f.FieldCode == req.FieldCode && f.Status == model.StatusActive {
|
||
return nil, fmt.Errorf("字段编码 %s 已存在（业务: %s, 报表: %s）", req.FieldCode, req.BusinessCode, req.ReportCode)
|
||
}
|
||
}
|
||
}
|
||
|
||
if req.ID != nil && *req.ID > 0 {
|
||
field.Id = *req.ID
|
||
if err := s.configLoader.UpdateField(ctx, field); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: *req.ID, Message: "更新成功"}, nil
|
||
}
|
||
|
||
id, err := s.configLoader.CreateField(ctx, field)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: id, Message: "创建成功"}, nil
|
||
}
|
||
|
||
// DeleteField 删除字段配置
|
||
func (s *ReportService) DeleteField(ctx context.Context, id int64) (*model.DeleteResult, error) {
|
||
field, err := s.configLoader.GetFieldByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.configLoader.DeleteField(ctx, id, field.BusinessCode, field.ReportCode); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.DeleteResult{Success: true, Message: "删除成功"}, nil
|
||
}
|
||
|
||
// GetField 获取单个字段配置
|
||
func (s *ReportService) GetField(ctx context.Context, id int64) (*model.FieldConfig, error) {
|
||
return s.configLoader.GetFieldByID(ctx, id)
|
||
}
|
||
|
||
// ============================================================
|
||
// 配置 CRUD: 抽取配置
|
||
// ============================================================
|
||
|
||
// SaveExtractConfig 保存抽取配置（新增/修改合一）
|
||
func (s *ReportService) SaveExtractConfig(ctx context.Context, req *model.SaveExtractConfigReq) (*model.SaveResult, error) {
|
||
if err := initTables(ctx); err != nil {
|
||
return nil, fmt.Errorf("初始化系统表失败: %w", err)
|
||
}
|
||
|
||
ec := &model.ExtractConfig{
|
||
SQLBaseDO: beans.SQLBaseDO{Creator: req.Operator, Updater: req.Operator},
|
||
BusinessCode: req.BusinessCode,
|
||
ReportCode: req.ReportCode,
|
||
ExtractCode: req.ExtractCode,
|
||
ExtractName: req.ExtractName,
|
||
SourceTableName: req.SourceTableName,
|
||
SourceTableAlias: req.SourceTableAlias,
|
||
TargetTableName: req.TargetTableName,
|
||
IsEnabled: req.IsEnabled,
|
||
ExtractType: req.ExtractType,
|
||
ExtractMode: req.ExtractMode,
|
||
ExtractKeyField: req.ExtractKeyField,
|
||
ExtractKeyFormat: req.ExtractKeyFormat,
|
||
GroupByFields: req.GroupByFields,
|
||
FilterExpression: req.FilterExpression,
|
||
JoinConfigs: req.JoinConfigs,
|
||
FieldMappings: req.FieldMappings,
|
||
TransformRules: req.TransformRules,
|
||
BatchSize: req.BatchSize,
|
||
Status: req.Status,
|
||
}
|
||
|
||
if req.Status == "" {
|
||
ec.Status = model.StatusActive
|
||
}
|
||
if ec.ExtractType == "" {
|
||
ec.ExtractType = model.ExtractTypeIncremental
|
||
}
|
||
if ec.ExtractMode == "" {
|
||
ec.ExtractMode = model.ExtractModeDirect
|
||
}
|
||
if ec.BatchSize == 0 {
|
||
ec.BatchSize = 1000
|
||
}
|
||
if ec.JoinConfigs == nil {
|
||
ec.JoinConfigs = []model.JoinConfig{}
|
||
}
|
||
if ec.FieldMappings == nil {
|
||
ec.FieldMappings = []model.FieldMapping{}
|
||
}
|
||
if ec.TransformRules == nil {
|
||
ec.TransformRules = []model.TransformRule{}
|
||
}
|
||
if ec.GroupByFields == nil {
|
||
ec.GroupByFields = []string{}
|
||
}
|
||
|
||
if req.ID != nil && *req.ID > 0 {
|
||
ec.Id = *req.ID
|
||
if err := s.configLoader.UpdateExtractConfig(ctx, ec); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: *req.ID, Message: "更新成功"}, nil
|
||
}
|
||
|
||
id, err := s.configLoader.CreateExtractConfig(ctx, ec)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.SaveResult{Success: true, ID: id, Message: "创建成功"}, nil
|
||
}
|
||
|
||
// DeleteExtractConfig 删除抽取配置
|
||
func (s *ReportService) DeleteExtractConfig(ctx context.Context, id int64) (*model.DeleteResult, error) {
|
||
ec, err := s.configLoader.GetExtractConfigByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.configLoader.DeleteExtractConfig(ctx, id, ec.BusinessCode, ec.ReportCode); err != nil {
|
||
return nil, err
|
||
}
|
||
return &model.DeleteResult{Success: true, Message: "删除成功"}, nil
|
||
}
|
||
|
||
// GetExtractConfig 获取单个抽取配置
|
||
func (s *ReportService) GetExtractConfig(ctx context.Context, id int64) (*model.ExtractConfig, error) {
|
||
return s.configLoader.GetExtractConfigByID(ctx, id)
|
||
}
|
||
|
||
// GetExtractConfigs 获取业务报表下所有抽取配置
|
||
func (s *ReportService) GetExtractConfigs(ctx context.Context, businessCode, reportCode string) ([]model.ExtractConfig, error) {
|
||
return s.configLoader.GetExtractConfigs(ctx, businessCode, reportCode)
|
||
}
|