control/internal/control.go

package internal

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"os"
	"os/exec"
	"strings"
	"time"

	"git.apinb.com/bsm-sdk/core/cache/redis"
	"git.apinb.com/bsm-sdk/core/with"
	"github.com/robfig/cron/v3"
)

var (
	CacheSrc     = "redis://null:Weidong2023~!@8.137.107.29:19379/1"
	RedisService *redis.RedisClient // Redis 客户端服务
)

func BootControl() {
	// 初始化 Redis 缓存
	RedisService = with.RedisCache(CacheSrc)
	scheduler := cron.New()
	scheduler.AddFunc("@every 5s", func() {
		ControlTask()
	})
	scheduler.Start()
}

type StatusData struct {
	Processes ProcessStatus `json:"processes"`
	Status    interface{}   `json:"status"`
	Account   interface{}   `json:"account"`
	Positions interface{}   `json:"positions"`
	Timestamp int64         `json:"timestamp"`
}

type ProcessStatus struct {
	PythonMain bool `json:"python_main"`
	XtMiniQmt  bool `json:"xt_mini_qmt"`
}

func ControlTask() {
	status := &StatusData{
		Timestamp: time.Now().Unix(),
	}

	// 检查 2 个线程 (python main.py，和 XtMiniQmt.exe) 是否正在运行
	status.Processes = checkProcesses()

	// 读取 http://127.0.0.1:5000/status
	status.Status = readAPIStatus()

	// 读取 account.json
	status.Account = readJSONFile("D:\\work\\qmt\\data\\account.json")

	// 读取 positions.json
	status.Positions = readJSONFile("D:\\work\\qmt\\data\\positions.json")

	// 将以上所有数据汇总，合本成一个 JSON，存储至 Redis,List 表，Key: qmt:status
	storeToRedis(status)
}

func checkProcesses() ProcessStatus {
	ps := ProcessStatus{
		PythonMain: false,
		XtMiniQmt:  false,
	}

	processes, err := getRunningProcesses()
	if err != nil {
		return ps
	}

	for _, proc := range processes {
		if strings.Contains(proc, "python") && strings.Contains(proc, "main.py") {
			ps.PythonMain = true
		}
		if strings.Contains(proc, "XtMiniQmt.exe") || strings.Contains(proc, "XtMiniQmt") {
			ps.XtMiniQmt = true
		}
	}

	return ps
}

func getRunningProcesses() ([]string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Windows 系统使用 tasklist 命令
	cmd := exec.CommandContext(ctx, "tasklist")
	output, err := cmd.Output()
	if err != nil {
		return nil, err
	}

	lines := strings.Split(string(output), "\n")
	var processes []string
	for _, line := range lines {
		if strings.TrimSpace(line) != "" {
			processes = append(processes, line)
		}
	}

	return processes, nil
}

func readAPIStatus() interface{} {
	client := &http.Client{
		Timeout: 5 * time.Second,
	}

	resp, err := client.Get("http://127.0.0.1:5000/status")
	if err != nil {
		return map[string]interface{}{
			"error": err.Error(),
		}
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return map[string]interface{}{
			"error": err.Error(),
		}
	}

	var result interface{}
	if err := json.Unmarshal(body, &result); err != nil {
		return string(body)
	}

	return result
}

func readJSONFile(filename string) interface{} {
	data, err := os.ReadFile(filename)
	if err != nil {
		return map[string]interface{}{
			"error": err.Error(),
		}
	}

	var result interface{}
	if err := json.Unmarshal(data, &result); err != nil {
		return map[string]interface{}{
			"error": err.Error(),
		}
	}

	return result
}

func storeToRedis(status *StatusData) {
	if RedisService == nil {
		return
	}

	data, err := json.Marshal(status)
	if err != nil {
		return
	}

	// 存储到 Redis List
	RedisService.Client.Set(context.Background(), "qmt:liao", string(data), time.Duration(5*time.Second))
}