Release v0.1.12.30

.gitignore

@@ -0,0 +1 @@
+logs/

internal/domain/tenant_config.go

@@ -0,0 +1,22 @@
+package domain
+
+import (
+	"git.x2erp.com/qdy/go-base/ctx"
+	"git.x2erp.com/qdy/go-base/models"
+)
+
+// 租户配置信息
+type TenantConfig struct {
+	models.BsonStringModel `bson:",inline"`
+	AgentUrl               string `bson:"agent_url,omitempty"`
+	AgentConcurrency       int    `bson:"agent_concurrency,omitempty"`
+}
+
+// 构造函数，设置默认值
+func NewTenantConfig(ctx ctx.RequestContext, tenantId string, url string, concurrency int) *TenantConfig {
+	return &TenantConfig{
+		BsonStringModel:  models.NewStringIDModelWithID(ctx, tenantId),
+		AgentConcurrency: concurrency,
+		AgentUrl:         url,
+	}
+}

main.go

@@ -3,222 +3,165 @@ package main
 import (
 	"encoding/json"
 	"log"
-	"net/http"
-	"strings"
-	"time"
+
+	"git.x2erp.com/qdy/go-db/factory/database"
+	"git.x2erp.com/qdy/go-db/factory/mongodb"
+	"git.x2erp.com/qdy/go-svc-worker/internal/domain"
 
 	"git.x2erp.com/qdy/go-base/config"
+	"git.x2erp.com/qdy/go-base/consul"
+	"git.x2erp.com/qdy/go-base/container"
+	"git.x2erp.com/qdy/go-base/ctx"
 	"git.x2erp.com/qdy/go-base/logger"
-	"git.x2erp.com/qdy/go-base/myservice"
+	"git.x2erp.com/qdy/go-base/middleware"
 	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory/database"
-	"git.x2erp.com/qdy/go-svc-worker/service"
-	"go-micro.dev/v4/metadata"
+	"git.x2erp.com/qdy/go-base/webx"
+	"git.x2erp.com/qdy/go-base/webx/health"
+	"git.x2erp.com/qdy/go-base/webx/router"
 )
 
-// 定义业务服务
-type DBFactory struct {
-	dbFactory *database.DBFactory
-}
-
-// 配置
 var (
-	serviceName    string
-	serviceVersion string
+	appName    = "svc-worker"
+	appVersion = "1"
 )
 
 func main() {
+	//0.日志
+	//logger.InitBootLog()
+	logBootFactory := logger.InitBootLog()
 
-	// ========== 第一阶段：强制写入文件的启动日志 ==========
-	// 这一步确保即使配置加载失败，也有日志记录
-	if err := logger.InitBootLog("svc-worker"); err != nil {
-		// 连启动日志都初始化失败，只能输出到控制台
-		log.Fatal("无法初始化启动日志: ", err)
-	}
+	//1.获取配置文件
+	cfg := config.GetConfig()
+	cfg.SetAppName(appName)
+	cfg.SetAppVersion(appVersion)
 
-	logger.BootLog("开始加载配置...")
-	cfg, err := config.GetConfig()
-	if err != nil {
-		log.Fatalf("Failed to create RabbitMQ factory: %v", err)
-	}
+	//2.创建关闭容器
+	ctr := container.NewContainer(cfg)
 
-	serviceConfig := cfg.GetService()
-	microConfig := cfg.GetMicro()
-	serviceName = serviceConfig.ServiceName
-	serviceVersion = serviceConfig.ServiceVersion
+	//注册日志，实现自动关闭
+	container.Reg(ctr, logBootFactory)
 
-	log.Printf("serviceName: %s", serviceName)
-	log.Printf("Port: %d", serviceConfig.Port)
-	log.Printf("Consul: %s", microConfig.RegistryAddress)
+	//3.创建数据库工厂--如果需求
+	dbFactory := container.Create(ctr, database.CreateDBFactory)
 
-	// 2. 初始化数据库
-	dbFactory, err := database.GetDBFactory()
-	if err != nil {
-		log.Fatal("数据库连接失败:", err)
-	}
-	defer func() {
-		if err := dbFactory.Close(); err != nil {
-			logger.Info("数据库关闭错误: %v", err)
-		}
-	}()
+	// 赋值认证中间件参数
+	middleware.JWTAuthMiddlewareInit(cfg)
 
-	// 3. 创建服务实例
-	dbfactory := &DBFactory{dbFactory: dbFactory}
+	//测试数据库连接
+	dbFactory.TestConnection()
 
-	// 4. 使用 micro.Start 启动服务
-	webService := myservice.Start(cfg)
+	// 创建mongodb
+	mongoDBFactory := container.Create(ctr, mongodb.CreateFactory)
+	mongoDBFactory.TestConnection()
 
-	// 7. 注册HTTP路由
-	webService.Handle("/", http.HandlerFunc(rootHandler))
-	webService.Handle("/health", http.HandlerFunc(dbfactory.healthHandler))
-	webService.Handle("/info", http.HandlerFunc(infoHandler))
-	webService.Handle("/api/data/agent/to/doris", authMiddleware(http.HandlerFunc(dbfactory.agentToDorisHandler)))
+	//得到webservice服务工厂
+	webxFactory := webx.GetWebServiceFactory()
 
-	logger.InitRuntimeLogger(serviceName, cfg.GetLog())
+	//建立hhtpService服务
+	webServcie, _ := webxFactory.CreateService(cfg.GetServiceConfig())
 
-	log.Println("日志系统初始化完成")
+	//建立路由-api
+	routerService := router.NewWebService(webServcie.GetRouter())
 
-	// 程序退出时停止ES写入器
-	defer logger.StopESWriter()
-	logger.Debug("测试日志是否写入ES-----")
+	//注册路由--api
+	registerDefaultRouter(routerService, mongoDBFactory)
 
-	//关闭-启动日志输出文件功能
-	logger.CloseBootLogger()
+	// 注册健康检查-api
+	health.RegisterConsulHealthCheck(routerService)
 
-	if err := webService.Run(); err != nil {
-		log.Fatal("服务运行失败:", err)
-	}
+	//启动服务
+	webServcie.Run()
 
-}
+	//启用运行日志
+	container.Create(ctr, logger.InitRuntimeLogger)
 
-// 根处理器
-func rootHandler(w http.ResponseWriter, r *http.Request) {
-	if r.URL.Path != "/" {
-		http.NotFound(w, r)
-		return
-	}
-	respondJSON(w, http.StatusOK, map[string]string{
-		"service": serviceName,
-		"status":  "running",
-		"mode":    "http-microservice",
-	})
+	//注册到注册中心
+	container.Create(ctr, consul.Register)
+	//等待关闭
+	webServcie.WaitForServiceShutdown(ctr)
 }
 
-// 健康检查处理器
-func (s *DBFactory) healthHandler(w http.ResponseWriter, r *http.Request) {
-	if err := s.dbFactory.TestConnection(s.dbFactory.GetDBType()); err != nil {
-		respondJSON(w, http.StatusServiceUnavailable, map[string]string{
-			"status": "down",
-			"error":  err.Error(),
-		})
-		return
-	}
-	respondJSON(w, http.StatusOK, map[string]string{
-		"status": "up",
-		"time":   time.Now().Format(time.RFC3339),
-	})
-}
-
-// 信息处理器
-func infoHandler(w http.ResponseWriter, r *http.Request) {
-	respondJSON(w, http.StatusOK, map[string]interface{}{
-		"service": serviceName,
-		"version": serviceVersion,
-		"api": map[string]string{
-			"POST /api/data/agent/to/doris": "同步数据到Doris",
-			"GET /health":                   "健康检查",
-			"GET /info":                     "服务信息",
-			"GET /":                         "根路径",
-		},
-		"features": []string{
-			"服务发现（Consul）",
-			"负载均衡",
-			"健康检查",
-			"HTTP API网关",
+func registerDefaultRouter(ws *router.RouterService, mongoDBFactory *mongodb.MongoDBFactory) {
+	// // GET示例：路径参数绑定
+	// ws.GET("/api/users/{id}",
+	// 	func(id string, reqCtx *ctx.RequestContext) (UserResponse, error) {
+
+	// 		log.Print("ctx TenantID:", reqCtx.TenantID)
+	// 		// id 自动从路径绑定
+	// 		// 注意：webx版本没有自动注入dbFactory
+	// 		return getUser(id, dbFactory) // 需要修改getUser函数以获取dbFactory
+	// 	},
+	// ).Use(middleware.JWTAuthMiddleware).Register()
+
+	// POST示例：Body参数绑定
+	ws.POST("/api/tenant/config",
+		func(req domain.TenantConfig, reqCtx *ctx.RequestContext) (*types.QueryResult[interface{}], error) {
+			log.Print("ctx TenantID:", reqCtx.TenantID)
+			//log.Print("mongoDBFactory:", mongoDBFactory.GetConfig().URI)
+			//log.Print("dbFactory:", dbFactory.GetDBName())
+			jsonBytes, _ := json.Marshal(req)
+			log.Printf("TenantConfig :%s", string(jsonBytes))
+
+			ok := mongoDBFactory.InsertOne("tenant_configs", req)
+			log.Print("TenantConfig InsertOne:", ok)
+			// req 自动从JSON Body绑定
+			return &types.QueryResult[interface{}]{
+				Success: ok,
+				Data:    mongoDBFactory.GetConfig().URI,
+			}, nil
 		},
-	})
-}
-
-// AgentToDoris处理器
-func (s *DBFactory) agentToDorisHandler(w http.ResponseWriter, r *http.Request) {
-	if r.Method != "POST" {
-		respondJSON(w, http.StatusMethodNotAllowed, types.QueryResult{
-			Error:   "只支持POST请求",
-			Success: false,
-		})
-		return
-	}
+	).Use(middleware.JWTAuthMiddleware).Register()
 
-	// 解析请求
-	var requestData types.QueryRequest
-	if err := json.NewDecoder(r.Body).Decode(&requestData); err != nil {
-		respondJSON(w, http.StatusBadRequest, map[string]string{
-			"error": "无效的JSON数据",
-		})
-		return
-	}
+	// 您的业务路由
+	ws.POST("/api/query/yaml",
+		func() (interface{}, error) {
+			return queryYaml(nil) // 需要修改queryYaml函数以获取dbFactory
+		},
+	).Use(middleware.JWTAuthMiddleware).Register()
 
-	// 处理业务逻辑
-	result := service.ServiceAgentToDoris(s.dbFactory, requestData)
+	ws.POST("/api/init/config/template",
+		func() (interface{}, error) {
+			return initConfigTemplate(nil) // 需要修改initConfigTemplate函数以获取dbFactory
+		},
+	).Use(middleware.JWTAuthMiddleware).Register()
 
-	respondJSON(w, http.StatusOK, result)
 }
 
-// 认证中间件
-func authMiddleware(next http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-
-		// JWT令牌认证
-		token := r.Header.Get("Authorization")
-		if token != "" && strings.HasPrefix(token, "Bearer ") {
-			token = token[7:]
-		}
-
-		// 双重认证：API密钥或JWT
-		if token == "" {
-			respondJSON(w, http.StatusUnauthorized, map[string]string{
-				"error": "需要API密钥或Bearer令牌",
-			})
-			return
-		}
-
-		// 验证JWT令牌
-		if token != "" && !isValidJWT(token) {
-			respondJSON(w, http.StatusUnauthorized, map[string]string{
-				"error": "无效的访问令牌",
-			})
-			return
-		}
-
-		// 将认证信息添加到上下文
-		ctx := r.Context()
-		if token != "" {
-			ctx = metadata.Set(ctx, "Authorization", "Bearer "+token)
-		}
-
-		next.ServeHTTP(w, r.WithContext(ctx))
-	})
+func queryYaml(dbFactory *database.DBFactory) (interface{}, error) {
+	// 您的业务逻辑
+	return map[string]interface{}{"message": "query yaml success"}, nil
 }
 
-// JWT验证
-func isValidJWT(token string) bool {
-	// TODO: 实现JWT验证逻辑
-	// 可以使用 github.com/golang-jwt/jwt/v5
-	// 临时实现：检查token是否有效格式
-	//if len(token) < 10 {
-	//	return false
-	//	}
-	return true // 临时返回true，实际需要验证签名和过期时间
+func initConfigTemplate(dbFactory *database.DBFactory) (interface{}, error) {
+	// 您的业务逻辑
+	return map[string]interface{}{"message": "init config success"}, nil
 }
 
-// JSON响应辅助函数
-func respondJSON(w http.ResponseWriter, status int, data interface{}) {
-	w.Header().Set("Content-Type", "application/json")
-	w.Header().Set("X-Service-Name", serviceName)
-	w.Header().Set("X-Service-Version", serviceVersion)
-	w.WriteHeader(status)
+// getSQLWithPagination 生成带分页的SQL语句（参数模式）
+// 返回SQL语句和参数映射
+func getSQLWithPaginationSQL(startRow, endRow int) (string, []interface{}) {
+	sql := `SELECT
+    CLOTHING_ID,
+    CLOTHING_YEAR,
+    CLOTHING_NAME
+
+   FROM (
+    SELECT a.*, ROWNUM as rn
+    FROM (
+        SELECT *
+        FROM X6_STOCK_DEV.A3_CLOTHING 
+     
+        ORDER BY CLOTHING_ID
+    ) a
+    WHERE ROWNUM <= :1
+)
+WHERE rn > :2`
 
-	if err := json.NewEncoder(w).Encode(data); err != nil {
-		log.Printf("JSON编码错误: %v", err)
+	// 创建参数映射
+	params := []interface{}{
+		endRow,
+		startRow - 1, // WHERE rn > :start_row 所以是startRow-1
 	}
+
+	return sql, params
 }

service/agentToDoris.go → service/agent_to_doris.go

@@ -2,35 +2,41 @@ package service
 
 import (
 	"fmt"
-	"log"
 	"strings"
 	"time"
 
+	"git.x2erp.com/qdy/go-base/logger"
 	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory/database"
 	"git.x2erp.com/qdy/go-db/factory/doris"
 	"git.x2erp.com/qdy/go-db/factory/http"
 )
 
 // queryToCSVAndInsert 处理单次查询并将结果插入到Doris
-func queryToCSVAndInsert(dbFactory *database.DBFactory, queryRequest types.QueryRequest) (*types.QueryResult, error) {
-
+func queryToCSVAndInsert(queryRequest types.QueryRequest) *types.QueryResult[interface{}] {
 	// 记录查询开始时间
 	queryStartTime := time.Now()
 
 	// 1. 获取HTTP工厂实例
 	httpFactory, err := http.GetHTTPFactory()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get HTTP factory: %v", err)
+		logger.Errorf("Failed to get HTTP factory: %v", err)
+		return &types.QueryResult[interface{}]{
+			Success: false,
+			Error:   fmt.Sprintf("failed to get HTTP factory: %v", err),
+		}
 	}
-	log.Println("HTTP factory created successfully")
+	logger.Debug("HTTP factory created successfully")
 
 	// 2. 获取Doris工厂实例
 	dorisFactory, err := doris.GetDorisFactory(httpFactory)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get Doris factory: %v", err)
+		logger.Errorf("Failed to get Doris factory: %v", err)
+		return &types.QueryResult[interface{}]{
+			Success: false,
+			Error:   fmt.Sprintf("failed to get Doris factory: %v", err),
+		}
 	}
-	log.Println("Doris factory created successfully")
+	logger.Debug("Doris factory created successfully")
 
 	// 3. 创建HTTP客户端
 	httpClient := httpFactory.CreateClient()
@@ -41,6 +47,7 @@ func queryToCSVAndInsert(dbFactory *database.DBFactory, queryRequest types.Query
 		PositionalParams: queryRequest.PositionalParams,
 		WriterHeader:     queryRequest.WriterHeader,
 	}
+
 	// 4. 发送POST请求到 /api/query/csv 获取CSV格式数据
 	resp, err := httpClient.PostWithAuth(
 		queryRequest.AgentUrl,
@@ -53,29 +60,32 @@ func queryToCSVAndInsert(dbFactory *database.DBFactory, queryRequest types.Query
 	queryDuration := queryEndTime.Sub(queryStartTime)
 
 	if err != nil {
-		return &types.QueryResult{
+		logger.Errorf("Query request failed: %v", err)
+		return &types.QueryResult[interface{}]{
 			Success:   false,
 			Error:     fmt.Sprintf("查询失败: %v", err),
 			QueryTime: queryDuration,
-		}, nil
+		}
 	}
 
 	if resp.StatusCode() != 200 {
-		return &types.QueryResult{
+		logger.Errorf("Query request failed with status code: %d", resp.StatusCode())
+		return &types.QueryResult[interface{}]{
 			Success:   false,
 			Error:     fmt.Sprintf("查询请求失败, 状态码: %d", resp.StatusCode()),
 			QueryTime: queryDuration,
-		}, nil
+		}
 	}
 
 	// 5. 获取CSV数据
 	csvData := string(resp.Body())
 	if len(csvData) == 0 {
-		return &types.QueryResult{
-			Success:   false,
+		logger.Warn("No data queried")
+		return &types.QueryResult[interface{}]{
+			Success:   true,
 			Error:     "没有查询到数据",
 			QueryTime: queryDuration,
-		}, nil
+		}
 	}
 
 	// 估算数据行数（CSV行数）
@@ -90,7 +100,7 @@ func queryToCSVAndInsert(dbFactory *database.DBFactory, queryRequest types.Query
 		totalRows = strings.Count(csvData, "\n")
 	}
 
-	log.Printf("查询成功，获取到 %d 行数据\n", totalRows)
+	logger.Debug("Query successful, retrieved %d rows of data", totalRows)
 
 	// 6. 插入数据到Doris
 	database := queryRequest.DorisDatabase
@@ -103,39 +113,33 @@ func queryToCSVAndInsert(dbFactory *database.DBFactory, queryRequest types.Query
 	saveDuration := saveEndTime.Sub(saveStartTime)
 
 	if err != nil {
-		return &types.QueryResult{
+		logger.Errorf("Failed to insert data into Doris: %v", err)
+		return &types.QueryResult[interface{}]{
 			Success:   false,
 			Error:     fmt.Sprintf("数据插入Doris失败: %v", err),
 			QueryTime: queryDuration,
 			SaveTime:  saveDuration,
 			Count:     totalRows,
-		}, nil
+		}
 	}
 
-	return &types.QueryResult{
+	logger.Debug("Data successfully inserted into Doris: database=%s, table=%s, rows=%d",
+		database, table, totalRows)
+
+	return &types.QueryResult[interface{}]{
 		Success:   true,
 		QueryTime: queryDuration,
 		SaveTime:  saveDuration,
 		Count:     totalRows,
-	}, nil
+	}
 }
 
 // ServiceHandler 服务处理器（供router使用）
-func ServiceAgentToDoris(dbFactory *database.DBFactory, queryRequest types.QueryRequest) *types.QueryResult {
-
+func ServiceAgentToDoris(queryRequest types.QueryRequest) *types.QueryResult[interface{}] {
 	// 添加调试日志，确认 queryRequest 是否有数据
-	//log.Printf("SQL: %s", queryRequest.SQL)
-	//log.Printf("AgentUrl: %s", queryRequest.AgentUrl)
-	//log.Printf("参数数量: %d", len(queryRequest.PositionalParams))
+	logger.Debug("Processing query request: SQL=%s, AgentUrl=%s, Params count=%d",
+		queryRequest.SQL, queryRequest.AgentUrl, len(queryRequest.PositionalParams))
 
 	// 执行查询并插入
-	executionResult, err := queryToCSVAndInsert(dbFactory, queryRequest)
-	if err != nil {
-		return &types.QueryResult{
-			Success: false,
-			Error:   "服务内部错误: " + err.Error(),
-		}
-	}
-
-	return executionResult
+	return queryToCSVAndInsert(queryRequest)
 }

tables/worker_config.go

@@ -0,0 +1,34 @@
+package tables
+
+import (
+	"time"
+
+	"git.x2erp.com/qdy/go-db/sqldef"
+)
+
+func init() {
+	sqldef.AddRegistration(func(r *sqldef.Registry) {
+		// 配置表
+		tb := sqldef.NewTable("worker_config", "配置表").
+			ID("worker_config_id", 32).Comment("配置主键，主键").End().
+			String("package_name", 32).NotNull().Comment("包名称").End().
+			String("struct_name", 64).NotNull().Comment("结构体名称").End().
+			String("config_tag", 128).Comment("描述)").End().
+			JSON("config_value").Comment("配置数据)").End().
+			String("creator", 32).NotNull().Comment("创建人").End().
+			DateTime("created_at").NotNull().Default("CURRENT_TIMESTAMP").Comment("创建时间").End()
+		r.RegisterTable(tb.Build())
+	})
+}
+
+type WorkerConfig struct {
+	WorkerConfigID string         `gorm:"column:worker_config_id;type:varchar(32);primaryKey;not null;comment:配置主键，主键"`
+	PackageName    string         `gorm:"column:package_name;type:varchar(32);not null;comment:包名称"`
+	StructName     string         `gorm:"column:struct_name;type:varchar(64);not null;comment:结构体名称"`
+	ConfigTag      *string        `gorm:"column:config_tag;type:varchar(128);comment:描述)"`
+	ConfigValue    map[string]any `gorm:"column:config_value;type:json;comment:配置数据)"`
+	Creator        string         `gorm:"column:creator;type:varchar(32);not null;comment:创建人"`
+	CreatedAt      time.Time      `gorm:"column:created_at;not null;default:CURRENT_TIMESTAMP;comment:创建时间"`
+}
+
+func (WorkerConfig) TableName() string { return "worker_config" }

test/my_mongodb_test.go

@@ -0,0 +1,69 @@
+package test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+
+	"git.x2erp.com/qdy/go-svc-worker/internal/domain"
+)
+
+func TestTenantConfigAPI_Simple(t *testing.T) {
+	fmt.Println("=== 简单POST测试 ===")
+
+	// 1. 创建测试数据（赋值）
+	config := domain.TenantConfig{
+		//ID:               "zhx-1",
+		AgentUrl:         "http://localhost:8090/api",
+		AgentConcurrency: 50,
+	}
+	config.ID = "zhx-1"
+
+	fmt.Printf("测试数据: %v\n", config)
+
+	// 2. 转换为JSON
+	jsonData, err := json.Marshal(config)
+	if err != nil {
+		fmt.Printf("JSON转换错误: %v\n", err)
+		return
+	}
+
+	fmt.Printf("JSON数据: %s\n", string(jsonData))
+
+	// 3. 创建POST请求
+	url := "http://localhost:9090/api/tenant/config" // 你的服务端口
+
+	// 创建请求
+	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
+	if err != nil {
+		fmt.Printf("创建请求失败: %v\n", err)
+		return
+	}
+
+	// 设置请求头
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer aaa") // 这里放你的真实token
+
+	// 创建HTTP客户端并发送请求
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		fmt.Printf("POST请求失败: %v\n", err)
+		return
+	}
+	defer resp.Body.Close()
+
+	// 4. 读取响应
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		fmt.Printf("读取响应失败: %v\n", err)
+		return
+	}
+
+	// 5. 打印结果
+	fmt.Printf("状态码: %d\n", resp.StatusCode)
+	fmt.Printf("响应: %s\n", string(body))
+}