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()" }