eta_crawler/services/commodity_trade_dalian.go

package services

import (
	"bytes"
	"eta/eta_crawler/models"
	"eta/eta_crawler/services/alarm_msg"
	"eta/eta_crawler/utils"
	"fmt"
	"github.com/PuerkitoBio/goquery"
	"github.com/mozillazg/go-pinyin"
	"io/ioutil"
	"log"
	"mime/multipart"
	"net/http"
	"strconv"
	"strings"
	"time"
)

type SearchList struct {
	VarietyName string `description:"商品名称"`
	CarietyCode string `description:"商品名称对应的编码"`
	List        []SearchContractId
	ListSearch  []*SearchContractId
}

type SearchContractId struct {
	ContractId string `description:"商品类型"`
}

// 同步 N天 之内的数据
func SyncRankingFromDalianDo() {

	for i := 7; i >= 0; i-- {
		SyncRankingFromDalianSearch(i)
	}
}

// 大连交易所持单排名
func SyncRankingFromDalianSearch(dayNum int) (err error) {
	fmt.Println("start")
	n := utils.GetRandInt(10, 120)
	time.Sleep(time.Duration(n) * time.Second)
	defer func() {
		if err != nil {
			fmt.Println("RefreshDataFromDalian Err:" + err.Error())
			msg := "失败提醒" + "RefreshDataFromDalian SyncRankingFromDalianSearch ErrMsg:" + err.Error()
			go alarm_msg.SendAlarmMsg(msg, 3)
			//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "RefreshDataFromDalian ErrMsg:"+err.Error(), utils.EmailSendToUsers)
		}
	}()
	//定义爬取时间
	endDate := time.Now().AddDate(0, 0, -dayNum).Format(utils.FormatDateTime)
	//endDate := time.Now().Format(utils.FormatDateTime)
	timeDate := utils.StrTimeToTime(endDate)
	currDate := timeDate.Format(utils.FormatDateUnSpace)
	year := timeDate.Year()
	month := timeDate.Format("01")
	var dayStr string
	day := timeDate.Day()
	if day < 10 {
		dayStr = "0" + strconv.Itoa(day)
	} else {
		dayStr = strconv.Itoa(day)
	}
	monthNum, _ := strconv.Atoi(month)
	month = strconv.Itoa(monthNum - 1) //获取时月份需要减一
	list, err := models.GetBaseFromTradeDalianDataList(timeDate.Format(utils.FormatDate))
	if err != nil {
		fmt.Println(err)
		return err
	}
	listDataMap := make(map[string]int)
	for _, v := range list {
		listDataMap[v.DealShortName+v.ClassifyType+v.DataTime] = v.BaseFromTradeDalianIndexId
	}

	var ContractId string
	var CarietyCode string
	var VarietyName string
	//模拟form表单请求
	url := "http://www.dce.com.cn/publicweb/quotesdata/memberDealPosiQuotes.html"
	method := "POST"
	payload := &bytes.Buffer{}
	writer := multipart.NewWriter(payload)
	_ = writer.WriteField("memberDealPosiQuotes.variety", CarietyCode)
	_ = writer.WriteField("memberDealPosiQuotes.trade_type", "0")
	_ = writer.WriteField("year", strconv.Itoa(year))
	_ = writer.WriteField("month", month)
	_ = writer.WriteField("day", dayStr)
	_ = writer.WriteField("contract.contract_id", ContractId)
	_ = writer.WriteField("contract.variety_id", CarietyCode)
	_ = writer.WriteField("currDate", currDate)
	err = writer.Close()
	if err != nil {
		utils.FileLog.Info("获取指标失败：" + currDate + VarietyName + ContractId)
		return err
	}
	client := &http.Client{}
	req, err := http.NewRequest(method, url, payload)
	if err != nil {
		return err
	}
	req.Header.Add("Cookie", "JSESSIONID=36ACF02A59227A3854F9D5D5E2FB5F2E; WMONID=R5ojcAIIcx-")
	req.Header.Set("Content-Type", writer.FormDataContentType())
	res, err := client.Do(req)
	if err != nil {
		fmt.Println("post SyncRankingFromDalianSearch err:"+err.Error())
		utils.FileLog.Info("post SyncRankingFromDalianSearch err:"+err.Error())
		return err
	}
	defer res.Body.Close()
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		msg := "失败提醒" + "RefreshDataFromDalian ErrMsg:" + err.Error() + "获取指标失败：" + currDate + VarietyName + ContractId
		go alarm_msg.SendAlarmMsg(msg, 3)
		//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "RefreshDataFromDalian ErrMsg:"+err.Error()+"获取指标失败："+currDate+VarietyName+ContractId, utils.EmailSendToUsers)
		return err
	}
	exitProductMaps, _, varietyArrMaps := DoSearch(string(body))
	var items []*SearchList
	for k, v := range exitProductMaps {
		item := new(SearchList)
		item.VarietyName = v
		item.CarietyCode = varietyArrMaps[k]
		htmlBody, err := GetDalianHtmlBody(dayNum, "", item.CarietyCode, item.VarietyName)
		if err != nil {
			if strings.Contains(err.Error(),"timed out") {
				continue
			}
			return err
		}
		listContractId := DoSearchMap(htmlBody)
		item.ListSearch = listContractId
		items = append(items, item)
	}
	//for k, v := range items {
	//	for _, v2 := range v.ListSearch {
	//		fmt.Println(currDate, v.VarietyName, v2.ContractId, v.CarietyCode, k)
	//	}
	//}
	SyncRankingFromDalian(dayNum, items)
	return err
}

// 处理搜索条件初始
func DoSearch(body string) (exitProductMaps, exitContractIdMaps, varietyArrMaps map[int]string) {
	var str string
	str = body
	doc, err := goquery.NewDocumentFromReader(strings.NewReader(str))
	if err != nil {
		log.Fatal(err)
	}
	exitProductMap := make(map[int]string)
	exitContractIdMap := make(map[int]string)
	varietyArrMap := make(map[int]string)
	//var productName string
	ul := doc.Find(".selBox ul")
	var pNum int
	var cidNum int
	var vNum int
	ul.Each(func(i int, s *goquery.Selection) {
		//解析标签
		//fmt.Println(i, s.Text())
		ulTxt := s.Text()
		//fmt.Println(ulTxt)
		if ulTxt != "" && (i == 0 || i == 2) {
			ulTxtArr := strings.Split(ulTxt, "\n")
			for _, v := range ulTxtArr {
				v = strings.Replace(v, " ", "", -1)
				v = strings.Replace(v, "\n", "", -1)
				v = strings.Replace(v, "	", "", -1)
				if v != "" && len(v) > 0 {
					exitProductMap[pNum] = v
					pNum++
				}
			}
		}
		if ulTxt != "" && i == 3 {
			//fmt.Println(ulTxt)
			cidTxtArr := strings.Split(ulTxt, " ")
			for _, v := range cidTxtArr {
				v = strings.Replace(v, "\n", "", -1)
				v = strings.Replace(v, " ", "", -1)
				v = strings.Replace(v, "	", "", -1)
				if v != "" {
					exitContractIdMap[cidNum] = v
					cidNum++
				}
			}
		}
	})
	varietyArr := strings.Split(str, "onclick=\"javascript:setVariety('")
	for _, v := range varietyArr {
		strnum := strings.Index(v, "');")
		if strnum > 0 {
			varietyStr := v[0:strnum]
			if len(varietyStr) < 10 {
				//fmt.Println(strnum, varietyStr)
				varietyArrMap[vNum] = varietyStr
				vNum++
			}
		}
	}
	exitProductMaps = exitProductMap
	exitContractIdMaps = exitContractIdMap
	varietyArrMaps = varietyArrMap
	return
}

// 处理搜索条件
func DoSearchMap(body string) (items []*SearchContractId) {
	var str string
	str = body
	doc, err := goquery.NewDocumentFromReader(strings.NewReader(str))
	if err != nil {
		log.Fatal(err)
	}
	exitContractIdMap := make(map[int]string)
	ul := doc.Find(".selBox ul")
	var cidNum int
	ul.Each(func(i int, s *goquery.Selection) {
		//解析标签
		ulTxt := s.Text()
		if ulTxt != "" && i == 3 {
			cidTxtArr := strings.Split(ulTxt, " ")
			for _, v := range cidTxtArr {
				v = strings.Replace(v, "\n", "", -1)
				v = strings.Replace(v, " ", "", -1)
				v = strings.Replace(v, "	", "", -1)
				if v != "" {
					exitContractIdMap[cidNum] = v
					cidNum++
				}
			}
		}
	})
	for _, v := range exitContractIdMap {
		item := new(SearchContractId)
		item.ContractId = v
		items = append(items, item)
	}
	return items
}

// 处理解析Html
func DoHtml(body, name, contractId string, dateTime time.Time, listDataMap map[string]int, listIndexCodeMap map[string]string, listDataMapVal map[string]int) (err error) {
	defer func() {
		if err != nil {
			fmt.Println("RefreshDataFromDaLian Err:" + err.Error())
			msg := "失败提醒" + "RefreshDataFromDalian ErrMsg:" + err.Error()
			go alarm_msg.SendAlarmMsg(msg, 3)
			//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "RefreshDataFromDaLian ErrMsg:"+err.Error(), utils.EmailSendToUsers)
		}
	}()
	str := body
	doc, err := goquery.NewDocumentFromReader(strings.NewReader(str))
	if err != nil {
		log.Fatal(err)
	}
	var isAdd bool
	addSql := `  INSERT INTO base_from_trade_dalian_index(rank,deal_short_name,deal_name,deal_code,deal_value,buy_short_name,deal_change,buy_name,buy_code,buy_value,buy_change,sold_short_name,sold_name,sold_code,sold_value,sold_change,frequency,classify_name,classify_type,create_time,modify_time,data_time) values  `
	table := doc.Find("table")
	var rank, shortName, dealValue, dealChange, buyName, buyValue, buyChange, soldName, soldValue, soldChange string
	table.Find("tr").Each(func(i int, tr *goquery.Selection) {
		tds := tr.Find("td")
		//fmt.Println(tds.Length(), "长度：", i)
		if tds.Length() == 0 || tds.Length() == 7 {
			tdText := tds.Text()
			utils.FileLog.Info(tdText)
		} else {
			item := new(models.BaseFromTradeDalianIndex)
			tds.Each(func(tk int, td *goquery.Selection) {
				tdText := td.Text()
				if tk == 0 { //名次
					rank = tdText
				}
				if tk == 1 { //会员简称
					shortName = tdText
				}
				if tk == 2 { //成交量
					dealValue = strings.Replace(tdText, ",", "", -1)
				}
				if tk == 3 { //增减
					dealChange = strings.Replace(tdText, ",", "", -1)
				}
				if tk == 5 { //会员简称
					buyName = tdText
				}
				if tk == 6 { //持买单量
					buyValue = strings.Replace(tdText, ",", "", -1)
				}
				if tk == 7 { //增减
					buyChange = strings.Replace(tdText, ",", "", -1)
				}

				if tk == 9 { //会员简称
					soldName = tdText
				}
				if tk == 10 { //持卖单量
					soldValue = strings.Replace(tdText, ",", "", -1)
				}
				if tk == 11 { //增减
					soldChange = strings.Replace(tdText, ",", "", -1)
				}
			})
			item.Rank = rank
			item.DealShortName = shortName
			item.DealName = shortName + "_" + contractId + "_成交量"

			item.DealValue = dealValue
			item.BuyShortName = buyName
			item.DealChange = dealChange
			item.BuyName = buyName + "_" + contractId + "_持买单量"

			item.BuyValue = buyValue
			item.BuyChange = buyChange
			item.SoldShortName = soldName
			item.SoldName = soldName + "_" + contractId + "_持卖单量"
			item.SoldValue = soldValue
			item.SoldChange = soldChange
			item.Frequency = "日度"
			item.ClassifyName = name
			item.ClassifyType = contractId
			item.CreateTime = time.Now().Format(utils.FormatDateTime)
			item.ModifyTime = time.Now().Format(utils.FormatDateTime)
			item.DataTime = dateTime.Format(utils.FormatDate)
			//处理指标Id
			if i != 23 {
				if val, ok := listIndexCodeMap[item.DealName]; ok {
					item.DealCode = val
				} else {
					item.DealCode = GetIndexCodeGeneratorPinYing(shortName, item.DealName, contractId, "deal", "DL")
				}
				if val, ok := listIndexCodeMap[item.BuyName]; ok {
					item.BuyCode = val
				} else {
					item.BuyCode = GetIndexCodeGeneratorPinYing(buyName, item.BuyName, contractId, "buy", "DL")
				}
				if val, ok := listIndexCodeMap[item.SoldName]; ok {
					item.SoldCode = val
				} else {
					item.SoldCode = GetIndexCodeGeneratorPinYing(soldName, item.SoldName, contractId, "sold", "DL")
				}
			}

			if i == 23 {
				item.Rank = "999"
				item.DealName = "top20_" + contractId + "_成交量(手)"
				if val, ok := listIndexCodeMap[item.DealName]; ok {
					item.DealCode = val
				} else {
					item.DealCode = GetIndexCodeGeneratorPinYing("top20", item.DealName, contractId, "deal", "DL")
				}
				item.BuyName = "top20_" + contractId + "_持买单量(手)"
				if val, ok := listIndexCodeMap[item.BuyName]; ok {
					item.BuyCode = val
				} else {
					item.BuyCode = GetIndexCodeGeneratorPinYing("top20", item.BuyName, contractId, "buy", "DL")
				}
				item.SoldName = "top20_" + contractId + "_持卖单量(手)"

				if val, ok := listIndexCodeMap[item.SoldName]; ok {
					item.SoldCode = val
				} else {
					item.SoldCode = GetIndexCodeGeneratorPinYing("top20", item.SoldName, contractId, "sold", "DL")
				}
			}
			if val, ok := listDataMap[item.DealShortName+item.ClassifyType+item.DataTime]; !ok {
				addSql += models.GetAddSql(item)
				isAdd = true
			} else {
				//更新
				if listDataMapVal[item.DealValue+item.BuyValue+item.SoldValue] != val {
					err := models.UpdateBaseFromTradeDalianIndex(item, val)
					if err != nil {
						fmt.Println("UpdateBaseFromTradeDalianIndex err:", err)
					}
				}
			}

		}
	})
	addSql = strings.TrimRight(addSql, ",")
	if isAdd {
		err = models.RefreshEdbDataByDaLian(addSql)
		if err != nil {
			return err
		}
	}
	return
}

func DlIndexCodeGenerator(indexName, suffix string) (ineIndexCode string) {
	ineIndexCode = fmt.Sprintf("DL%s", strconv.FormatInt(time.Now().UnixNano(), 10)+suffix)
	err := models.AddBaseFromTradeMapping(indexName, ineIndexCode, "DL")
	if err != nil {
		fmt.Println("add Code err:", err)
	}
	return ineIndexCode
}

func GetIndexCodeGeneratorPinYing(shortName, indexName, contractCode, suffix, exchange string) string {
	if shortName == "" {
		indexCode = ""
		return indexCode
	}

	if shortName == "top20" {
		indexCode = "top20" + contractCode + suffix
	} else {
		//取公司的全拼
		a := pinyin.NewArgs()
		rows := pinyin.Pinyin(shortName, a)
		strResult := ""
		for i := 0; i < len(rows); i++ {
			if len(rows[i]) != 0 {
				strResult += rows[i][0]
			}
		}
		indexCode = strResult + contractCode + suffix
	}
	err := models.AddBaseFromTradeMapping(indexName, indexCode, exchange)
	if err != nil {
		fmt.Println("add Code err:", err)
	}
	return indexCode
}

// 大连交易所持单排名
func GetDalianHtmlBody(dayNum int, contractId, carietyCode, varietyName string) (body string, err error) {
	defer func() {
		if err != nil {
			fmt.Println("GetDalianHtmlBody Err:" + err.Error())
			msg := "失败提醒" + "GetDalianHtmlBody ErrMsg:" + err.Error()
			go alarm_msg.SendAlarmMsg(msg, 3)
			//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "GetDalianHtmlBody ErrMsg:"+err.Error(), utils.EmailSendToUsers)
		}
	}()

	n := utils.GetRandInt(5, 20)
	time.Sleep(time.Duration(n) * time.Second)
	//定义爬取时间
	endDate := time.Now().AddDate(0, 0, -dayNum).Format(utils.FormatDateTime)
	//endDate := time.Now().Format(utils.FormatDateTime)
	timeDate := utils.StrTimeToTime(endDate)
	currDate := timeDate.Format(utils.FormatDateUnSpace)
	year := timeDate.Year()
	month := timeDate.Format("01")
	var dayStr string
	day := timeDate.Day()
	if day < 10 {
		dayStr = "0" + strconv.Itoa(day)
	} else {
		dayStr = strconv.Itoa(day)
	}
	monthNum, _ := strconv.Atoi(month)
	month = strconv.Itoa(monthNum - 1) //获取时月份需要减一

	list, err := models.GetBaseFromTradeDalianDataList(timeDate.Format(utils.FormatDate))
	listDataMap := make(map[string]int)
	for _, v := range list {
		listDataMap[v.DealShortName+v.ClassifyType+v.DataTime] = v.BaseFromTradeDalianIndexId
	}
	if err != nil {
		fmt.Println(err)
		return
	}
	//模拟form表单请求
	url := "http://www.dce.com.cn/publicweb/quotesdata/memberDealPosiQuotes.html"
	method := "POST"
	payload := &bytes.Buffer{}
	writer := multipart.NewWriter(payload)
	_ = writer.WriteField("memberDealPosiQuotes.variety", carietyCode)
	_ = writer.WriteField("memberDealPosiQuotes.trade_type", "0")
	_ = writer.WriteField("year", strconv.Itoa(year))
	_ = writer.WriteField("month", month)
	_ = writer.WriteField("day", dayStr)
	_ = writer.WriteField("contract.contract_id", contractId)
	_ = writer.WriteField("contract.variety_id", carietyCode)
	_ = writer.WriteField("currDate", currDate)
	err = writer.Close()
	if err != nil {
		utils.FileLog.Info("获取指标失败：" + currDate + varietyName + contractId)
		return
	}
	client := &http.Client{}
	req, err := http.NewRequest(method, url, payload)
	if err != nil {
		return
	}
	req.Header.Add("Cookie", "JSESSIONID=36ACF02A59227A3854F9D5D5E2FB5F2E; WMONID=R5ojcAIIcx-")
	req.Header.Set("Content-Type", writer.FormDataContentType())
	res, err := client.Do(req)
	if err != nil {
		return
	}
	defer res.Body.Close()
	htmlBody, err := ioutil.ReadAll(res.Body)
	if err != nil {
		msg := "失败提醒" + "GetDalianHtmlBody ErrMsg:" + err.Error() + "获取指标失败：" + currDate + varietyName + contractId
		go alarm_msg.SendAlarmMsg(msg, 3)
		//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "GetDalianHtmlBody ErrMsg:"+err.Error()+"获取指标失败："+currDate+varietyName+contractId, utils.EmailSendToUsers)
		return
	}
	body = string(htmlBody)
	return
}

// 大连交易所持单排名
func SyncRankingFromDalian(dayNum int, searchList []*SearchList) (err error) {
	fmt.Println("start")
	defer func() {
		if err != nil {
			fmt.Println("RefreshDataFromDalian SyncRankingFromDalian Err:" + err.Error())
			msg := "失败提醒" + "RefreshDataFromDalian SyncRankingFromDalian ErrMsg:" + err.Error()
			go alarm_msg.SendAlarmMsg(msg, 3)
			//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "RefreshDataFromDalian ErrMsg:"+err.Error(), utils.EmailSendToUsers)
		}
	}()
	//定义爬取时间
	endDate := time.Now().AddDate(0, 0, -dayNum).Format(utils.FormatDateTime)
	//endDate := time.Now().Format(utils.FormatDateTime)
	timeDate := utils.StrTimeToTime(endDate)
	currDate := timeDate.Format(utils.FormatDateUnSpace)
	year := timeDate.Year()
	month := timeDate.Format("01")
	var dayStr string
	day := timeDate.Day()
	if day < 10 {
		dayStr = "0" + strconv.Itoa(day)
	} else {
		dayStr = strconv.Itoa(day)
	}
	monthNum, _ := strconv.Atoi(month)
	month = strconv.Itoa(monthNum - 1) //获取时月份需要减一

	list, err := models.GetBaseFromTradeDalianDataList(timeDate.Format(utils.FormatDate))
	listDataMap := make(map[string]int)
	listDataMapVal := make(map[string]int)
	for _, v := range list {
		listDataMap[v.DealShortName+v.ClassifyType+v.DataTime] = v.BaseFromTradeDalianIndexId
	}
	for _, v := range list {
		listDataMapVal[v.DealValue+v.BuyValue+v.SoldValue] = v.BaseFromTradeDalianIndexId
	}
	if err != nil {
		fmt.Println(err)
		return err
	}
	listIndexCode, err := models.GetIndexCodeMapList("DL") //获取往期指标
	if err != nil {
		fmt.Println(err)
		return err
	}
	listIndexCodeMap := make(map[string]string)
	for _, v := range listIndexCode {
		listIndexCodeMap[v.IndexName] = v.IndexCode
	}
	for _, v := range searchList {
		for _, v2 := range v.ListSearch {
			n := utils.GetRandInt(5, 20)
			time.Sleep(time.Duration(n) * time.Second)
			//模拟form表单请求
			url := "http://www.dce.com.cn/publicweb/quotesdata/memberDealPosiQuotes.html"
			method := "POST"
			payload := &bytes.Buffer{}
			writer := multipart.NewWriter(payload)
			_ = writer.WriteField("memberDealPosiQuotes.variety", v.CarietyCode)
			_ = writer.WriteField("memberDealPosiQuotes.trade_type", "0")
			_ = writer.WriteField("year", strconv.Itoa(year))
			_ = writer.WriteField("month", month)
			_ = writer.WriteField("day", dayStr)
			_ = writer.WriteField("contract.contract_id", v2.ContractId)
			_ = writer.WriteField("contract.variety_id", v.CarietyCode)
			_ = writer.WriteField("currDate", currDate)
			err := writer.Close()
			fmt.Println(currDate, v.VarietyName, v2.ContractId)
			if err != nil {
				utils.FileLog.Info("获取指标失败：" + currDate + v.VarietyName + v2.ContractId)
				return err
			}
			client := &http.Client{}
			req, err := http.NewRequest(method, url, payload)
			if err != nil {
				return err
			}

			req.Header.Add("Cookie", "JSESSIONID=36ACF02A59227A3854F9D5D5E2FB5F2E; WMONID=R5ojcAIIcx-")
			req.Header.Set("Content-Type", writer.FormDataContentType())
			res, err := client.Do(req)
			if err != nil {
				if strings.Contains(err.Error(),"timed out") {
					continue
				}
				fmt.Println("post SyncRankingFromDalian err:"+err.Error())
				utils.FileLog.Info("post SyncRankingFromDalian err:"+err.Error())
				return err
			}
			defer res.Body.Close()
			body, err := ioutil.ReadAll(res.Body)
			if err != nil {
				msg := "失败提醒" + "RefreshDataFromDalian ErrMsg:" + err.Error() + "获取指标失败：" + currDate + v.VarietyName + v2.ContractId
				go alarm_msg.SendAlarmMsg(msg, 3)
				//go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "RefreshDataFromDalian ErrMsg:"+err.Error()+"获取指标失败："+currDate+v.VarietyName+v2.ContractId, utils.EmailSendToUsers)
				return err
			}
			err = DoHtml(string(body), v.VarietyName, v2.ContractId, timeDate, listDataMap, listIndexCodeMap, listDataMapVal)
			if err != nil {
				return err
			}
		}
	}
	return err
}