eta_index_lib/services/base_from_python.go

package services

import (
	"bytes"
	"encoding/json"
	"eta/eta_index_lib/models"
	"eta/eta_index_lib/utils"
	"fmt"
	"github.com/shopspring/decimal"
	"os"
	"os/exec"
	"reflect"
	"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/eta/eta_index_lib/test2.py")
	//cmd := exec.Command("python3", "-c", "./test2.py")
	cmd := exec.Command("python3", "/Users/roc/go/src/eta/eta_index_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
}

// EdbDataStrFromPython 通过python代码获取到的指标数据（interface数据）
type EdbDataStrFromPython struct {
	Date  map[int]string      `json:"date"`
	Value map[int]interface{} `json:"value"`
}

// ExecPythonCode 执行Python代码
func ExecPythonCode(edbCode, reqCode string) (dataMap models.EdbDataFromPython, err error, errMsg string) {
	defer func() {
		if err != nil {
			fmt.Println("err:", err)
		}
	}()

	codeStr := strings.ToLower(reqCode)
	// 限制python代码的执行，如果包含特殊的关键词，则报错
	if strings.Contains(codeStr, "drop table") || strings.Contains(codeStr, "delete") || strings.Contains(codeStr, "update") || strings.Contains(codeStr, "alter") || strings.Contains(codeStr, "truncate") {
		err = fmt.Errorf("不支持该命令")
		return
	}

	//获取python文件的绝对地址
	pythonFile, err := getPythonFileAbsolutePath(edbCode)
	if err != nil {
		return
	}
	pythonCodeStr := 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(pythonCodeStr))
	if err != nil {
		return
	}
	fileHandle.Close()

	cmd := exec.Command(utils.PYTHON_PATH, pythonFile)
	var out, errMsgOut bytes.Buffer
	cmd.Stdout = &out
	cmd.Stderr = &errMsgOut
	err = cmd.Run()
	if err != nil {
		errMsg = errMsgOut.String()
		return
	}

	outputByte := out.String()
	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 dataMapStr EdbDataStrFromPython

	//先将value转为interface
	err = json.Unmarshal([]byte(resultStr), &dataMapStr)
	if err != nil {
		return
	}

	dataMap.Date = make(map[int]string)
	dataMap.Value = make(map[int]float64)

	//将value为nil的给过滤掉
	i := 0
	lenData := len(dataMapStr.Date)
	for k := 0; k < lenData; k++ {
		date := dataMapStr.Date[k]
		tmpValue := dataMapStr.Value[k]
		if reflect.TypeOf(tmpValue) != nil {
			if reflect.TypeOf(tmpValue).Kind() == reflect.Float64 {
				dataMap.Date[i] = date
				dataMap.Value[i], _ = decimal.NewFromFloat(reflect.ValueOf(tmpValue).Float()).Truncate(4).Float64() //保留4位小数
				i++
			}
		}
	}

	//err = 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\nfrom pymongo import MongoClient\nimport pytz\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", utils.PYTHON_MYSQL_HOST, utils.PYTHON_MYSQL_USER, utils.PYTHON_MYSQL_PASSWD, utils.PYTHON_MYSQL_DB)

	// mongo部分
	if utils.PYTHON_MONGO_HOST != `` {
		str += getPythonFront2Str()
	}

	// 格式化数据部分
	str += getPythonFront3Str()

	return str
}

// getPythonFront2Str
// @Description: mongo部分
// @author: Roc
// @datetime 2024-05-07 18:38:03
// @return string
func getPythonFront2Str() string {
	str := fmt.Sprintf("\nfrom pymongo import MongoClient\nfrom dateutil.tz import tzlocal\n# MongoDB 连接配置\nmongo_config = {\n    'host': '%s',  # 替换为你的 MongoDB 连接字符串\n    'database': '%s',  # 替换为你的数据库名\n    'collection': '%s',  # 替换为你的集合名\n    'auth_mechanism': '%s'  # 替换为你的认证机制\n}\n\n# 创建 MongoClient 并连接到数据库\nclient = MongoClient(mongo_config['host'], authMechanism=mongo_config['auth_mechanism'])\nmgo_db = client[mongo_config['database']]\ncollection = mgo_db[mongo_config['collection']]\n\n# 定义时区\nutc_tz = pytz.utc\nlocal_tz = tzlocal()  # 本地时区", utils.PYTHON_MONGO_HOST, utils.PYTHON_MONGO_DATABASE, "edb_data_business", utils.PYTHON_MONGO_AUTH_MECHANISM)
	return str
}

// getPythonFront3Str
// @Description: 第三部分
// @author: Roc
// @datetime 2024-05-07 18:37:56
// @return string
func getPythonFront3Str() string {
	str := fmt.Sprintf("\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(0, data.index.size):  # 迭代 所有的指标\n        date=data[index_str][num]\n        if isinstance(date,str) is False:\n            index_list.append(data[index_str][num].strftime(\"%%Y-%%m-%%d\"))\n        else:\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    return pd_data\n\n")
	return str
}

// getPythonLaterStr 获取python结尾的代码
func getPythonLaterStr() string {
	return "\n\nprint(\"result=\", result.to_json())\ndb.close()"
}