feat:新增python代码运算逻辑

controllers/base_from_calculate.go

@@ -764,6 +764,24 @@ func (this *CalculateController) Refresh() {
 				errMsg = "RefreshAllCalculateLjztbpj Err:" + err.Error()
 				break
 			}
+		case utils.DATA_SOURCE_PYTHON: //python代码运算
+			edbPythonCode, err := models.GetEdbPythonCodeById(edbInfo.EdbInfoId)
+			if err != nil {
+				errMsg = "获取python代码失败 Err:" + err.Error()
+				break
+			}
+			edbData, err := services.ExecPythonCode(edbInfo.EdbCode, edbPythonCode.PythonCode)
+			if err != nil {
+				br.Msg = "获取数据失败"
+				br.ErrMsg = "python代码获取数据失败，err:" + err.Error()
+				return
+			}
+			err = models.RefreshAllPythonEdb(edbInfo, edbData)
+			if err != nil && err.Error() != utils.ErrNoRow() {
+				errMsg = "RefreshAllPythonEdb Err:" + err.Error()
+				break
+			}
+
 		default:
 			br.Msg = "来源异常，请联系相关开发！"
 			br.ErrMsg = "来源异常，请联系相关开发"

controllers/base_from_python.go

@@ -0,0 +1,398 @@
+package controllers
+
+import (
+	"encoding/json"
+	"fmt"
+	"hongze/hongze_edb_lib/models"
+	"hongze/hongze_edb_lib/services"
+	"hongze/hongze_edb_lib/utils"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// PythonController 计算指标
+type PythonController struct {
+	BaseAuthController
+}
+
+// ExcePython
+// @Title 执行python代码接口
+// @Description 执行python代码接口
+// @Success 200 {object} models.ExecPythonEdbReq
+// @router /exec [post]
+func (this *PythonController) ExcePython() {
+	br := new(models.BaseResponse).Init()
+	defer func() {
+		this.Data["json"] = br
+		this.ServeJSON()
+	}()
+	//source := utils.DATA_SOURCE_WIND
+	var req models.ExecPythonEdbReq
+	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
+	if err != nil {
+		br.Msg = "参数解析异常！"
+		br.ErrMsg = "参数解析失败，Err:" + err.Error()
+		return
+	}
+
+	if req.PythonCode == "" {
+		br.Msg = "请填写python代码"
+		return
+	}
+	req.PythonCode, err = url.QueryUnescape(req.PythonCode)
+	if err != nil {
+		br.Msg = "python代码解析失败"
+		return
+	}
+	edbData, err := services.ExecPythonCode("test", req.PythonCode)
+	if err != nil {
+		br.Msg = "获取数据失败"
+		br.ErrMsg = "python代码获取数据失败，err:" + err.Error()
+		return
+	}
+	br.Ret = 200
+	br.Success = true
+	br.Msg = "保存成功"
+	br.Data = edbData
+	br.IsAddLog = true
+}
+
+// Add
+// @Title 编辑指标接口
+// @Description 编辑指标接口
+// @Success 200 {object} models.EditEdbInfoReq
+// @router /add [post]
+func (this *PythonController) Add() {
+	br := new(models.BaseResponse).Init()
+	defer func() {
+		this.Data["json"] = br
+		this.ServeJSON()
+	}()
+	//source := utils.DATA_SOURCE_WIND
+	var req models.AddPythonEdbReq
+	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
+	if err != nil {
+		br.Msg = "参数解析异常！"
+		br.ErrMsg = "参数解析失败，Err:" + err.Error()
+		return
+	}
+	req.EdbName = strings.Trim(req.EdbName, " ")
+	if req.EdbName == "" {
+		br.Msg = "指标名称不能为空"
+		return
+	}
+
+	if req.Frequency == "" {
+		br.Msg = "频率不能为空"
+		return
+	}
+
+	if req.Unit == "" {
+		br.Msg = "单位不能为空"
+		return
+	}
+
+	if req.ClassifyId <= 0 {
+		br.Msg = "请选择分类"
+		return
+	}
+
+	if req.PythonCode == "" {
+		br.Msg = "请填写python代码"
+		return
+	}
+
+	//加入缓存机制，避免创建同一个名称的指标 start
+	redisKey := fmt.Sprint("edb_info:python:add:", utils.DATA_SOURCE_PYTHON, ":", req.EdbName)
+	isExist := utils.Rc.IsExist(redisKey)
+	if isExist {
+		br.Msg = "指标正在处理,请勿重复提交"
+		return
+	} else {
+		//设置3分钟缓存
+		utils.Rc.SetNX(redisKey, 1, time.Second*300)
+		defer func() {
+			utils.Rc.Delete(redisKey)
+		}()
+	}
+
+	//获取指标数据
+	req.PythonCode, err = url.QueryUnescape(req.PythonCode)
+	if err != nil {
+		br.Msg = "python代码解析失败"
+		return
+	}
+	edbData, err := services.ExecPythonCode(req.EdbName, req.PythonCode)
+	if err != nil {
+		br.Msg = "获取数据失败"
+		br.ErrMsg = "python代码获取数据失败，err:" + err.Error()
+		return
+	}
+	var condition string
+	var pars []interface{}
+	condition += " AND edb_name=? "
+	pars = append(pars, req.EdbName)
+
+	count, err := models.GetEdbInfoCountByCondition(condition, pars)
+	if err != nil {
+		br.Msg = "判断指标名称是否存在失败"
+		br.ErrMsg = "判断指标名称是否存在失败,Err:" + err.Error()
+		return
+	}
+	if count > 0 {
+		br.Msg = "指标名称已存在，请重新填写"
+		br.ErrMsg = "指标名称已存在，请重新填写"
+		return
+	}
+	//指标code生成
+	randStr := utils.GetRandDigit(4)
+	edbCode := `C` + time.Now().Format("060102") + randStr
+
+	timestamp := strconv.FormatInt(time.Now().UnixNano(), 10)
+	uniqueCode := utils.MD5(utils.DATA_PREFIX + "_" + timestamp)
+	edbInfo := &models.EdbInfo{
+		Source:          utils.DATA_SOURCE_PYTHON,
+		SourceName:      "代码运算",
+		EdbCode:         edbCode,
+		EdbName:         req.EdbName,
+		EdbNameSource:   req.EdbName,
+		Frequency:       req.Frequency,
+		Unit:            req.Unit,
+		ClassifyId:      req.ClassifyId,
+		SysUserId:       req.AdminId,
+		SysUserRealName: req.AdminName,
+		CreateTime:      time.Now(),
+		ModifyTime:      time.Now(),
+		UniqueCode:      uniqueCode,
+		EdbType:         2,
+	}
+
+	edbInfoId, err := models.AddEdbInfo(edbInfo)
+	if err != nil {
+		br.Msg = "生成python运算指标失败"
+		br.Msg = "生成python运算指标失败,AddEdbInfo Err:" + err.Error()
+		return
+	}
+	edbInfo.EdbInfoId = int(edbInfoId)
+
+	//处理同名指标
+	{
+		edbNameList, err := models.GetEdbInfoByName(req.EdbName)
+		if err != nil {
+			br.Msg = "保存失败"
+			br.ErrMsg = "获取指标信息失败，Err:" + err.Error()
+			return
+		}
+		if len(edbNameList) >= 2 {
+			for _, v := range edbNameList {
+				edbName := v.EdbName + "(" + v.SourceName + ")"
+				err = models.ModifyEdbInfoNameSource(edbName, v.EdbInfoId)
+				if err != nil {
+					br.Msg = "保存失败"
+					br.ErrMsg = "修改指标名称失败，Err:" + err.Error()
+					return
+				}
+			}
+		}
+	}
+
+	// 存储python代码
+	edbPythonCode := &models.EdbPythonCode{
+		EdbInfoId:  int(edbInfoId),
+		EdbCode:    edbInfo.EdbCode,
+		PythonCode: req.PythonCode,
+		ModifyTime: time.Now(),
+		CreateTime: time.Now(),
+	}
+	_, err = models.AddEdbPythonCode(edbPythonCode)
+	if err != nil {
+		br.Msg = "生成python运算指标失败"
+		br.Msg = "生成python运算指标失败,存储python代码失败,AddEdbPythonCode Err:" + err.Error()
+		return
+	}
+
+	//pythonCode
+	err = models.AddPythonEdb(edbInfo.EdbInfoId, edbInfo.EdbCode, edbData)
+	if err != nil {
+		br.Msg = "生成计算指标失败"
+		br.Msg = "生成计算指标失败,AddPythonEdb Err:" + err.Error()
+		return
+	}
+	maxAndMinItem, err := models.GetEdbInfoMaxAndMinInfo(utils.DATA_SOURCE_PYTHON, edbCode)
+	if err != nil && err.Error() != utils.ErrNoRow() {
+		br.Msg = "生成计算指标失败"
+		br.Msg = "生成计算指标失败,GetEdbInfoMaxAndMinInfo Err:" + err.Error()
+		return
+	}
+
+	if maxAndMinItem != nil {
+		err = models.ModifyEdbInfoMaxAndMinInfo(int(edbInfoId), maxAndMinItem)
+	}
+	resp := models.AddEdbInfoResp{
+		EdbInfoId:  int(edbInfoId),
+		UniqueCode: uniqueCode,
+	}
+	br.Ret = 200
+	br.Success = true
+	br.Msg = "保存成功"
+	br.Data = resp
+	br.IsAddLog = true
+}
+
+// Edit
+// @Title 编辑指标接口
+// @Description 编辑指标接口
+// @Success 200 {object} models.EditEdbInfoReq
+// @router /edit [post]
+func (this *PythonController) Edit() {
+	br := new(models.BaseResponse).Init()
+	defer func() {
+		this.Data["json"] = br
+		this.ServeJSON()
+	}()
+	//source := utils.DATA_SOURCE_WIND
+	var req models.AddPythonEdbReq
+	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
+	if err != nil {
+		br.Msg = "参数解析异常！"
+		br.ErrMsg = "参数解析失败，Err:" + err.Error()
+		return
+	}
+	req.EdbName = strings.Trim(req.EdbName, " ")
+
+	if req.EdbInfoId <= 0 {
+		br.Msg = "指标id不能为空"
+		return
+	}
+	if req.EdbName == "" {
+		br.Msg = "指标名称不能为空"
+		return
+	}
+
+	if req.Frequency == "" {
+		br.Msg = "频率不能为空"
+		return
+	}
+
+	if req.Unit == "" {
+		br.Msg = "单位不能为空"
+		return
+	}
+
+	if req.ClassifyId <= 0 {
+		br.Msg = "请选择分类"
+		return
+	}
+
+	if req.PythonCode == "" {
+		br.Msg = "请填写python代码"
+		return
+	}
+
+	//加入缓存机制，避免创建同一个名称的指标 start
+	redisKey := fmt.Sprint("edb_info:python:edit:", utils.DATA_SOURCE_PYTHON, ":", req.EdbInfoId)
+	isExist := utils.Rc.IsExist(redisKey)
+	if isExist {
+		br.Msg = "指标正在处理,请勿重复提交"
+		return
+	} else {
+		//设置3分钟缓存
+		utils.Rc.SetNX(redisKey, 1, time.Second*300)
+		defer func() {
+			utils.Rc.Delete(redisKey)
+		}()
+	}
+
+	edbInfo, err := models.GetEdbInfoById(req.EdbInfoId)
+	if err != nil {
+		br.Msg = "获取指标信息失败"
+		br.ErrMsg = "获取指标信息失败，err:" + err.Error()
+		return
+	}
+
+	//获取指标数据
+	req.PythonCode, err = url.QueryUnescape(req.PythonCode)
+	if err != nil {
+		br.Msg = "python代码解析失败"
+		return
+	}
+	edbData, err := services.ExecPythonCode(req.EdbName, req.PythonCode)
+	if err != nil {
+		br.Msg = "获取数据失败"
+		br.ErrMsg = "python代码获取数据失败，err:" + err.Error()
+		return
+	}
+
+	// 判断指标名称是否重复
+	{
+		var condition string
+		var pars []interface{}
+		condition += " AND edb_name=? "
+		pars = append(pars, req.EdbName)
+
+		condition += " AND edb_info_id<>? "
+		pars = append(pars, req.EdbInfoId)
+
+		count, err := models.GetEdbInfoCountByCondition(condition, pars)
+		if err != nil {
+			br.Msg = "判断指标名称是否存在失败"
+			br.ErrMsg = "判断指标名称是否存在失败,Err:" + err.Error()
+			return
+		}
+		if count > 0 {
+			br.Msg = "指标名称已存在，请重新填写"
+			br.ErrMsg = "指标名称已存在，请重新填写"
+			return
+		}
+	}
+	edbInfo.EdbName = utils.TrimStr(req.EdbName)
+	edbInfo.Unit = req.Unit
+	edbInfo.ClassifyId = req.ClassifyId
+	edbInfo.Frequency = req.Frequency
+	err = edbInfo.Update([]string{"EdbName", "Unit", "ClassifyId", "Frequency"})
+	if err != nil {
+		br.Msg = "修改python运算指标失败"
+		br.Msg = "修改python运算指标失败,ModifyEdbInfo Err:" + err.Error()
+		return
+	}
+
+	// 修改存储python代码
+	edbPythonCode, err := models.GetEdbPythonCodeById(edbInfo.EdbInfoId)
+	edbPythonCode.PythonCode = req.PythonCode
+	edbPythonCode.ModifyTime = time.Now()
+	err = edbPythonCode.Update([]string{"PythonCode", "ModifyTime"})
+	if err != nil {
+		br.Msg = "修改python运算指标失败"
+		br.Msg = "修改python运算指标失败,存储python代码失败,EditEdbPythonCode Err:" + err.Error()
+		return
+	}
+
+	//刷新数据
+	err = models.RefreshAllPythonEdb(edbInfo, edbData)
+	if err != nil {
+		br.Msg = "生成计算指标失败"
+		br.Msg = "生成计算指标失败,AddPythonEdb Err:" + err.Error()
+		return
+	}
+	maxAndMinItem, err := models.GetEdbInfoMaxAndMinInfo(utils.DATA_SOURCE_PYTHON, edbInfo.EdbCode)
+	if err != nil && err.Error() != utils.ErrNoRow() {
+		br.Msg = "生成计算指标失败"
+		br.Msg = "生成计算指标失败,GetEdbInfoMaxAndMinInfo Err:" + err.Error()
+		return
+	}
+
+	if maxAndMinItem != nil {
+		err = models.ModifyEdbInfoMaxAndMinInfo(edbInfo.EdbInfoId, maxAndMinItem)
+	}
+	resp := models.AddEdbInfoResp{
+		EdbInfoId:  edbInfo.EdbInfoId,
+		UniqueCode: edbInfo.UniqueCode,
+	}
+	br.Ret = 200
+	br.Success = true
+	br.Msg = "修改成功"
+	br.Data = resp
+	br.IsAddLog = true
+}

models/base_from_python.go

@@ -0,0 +1,275 @@
+package models
+
+import (
+	"errors"
+	"fmt"
+	"github.com/beego/beego/v2/client/orm"
+	"github.com/shopspring/decimal"
+	"hongze/hongze_edb_lib/services"
+	"hongze/hongze_edb_lib/utils"
+	"strings"
+	"time"
+)
+
+// EdbDataPython python指标数据结构体
+type EdbDataPython struct {
+	EdbDataId     int `orm:"column(edb_data_id);pk"`
+	EdbInfoId     int
+	EdbCode       string
+	DataTime      string
+	Value         float64
+	CreateTime    time.Time
+	ModifyTime    time.Time
+	DataTimestamp int64
+}
+
+// AddPythonEdb 新增python运算指标
+func AddPythonEdb(edbInfoId int, edbCode string, item services.EdbDataFromPython) (err error) {
+	var errMsg string
+	o := orm.NewOrm()
+	defer func() {
+		if err != nil {
+			go utils.SendEmail(utils.APP_NAME_CN+"【"+utils.RunMode+"】"+"失败提醒", " 同花顺数据获取失败:err:"+errMsg, utils.EmailSendToUsers)
+		}
+	}()
+
+	var isAdd bool
+	addSql := ` INSERT INTO edb_data_python (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
+	for k, dateTimeStr := range item.Date {
+		//格式化时间
+		currentDate, tmpErr := time.Parse(utils.FormatDate, dateTimeStr)
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+		timestamp := currentDate.UnixNano() / 1e6
+		timestampStr := fmt.Sprintf("%d", timestamp)
+
+		//值
+		val := item.Value[k]
+		saveVal := utils.SubFloatToString(val, 20)
+		addSql += GetAddSql(fmt.Sprint(edbInfoId), edbCode, dateTimeStr, timestampStr, saveVal)
+
+		isAdd = true
+	}
+
+	if isAdd {
+		addSql = strings.TrimRight(addSql, ",")
+		_, err = o.Raw(addSql).Exec()
+		if err != nil {
+			errMsg = " tx.Exec Err :" + err.Error()
+			return
+		}
+	}
+	return
+}
+
+// EditPythonEdb 编辑python运算指标
+func EditPythonEdb(edbInfoId int, edbCode string, item services.EdbDataFromPython) (err error) {
+	var errMsg string
+	o := orm.NewOrm()
+	defer func() {
+		if err != nil {
+			go utils.SendEmail(utils.APP_NAME_CN+"【"+utils.RunMode+"】"+"失败提醒", " 同花顺数据获取失败:err:"+errMsg, utils.EmailSendToUsers)
+		}
+	}()
+
+	var isAdd bool
+	addSql := ` INSERT INTO edb_data_python (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
+	for k, dateTimeStr := range item.Date {
+		//格式化时间
+		currentDate, tmpErr := time.Parse(utils.FormatDate, dateTimeStr)
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+		timestamp := currentDate.UnixNano() / 1e6
+		timestampStr := fmt.Sprintf("%d", timestamp)
+
+		//值
+		val := item.Value[k]
+		saveVal := utils.SubFloatToString(val, 20)
+		addSql += GetAddSql(fmt.Sprint(edbInfoId), edbCode, dateTimeStr, timestampStr, saveVal)
+
+		isAdd = true
+	}
+
+	if isAdd {
+		addSql = strings.TrimRight(addSql, ",")
+		_, err = o.Raw(addSql).Exec()
+		if err != nil {
+			errMsg = " tx.Exec Err :" + err.Error()
+			return
+		}
+	}
+	return
+}
+
+// RefreshAllPythonEdb 刷新所有 python运算指标
+func RefreshAllPythonEdb(edbInfo *EdbInfo, item services.EdbDataFromPython) (err error) {
+	o := orm.NewOrm()
+	to, err := o.Begin()
+	if err != nil {
+		return
+	}
+	defer func() {
+		if err != nil {
+			fmt.Println("RefreshAllPythonEdb,Err:" + err.Error())
+			_ = to.Rollback()
+		} else {
+			_ = to.Commit()
+		}
+	}()
+
+	pythonDataMap := make(map[string]float64)
+	pythonDate := make([]string, 0)
+	for k, dateTimeStr := range item.Date {
+		pythonDataMap[dateTimeStr] = item.Value[k]
+		pythonDate = append(pythonDate, dateTimeStr)
+	}
+
+	//查询当前指标现有的数据
+	var condition string
+	var pars []interface{}
+	condition += " AND edb_info_id=? "
+	pars = append(pars, edbInfo.EdbInfoId)
+
+	//所有的数据
+	dataList, err := GetAllEdbDataPythonByEdbInfoId(edbInfo.EdbInfoId)
+	if err != nil {
+		return err
+	}
+
+	//待修改的指标数据map（index：日期，value：值）
+	updateEdbDataMap := make(map[string]float64)
+	removeDateList := make([]string, 0) //需要删除的日期
+	for _, v := range dataList {
+		currDataTime := v.DataTime
+		pythonData, ok := pythonDataMap[currDataTime]
+		if !ok {
+			// 如果python运算出来的数据中没有该日期，那么需要移除该日期的数据
+			removeDateList = append(removeDateList, currDataTime)
+		} else {
+			currValue, _ := decimal.NewFromFloat(pythonData).Truncate(4).Float64() //保留4位小数
+			//如果计算出来的值与库里面的值不匹配，那么就去修改该值
+			if v.Value != currValue {
+				//将计算后的数据存入待拼接指标map里面，以便后续计算
+				updateEdbDataMap[currDataTime] = currValue
+			}
+		}
+		//移除python指标数据中当天的日期
+		delete(pythonDataMap, currDataTime)
+	}
+
+	//sort.Strings(tbzEdbDataTimeList)
+
+	//新增的数据入库
+	{
+		addDataList := make([]*EdbDataPython, 0)
+		for dataTime, dataValue := range pythonDataMap {
+			//时间戳
+			currentDate, _ := time.Parse(utils.FormatDate, dataTime)
+			timestamp := currentDate.UnixNano() / 1e6
+
+			edbDataPython := &EdbDataPython{
+				EdbInfoId:     edbInfo.EdbInfoId,
+				EdbCode:       edbInfo.EdbCode,
+				DataTime:      dataTime,
+				Value:         dataValue,
+				CreateTime:    time.Now(),
+				ModifyTime:    time.Now(),
+				DataTimestamp: timestamp,
+			}
+			addDataList = append(addDataList, edbDataPython)
+		}
+		//最后如果还有需要新增的数据，那么就统一入库
+		if len(addDataList) > 0 {
+			_, tmpErr := o.InsertMulti(len(addDataList), addDataList)
+			if tmpErr != nil {
+				err = tmpErr
+				return
+			}
+		}
+	}
+
+	//删除已经不存在的累计同比拼接指标数据（由于同比值当日的数据删除了）
+	{
+		if len(removeDateList) > 0 {
+			removeDateStr := strings.Join(removeDateList, `","`)
+			removeDateStr = `"` + removeDateStr + `"`
+			//如果拼接指标变更了，那么需要删除所有的指标数据
+			tableName := GetEdbDataTableName(edbInfo.Source)
+			sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (%s) `, tableName, removeDateStr)
+
+			_, err = o.Raw(sql, edbInfo.EdbInfoId).Exec()
+			if err != nil {
+				err = errors.New("删除不存在的Python运算指标数据失败,Err:" + err.Error())
+				return
+			}
+		}
+	}
+
+	//修改现有的数据中对应的值
+	{
+		tableName := GetEdbDataTableName(edbInfo.Source)
+		for edbDate, edbDataValue := range updateEdbDataMap {
+			sql := fmt.Sprintf(` UPDATE  %s set value = ?,modify_time=now() WHERE edb_info_id = ? and data_time = ? `, tableName)
+
+			_, err = o.Raw(sql, edbDataValue, edbInfo.EdbInfoId, edbDate).Exec()
+			if err != nil {
+				err = errors.New("更新现有的Python运算指标数据失败,Err:" + err.Error())
+				return
+			}
+		}
+	}
+	return
+}
+
+// GetAllEdbDataPythonByEdbInfoId 根据指标id获取全部的数据
+func GetAllEdbDataPythonByEdbInfoId(edbInfoId int) (items []*EdbDataPython, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_data_python WHERE edb_info_id=? ORDER BY data_time DESC `
+	_, err = o.Raw(sql, edbInfoId).QueryRows(&items)
+	return
+}
+
+// EdbInfoPythonSaveReq 计算（运算）指标请求参数
+type EdbInfoPythonSaveReq struct {
+	AdminId          int    `description:"添加人id"`
+	AdminName        string `description:"添加人名称"`
+	EdbName          string `description:"指标名称"`
+	Frequency        string `description:"频率"`
+	Unit             string `description:"单位"`
+	ClassifyId       int    `description:"分类id"`
+	CalculateFormula string `description:"计算公式"`
+	EdbInfoIdArr     []struct {
+		EdbInfoId int    `description:"指标id"`
+		FromTag   string `description:"指标对应标签"`
+	}
+}
+
+// ExecPythonEdbReq 执行python代码运算指标的请求参数
+type ExecPythonEdbReq struct {
+	PythonCode string `description:"python代码"`
+}
+
+// AddPythonEdbReq 添加python代码运算指标的请求参数
+type AddPythonEdbReq struct {
+	AdminId    int    `description:"添加人id"`
+	AdminName  string `description:"添加人名称"`
+	EdbInfoId  int    `description:"指标id"`
+	EdbName    string `description:"指标名称"`
+	Frequency  string `description:"频度"`
+	Unit       string `description:"单位"`
+	ClassifyId int    `description:"分类id"`
+	PythonCode string `description:"python代码"`
+}
+
+// EdbPythonEdbReq 编辑python代码运算指标的请求参数
+type EdbPythonEdbReq struct {
+	EdbName    string `description:"指标名称"`
+	Frequency  string `description:"频度"`
+	Unit       string `description:"单位"`
+	ClassifyId int    `description:"分类id"`
+	PythonCode string `description:"python代码"`
+}

models/db.go

@@ -36,5 +36,7 @@ func init() {
 		new(EdbDataCalculateLjztbpj),
 		new(EdbInfo),
 		new(EdbInfoCalculateMapping),
+		new(EdbPythonCode),
+		new(EdbDataPython),
 	)
 }

models/edb_data_table.go

@@ -57,8 +57,12 @@ func GetEdbDataTableName(source int) (tableName string) {
 		tableName = "edb_data_calculate_ljztbpj"
 	case utils.DATA_SOURCE_LT:
 		tableName = "edb_data_lt"
+	case utils.DATA_SOURCE_COAL:
+		tableName = "edb_data_coal"
+	case utils.DATA_SOURCE_PYTHON:
+		tableName = "edb_data_python"
 	default:
 		tableName = ""
 	}
 	return
-}
+}

models/edb_info.go

@@ -84,6 +84,13 @@ func GetEdbInfoById(edbInfoId int) (item *EdbInfo, err error) {
 	return
 }
 
+// Update 更新EdbInfo信息
+func (edbInfo *EdbInfo) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(edbInfo, cols...)
+	return
+}
+
 // EdbInfoSearchData
 type EdbInfoSearchData struct {
 	EdbDataId int     `description:"数据ID"`

models/edb_python_code.go

@@ -0,0 +1,38 @@
+package models
+
+import (
+	"github.com/beego/beego/v2/client/orm"
+	"time"
+)
+
+// EdbPythonCode python指标运算代码
+type EdbPythonCode struct {
+	EdbPythonCodeId int    `orm:"column(edb_python_code_id);pk"`
+	EdbInfoId       int    `description:"指标id"`
+	EdbCode         string `description:"指标编码"`
+	PythonCode      string `description:"python代码"`
+	ModifyTime      time.Time
+	CreateTime      time.Time
+}
+
+// Update 更新EdbPythonCode信息
+func (edbPythonCode *EdbPythonCode) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(edbPythonCode, cols...)
+	return
+}
+
+// AddEdbPythonCode python指标运算代码
+func AddEdbPythonCode(item *EdbPythonCode) (lastId int64, err error) {
+	o := orm.NewOrm()
+	lastId, err = o.Insert(item)
+	return
+}
+
+// GetEdbPythonCodeById 根据指标id获取python代码
+func GetEdbPythonCodeById(edbInfoId int) (item *EdbPythonCode, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_python_code WHERE edb_info_id=? `
+	err = o.Raw(sql, edbInfoId).QueryRow(&item)
+	return
+}

routers/commentsRouter_controllers.go

@@ -178,6 +178,33 @@ func init() {
             Filters: nil,
             Params: nil})
 
+    beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"] = append(beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"],
+        beego.ControllerComments{
+            Method: "Add",
+            Router: "/add",
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
+    beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"] = append(beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"],
+        beego.ControllerComments{
+            Method: "Edit",
+            Router: "/edit",
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
+    beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"] = append(beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:PythonController"],
+        beego.ControllerComments{
+            Method: "ExcePython",
+            Router: "/exec",
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
     beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:ShController"] = append(beego.GlobalControllerRouter["hongze/hongze_edb_lib/controllers:ShController"],
         beego.ControllerComments{
             Method: "Add",

routers/router.go

@@ -90,6 +90,11 @@ func init() {
 				&controllers.GieController{},
 			),
 		),
+		beego.NSNamespace("/python",
+			beego.NSInclude(
+				&controllers.PythonController{},
+			),
+		),
 	)
 	beego.AddNamespace(ns)
 }

services/base_from_python.go

@@ -0,0 +1,138 @@
+package services
+
+import (
+	"encoding/json"
+	"fmt"
+	"hongze/hongze_edb_lib/utils"
+	"os"
+	"os/exec"
+	"strings"
+)
+
+func Test() (err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("err:", err)
+		}
+	}()
+	///usr/local/bin/python3.9
+	//exec.Command("bin/bash", "-c")
+	//cmd := exec.Command("/usr/local/bin/python3.9", "-c", "/Users/roc/go/src/hongze/hongze_edb_lib/test2.py")
+	//cmd := exec.Command("python3", "-c", "./test2.py")
+	cmd := exec.Command("python3", "/Users/roc/go/src/hongze/hongze_edb_lib/test2.py")
+	outputByte, err := cmd.Output()
+	//fmt.Println(err)
+	fmt.Println("start")
+	if err != nil {
+		return
+	}
+	fmt.Println(string(outputByte))
+	arr := strings.Split(string(outputByte), "result=")
+	arrLen := len(arr)
+	//fmt.Println(arr)
+	if arrLen <= 1 {
+		err = fmt.Errorf("python运算结果异常")
+		return
+	}
+	resultStr := arr[arrLen-1]
+	fmt.Println(resultStr)
+
+	var dataMap map[string]float64
+	json.Unmarshal([]byte(resultStr), &dataMap)
+	fmt.Println(dataMap)
+	//i, err := python3.Py_Main(os.Args)
+	//if err != nil {
+	//	fmt.Printf("error launching the python interpreter: %s\n", err)
+	//	os.Exit(1)
+	//}
+	////commStr := "#!/usr/bin/python\n# -*- coding: UTF-8 -*-\n\nimport pymysql\nimport pandas as pd\nfrom test_bak import sql_config\n\ndb = pymysql.connect(**sql_config)\ndb.autocommit(1)\ncursor = db.cursor()\npandas_fetch_all = pd.read_sql\n\n# 返回数据\nresult = {}\n\n# 格式化返回数据\ndef format_data(data: pd.DataFrame,\n                index_str: str = \"data_time\",\n                value_str: str = \"value\"\n                ) -> dict:\n    \"\"\"\n        Parameters\n        ----------\n        data : pandas的DataFrame数据结构.\n        index_str : 对象下标字符串，在pandas的DataFrame中的列名.\n        value_str : 对象值字符串，在pandas的DataFrame中的列名\n\n        Returns\n        -------\n        DataFrame or Iterator[DataFrame]\n        例子：{'2007-01-09': 3220.0, '2007-01-10': 3230.0}\n\n        \"\"\"\n    tmp_result = {}\n    for num in range(1, data.index.size):  # 迭代 所有的指标\n        tmp_result[data[index_str][num]] = data[value_str][num]\n    return tmp_result\n\n\ndef query():\n    edb_code = 's0033227'\n    data_time = '2002-03-17'\n    # field_name = '平均温度'\n    sql1 = f\"\"\"SELECT data_time,`value` FROM edb_data_wind WHERE edb_code = '{edb_code}' and data_time > '{data_time}' ;\"\"\"\n    raw = pandas_fetch_all(sql1, db)\n    raw['data_time_str'] = raw['data_time'].apply(lambda x: x.strftime(\"%Y-%m-%d\"))\n    format_result = format_data(raw, \"data_time_str\", \"value\")\n    print(format_result)\n    return format_result\n\n\nresult = query()\ndb.close()\n"
+	////i := python3.PyRun_SimpleString(commStr)
+	//fmt.Println(i)
+	return
+}
+
+// EdbDataFromPython 通过python代码获取到的指标数据
+type EdbDataFromPython struct {
+	Date  map[int]string  `json:"date"`
+	Value map[int]float64 `json:"value"`
+}
+
+// ExecPythonCode 执行Python代码
+func ExecPythonCode(edbCode, reqCode string) (dataMap EdbDataFromPython, err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("err:", err)
+		}
+	}()
+
+	//获取python文件的绝对地址
+	pythonFile, err := getPythonFileAbsolutePath(edbCode)
+	if err != nil {
+		return
+	}
+	pthonCodeStr := getPythonFrontStr() + reqCode + getPythonLaterStr()
+	fileHandle, err := os.OpenFile(pythonFile, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0766)
+	if err != nil {
+		return
+	}
+	//defer func() {
+	//	os.Remove(pythonFile)
+	//}()
+	_, err = fileHandle.Write([]byte(pthonCodeStr))
+	if err != nil {
+		return
+	}
+	fileHandle.Close()
+
+	cmd := exec.Command("python3", pythonFile)
+	outputByte, err := cmd.Output()
+	//fmt.Println(err)
+	if err != nil {
+		return
+	}
+	//fmt.Println(string(outputByte))
+	arr := strings.Split(string(outputByte), "result=")
+	arrLen := len(arr)
+	//fmt.Println(arr)
+	if arrLen <= 1 {
+		err = fmt.Errorf("python运算结果异常")
+		return
+	}
+	resultStr := arr[arrLen-1]
+	fmt.Println(resultStr)
+
+	json.Unmarshal([]byte(resultStr), &dataMap)
+	//fmt.Println(dataMap)
+	return
+}
+
+// getPythonFileAbsolutePath 获取python文件的绝对地址
+func getPythonFileAbsolutePath(edbCode string) (pythonFile string, err error) {
+	uploadDir := utils.STATIC_DIR + "python/"
+	err = os.MkdirAll(uploadDir, 0766)
+	if err != nil {
+		return
+	}
+	pythonFile = uploadDir + fmt.Sprint(edbCode, "_", utils.GetRandDigit(16), ".py")
+	if utils.RunMode != "release" {
+		dir, tmpErr := os.Getwd()
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+		pythonFile = dir + "/" + pythonFile
+	}
+	return
+}
+
+// getPythonFrontStr 获取python前面的代码
+func getPythonFrontStr() string {
+	//return "#!/usr/bin/python\n# -*- coding: UTF-8 -*-\nimport json\n\nimport pymysql\nimport pandas as pd\n\nsql_config = {\n    'host': 'rm-uf67kg347rhjfep5c1o.mysql.rds.aliyuncs.com',\n    'port': 3306,#主机号\n    'user': 'hz_technology',#账户名\n    'passwd': 'hongze@2021',#密码\n    'db': 'test_hz_data',\n    'charset': 'utf8mb4',\n    'cursorclass': pymysql.cursors.DictCursor\n}\n\ndb = pymysql.connect(**sql_config)\ndb.autocommit(1)\ncursor = db.cursor()\npandas_fetch_all = pd.read_sql\n\n# 返回数据\nresult = {}\n\n# 格式化返回数据\ndef format_data(data: pd.DataFrame,\n                index_str: str = \"data_time\",\n                value_str: str = \"value\"\n                ) -> pd.DataFrame:\n    \"\"\"\n        Parameters\n        ----------\n        data : pandas的DataFrame数据结构.\n        index_str : 对象下标字符串，在pandas的DataFrame中的列名.\n        value_str : 对象值字符串，在pandas的DataFrame中的列名\n\n        Returns\n        -------\n        DataFrame or Iterator[DataFrame]\n        例子：{'2007-01-09': 3220.0, '2007-01-10': 3230.0}\n\n        \"\"\"\n    index_list = []  # 空列表\n    value_list = []  # 空列表\n\n    for num in range(1, data.index.size):  # 迭代 所有的指标\n        index_list.append(data[index_str][num])\n        value_list.append(data[value_str][num])\n\n    tmp_data = {\n        \"date\": index_list,\n        \"value\": value_list\n    }\n    pd_data = pd.DataFrame(tmp_data)\n    # print(pd_data)\n    return pd_data\n"
+	str := fmt.Sprintf("#!/usr/bin/python\n# -*- coding: UTF-8 -*-\nimport json\n\nimport pymysql\nimport pandas as pd\n\nsql_config = {\n    'host': '%s',\n    'port': 3306,#主机号\n    'user': '%s',#账户名\n    'passwd': '%s',#密码\n    'db': '%s',\n    'charset': 'utf8mb4',\n    'cursorclass': pymysql.cursors.DictCursor\n}\n\ndb = pymysql.connect(**sql_config)\ndb.autocommit(1)\ncursor = db.cursor()\npandas_fetch_all = pd.read_sql\n\n# 返回数据\nresult = {}\n\n# 格式化返回数据\ndef format_data(data: pd.DataFrame,\n                index_str: str = \"data_time\",\n                value_str: str = \"value\"\n                ) -> pd.DataFrame:\n    \"\"\"\n        Parameters\n        ----------\n        data : pandas的DataFrame数据结构.\n        index_str : 对象下标字符串，在pandas的DataFrame中的列名.\n        value_str : 对象值字符串，在pandas的DataFrame中的列名\n\n        Returns\n        -------\n        DataFrame or Iterator[DataFrame]\n        例子：{'2007-01-09': 3220.0, '2007-01-10': 3230.0}\n\n        \"\"\"\n    index_list = []  # 空列表\n    value_list = []  # 空列表\n\n    for num in range(1, data.index.size):  # 迭代 所有的指标\n        index_list.append(data[index_str][num])\n        value_list.append(data[value_str][num])\n\n    tmp_data = {\n        \"date\": index_list,\n        \"value\": value_list\n    }\n    pd_data = pd.DataFrame(tmp_data)\n    # print(pd_data)\n    return pd_data\n", utils.PYTHON_MYSQL_HOST, utils.PYTHON_MYSQL_USER, utils.PYTHON_MYSQL_PASSWD, utils.PYTHON_MYSQL_DB)
+	return str
+}
+
+// getPythonFrontStr 获取python结尾的代码
+func getPythonLaterStr() string {
+	return "\nprint(\"result=\",result.to_json())\ndb.close()"
+}

utils/config.go

@@ -8,15 +8,23 @@ import (
 )
 
 var (
-	RunMode               string //运行模式
-	MYSQL_URL             string //数据库连接
-	MYSQL_URL_EDB         string
-	MYSQL_URL_GL          string
+	RunMode       string //运行模式
+	MYSQL_URL     string //数据库连接
+	MYSQL_URL_EDB string
+	MYSQL_URL_GL  string
+
+	PYTHON_MYSQL_HOST   string //python数据库链接主机地址
+	PYTHON_MYSQL_USER   string //python数据库链接账号
+	PYTHON_MYSQL_PASSWD string //python数据库链接密码
+	PYTHON_MYSQL_DB     string //python数据库链接数据库名
 
 	REDIS_CACHE string       //缓存地址
 	Rc          *cache.Cache //redis缓存
 	Re          error        //redis错误
 )
+var (
+	STATIC_DIR string
+)
 
 func init() {
 	tmpRunMode, err := web.AppConfig.String("run_mode")
@@ -48,6 +56,12 @@ func init() {
 	MYSQL_URL_EDB = config["mysql_url_edb"]
 	MYSQL_URL_GL = config["mysql_url_gl"]
 
+	//python链接
+	PYTHON_MYSQL_HOST = config["python_mysql_host"]
+	PYTHON_MYSQL_USER = config["python_mysql_user"]
+	PYTHON_MYSQL_PASSWD = config["python_mysql_passwd"]
+	PYTHON_MYSQL_DB = config["python_mysql_db"]
+
 	REDIS_CACHE = config["beego_cache"]
 	if len(REDIS_CACHE) <= 0 {
 		panic("redis链接参数没有配置")
@@ -60,8 +74,11 @@ func init() {
 
 	if RunMode == "release" {
 		Hz_Ths_Data_Url = "http://172.19.173.231:7000/"
+		STATIC_DIR = "/home/static/imgs/"
 	} else {
 		Hz_Ths_Data_Url = "http://139.196.136.213:7000/"
+
+		STATIC_DIR = "static/imgs/"
 	}
 
 }

utils/constants.go

@@ -53,6 +53,8 @@ const (
 	DATA_SOURCE_CALCULATE_ZJPJ                  //直接拼接->23
 	DATA_SOURCE_CALCULATE_LJZTBPJ               //累计值同比拼接->24
 	DATA_SOURCE_LT                              //路透->25
+	DATA_SOURCE_COAL                            //煤炭网->26
+	DATA_SOURCE_PYTHON                          //python代码->27
 )
 
 //基础数据初始化日期