package services
import (
"bytes"
"encoding/json"
"fmt"
"github.com/shopspring/decimal"
"hongze/hongze_edb_lib/utils"
"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/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"`
}
// 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 EdbDataFromPython, err error, errMsg string) {
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(utils.PYTHON_PATH, pythonFile)
var out, errMsgOut bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &errMsgOut
err = cmd.Start()
if err != nil {
errMsg = errMsgOut.String()
}
err = cmd.Wait()
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 := 1; 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\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\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 "\n\nprint(\"result=\", result.to_json())\ndb.close()"
}