分支合并

src/jfw/modules/weixin/src/github.com/wizjin/weixin/weixin.go

@@ -15,6 +15,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"qfw/util/redis"
 	"regexp"
 	"sort"
 	"time"
@@ -275,6 +276,9 @@ type Weixin struct {
 	appId        string
 	appSecret    string
 	refreshToken bool
+	//
+	refreshTokenChan  chan string
+	refreshTicketChan chan bool
 }
 
 // Convert qr scene to url
@@ -291,9 +295,27 @@ func New(token string, appid string, secret string) *Weixin {
 	wx.refreshToken = false
 	if len(appid) > 0 && len(secret) > 0 {
 		wx.tokenChan = make(chan accessToken)
-		go createAccessToken(wx.tokenChan, appid, secret, &wx.refreshToken)
+		wx.refreshTokenChan = make(chan string)
+		go createAccessToken(wx.tokenChan, appid, &wx.refreshToken, wx.refreshTokenChan)
 		wx.ticketChan = make(chan jsApiTicket)
-		go createJsApiTicket(wx.tokenChan, wx.ticketChan)
+		wx.refreshTicketChan = make(chan bool)
+		go createJsApiTicket(appid, wx.ticketChan, wx.refreshTicketChan)
+		go func() {
+			for {
+				select {
+				case t := <-wx.refreshTokenChan:
+					log.Println("重新获取 access token")
+					c := redis.RedisPool["sso"].Get()
+					c.Do("PUBLISH", "WxRefreshSub", fmt.Sprintf("token_%s %s", wx.appId, t))
+					c.Close()
+				case <-wx.refreshTicketChan:
+					log.Println("重新获取 jsApi ticket")
+					c := redis.RedisPool["sso"].Get()
+					c.Do("PUBLISH", "WxRefreshSub", fmt.Sprintf("ticket_%s", wx.appId))
+					c.Close()
+				}
+			}
+		}()
 	}
 	return wx
 }
@@ -313,8 +335,12 @@ func (wx *Weixin) GetAppSecret() string {
 }
 
 func (wx *Weixin) RefreshAccessToken() {
-	wx.refreshToken = true
-	<-wx.tokenChan
+	for i := 0; i < 10; i++ {
+		go func() {
+			wx.refreshToken = true
+			<-wx.tokenChan
+		}()
+	}
 }
 
 // Register request callback.
@@ -333,6 +359,11 @@ func (wx *Weixin) GetToken() string {
 	return token.token
 }
 
+func (wx *Weixin) GetTicket() string {
+	ticket := <-wx.ticketChan
+	return ticket.ticket
+}
+
 // Post text message
 func (wx *Weixin) PostText(touser string, text string) error {
 	var msg struct {
@@ -815,7 +846,27 @@ func checkSignature(t string, w http.ResponseWriter, r *http.Request) bool {
 	return fmt.Sprintf("%x", h.Sum(nil)) == signature
 }
 
-func authAccessToken(appid string, secret string) (string, time.Duration, int64) {
+func authAccessToken(appid, t string, r chan string) (string, int64, int64) {
+	r <- t
+	ret, err := redis.GetNewBytes("sso", fmt.Sprintf("WxToken_%s", appid))
+	if err != nil {
+		log.Println("redis中获取Token出错", err)
+	} else if ret != nil {
+		var res struct {
+			AccessToken string `json:"access_token"`
+			ExpiresIn   int64  `json:"expires_in"`
+			LastTime    int64  `json:"last_time"`
+		}
+		if err := json.Unmarshal(*ret, &res); err != nil {
+			log.Println("Parse access token failed: ", err)
+		} else {
+			return res.AccessToken, res.ExpiresIn, res.LastTime
+		}
+	}
+	return "", 0, 0
+}
+
+/*func authAccessToken(appid string, secret string) (string, time.Duration, int64) {
 	resp, err := http.Get(weixinHost + "/token?grant_type=client_credential&appid=" + appid + "&secret=" + secret)
 	if err != nil {
 		log.Println("Get access token failed: ", err)
@@ -838,9 +889,30 @@ func authAccessToken(appid string, secret string) (string, time.Duration, int64)
 		}
 	}
 	return "", 0, 0
+}*/
+func getJsApiTicket(appid string, r chan bool) (*jsApiTicket, error) {
+	r <- true
+	ret, err := redis.GetNewBytes("sso", fmt.Sprintf("WxTicket_%s", appid))
+	if err != nil {
+		log.Println("redis中获取Token出错", err)
+	} else if ret != nil {
+		var res struct {
+			Ticket    string `json:"ticket"`
+			ExpiresIn int64  `json:"expires_in"`
+		}
+		if err := json.Unmarshal(*ret, &res); err != nil {
+			return nil, err
+		}
+		var ticket jsApiTicket
+		ticket.ticket = res.Ticket
+		ticket.expires = time.Unix(res.ExpiresIn, 0)
+		return &ticket, nil
+	}
+	time.Sleep(500 * time.Millisecond)
+	return nil, nil
 }
 
-func getJsApiTicket(c chan accessToken) (*jsApiTicket, error) {
+/*func getJsApiTicket(c chan accessToken) (*jsApiTicket, error) {
 	reply, err := sendGetRequest(weixinJsApiTicketURL+"?type=jsapi&access_token=", c)
 	if err != nil {
 		return nil, err
@@ -857,61 +929,58 @@ func getJsApiTicket(c chan accessToken) (*jsApiTicket, error) {
 	ticket.expires = time.Now().Add(time.Duration(res.ExpiresIn * 1000 * 1000 * 1000))
 	return &ticket, nil
 
-}
-
-type countTocken struct {
-	allTimes int64
-	dayTimes int64
-	day      int
-}
-
-var CountT = countTocken{0, 0, time.Now().Day()}
-
-func init() {
-	go func() {
-		for {
-			if time.Now().Day() != CountT.day {
-				CountT.dayTimes = 0
-				CountT.day = time.Now().Day()
-				log.Println("wx-tocken改变统计时间：", CountT.day, ",次数：", CountT.dayTimes)
-			}
-			time.Sleep(30 * time.Second)
-		}
-	}()
-}
+}*/
 
-func createAccessToken(c chan accessToken, appid string, secret string, refresh *bool) {
+func createAccessToken(c chan accessToken, appid string, refresh *bool, r chan string) {
 	lt := int64(0)
 	token := accessToken{"", time.Now(), &lt}
 	c <- token
 	for {
-		if *refresh || (time.Now().Unix()-*token.lasttime) > 90*60 || time.Since(token.expires).Seconds() >= 0 {
-			CountT.allTimes++
-			CountT.dayTimes++
-			*refresh = false
-			var expires time.Duration
-			var lasttime int64
-			token.token, expires, lasttime = authAccessToken(appid, secret)
-			log.Println("获取tocken:", CountT.dayTimes, "token", token.token)
-			token.expires = time.Now().Add(expires)
-			*token.lasttime = lasttime
+		reTry := 0
+		for {
+			if *refresh || (time.Now().Unix()-*token.lasttime) > 5400 || time.Since(token.expires).Seconds() >= 0 {
+				if reTry > 0 {
+					time.Sleep(time.Second)
+				}
+				reTry++
+				subToken := ""
+				if *refresh || *token.lasttime == -1 {
+					subToken = token.token
+				}
+				*refresh = false
+				var expires int64
+				var lasttime int64
+				token.token, expires, lasttime = authAccessToken(appid, subToken, r)
+				if (*refresh || *token.lasttime == -1) && token.expires.Unix() == expires {
+					continue
+				}
+				token.expires = time.Unix(expires, 0)
+				*token.lasttime = lasttime
+			} else {
+				break
+			}
 		}
 		c <- token
-		if CountT.dayTimes > 100 {
-			log.Println(">>-->>获取tocken%d次...<<--<<", CountT.dayTimes)
-			time.Sleep(1 * time.Minute)
-		}
 	}
 }
 
-func createJsApiTicket(cin chan accessToken, c chan jsApiTicket) {
+func createJsApiTicket(appid string, c chan jsApiTicket, r chan bool) {
 	ticket := jsApiTicket{"", time.Now()}
 	c <- ticket
 	for {
-		if time.Since(ticket.expires).Seconds() >= 0 {
-			t, err := getJsApiTicket(cin)
-			if err == nil {
-				ticket = *t
+		reTry := 0
+		for {
+			if time.Since(ticket.expires).Seconds() >= 0 {
+				if reTry > 0 {
+					time.Sleep(time.Second)
+				}
+				reTry++
+				t, err := getJsApiTicket(appid, r)
+				if err == nil && t != nil {
+					ticket = *t
+				}
+			} else {
+				break
 			}
 		}
 		c <- ticket
@@ -939,7 +1008,7 @@ func sendGetRequest(reqURL string, c chan accessToken) ([]byte, error) {
 			case 0:
 				return reply, nil
 			case 42001, 40001:
-				*token.lasttime = 0
+				*token.lasttime = -1
 				<-c
 				log.Println("reget the tocken by ", result.ErrorCode, result.ErrorMessage)
 				continue
@@ -972,7 +1041,7 @@ func postRequest(reqURL string, c chan accessToken, data []byte) ([]byte, error)
 			case 0:
 				return reply, nil
 			case 42001, 40001:
-				*token.lasttime = 0
+				*token.lasttime = -1
 				<-c
 				log.Println("reget the tocken by ", result.ErrorCode, result.ErrorMessage)
 				continue

src/jfw/modules/weixin/src/main.go

@@ -38,6 +38,7 @@ func main() {
 	http.HandleFunc("/wx/lsmstr/", wx.NewStrQrHandle)
 	http.HandleFunc("/wx/lsm/", wx.AdvHandle)
 	http.HandleFunc("/wx/token", wx.GetToken)
+	http.HandleFunc("/wx/ticket", wx.GetTicket)
 	// 设置监听的端口
 	err := endless.ListenAndServe(":"+Sysconfig["weixinport"].(string), nil, func() {})
 	if err != nil {

src/jfw/modules/weixin/src/wx/wx.go

@@ -619,6 +619,7 @@ func Subscribe(w ResponseWriter, r *Request) {
 	} else if strings.Count(r.EventKey, "_") >= 2 {
 		source = strings.SplitN(r.EventKey, "_", 2)[1]
 	}
+	go saveFixedCode(openid, r.Event, source)
 	log.Println("source:", source)
 	var shareData interface{}
 	shareData = redis.Get("sso", "p_shareData_"+source)
@@ -949,6 +950,7 @@ func ScanHandler(w ResponseWriter, r *Request) {
 		return
 	}
 	log.Println("openid======", openid, r.EventKey)
+	go saveFixedCode(openid, r.Event, r.EventKey)
 	if r.EventKey == "69" { //数据报告-扫码回复图文
 		datareportReply := util.ObjToMap(config.Sysconfig["datareportUserReply"])
 		if datareportReply != nil && len(*datareportReply) > 0 {
@@ -1494,6 +1496,11 @@ func GetToken(w http.ResponseWriter, r *http.Request) {
 	fmt.Fprint(w, Mux.GetToken())
 }
 
+//获取Ticket
+func GetTicket(w http.ResponseWriter, r *http.Request) {
+	fmt.Fprint(w, Mux.GetTicket())
+}
+
 //生成推广临时二维码地址
 func AdvHandle(w http.ResponseWriter, r *http.Request) {
 	param := r.RequestURI[8:]
@@ -1768,13 +1775,17 @@ func downloadUserFace(url string) string {
 		util.Try(func() {
 			os.MkdirAll(config.Sysconfig["headimage"].(string)+fmt.Sprintf("/upload/%s/%s/%s", tn.Format("2006"), tn.Format("01"), tn.Format("02")), 0777)
 			fi, err := os.OpenFile(config.Sysconfig["headimage"].(string)+filename, os.O_CREATE|os.O_TRUNC|os.O_SYNC|os.O_RDWR, 0x666)
-			defer fi.Close()
-			resp, err := http.Get(url)
-			defer resp.Body.Close()
 			if err == nil {
-				io.Copy(fi, resp.Body)
+				log.Println("download userface OpenFile err:", err.Error())
 			} else {
-				log.Println("download userface err:", err.Error())
+				defer fi.Close()
+				resp, err := http.Get(url)
+				if err == nil {
+					defer resp.Body.Close()
+					io.Copy(fi, resp.Body)
+				} else {
+					log.Println("download userface err:", err.Error())
+				}
 			}
 		}, func(e interface{}) {})
 	}()
@@ -1804,3 +1815,17 @@ func HashVal(src string) int {
 	}
 	return check
 }
+
+//记录固定码的扫码日志
+func saveFixedCode(openid, t, code string) {
+	defer util.Catch()
+	if !numCodeReg.MatchString(code) {
+		return
+	}
+	tools.MQFW.Save("wxfixedcode", map[string]interface{}{
+		"createtime": time.Now().Unix(),
+		"type":       t,
+		"code":       code,
+		"openid":     openid,
+	})
+}

src/jfw/modules/wxtoken/src/config.json

@@ -0,0 +1,9 @@
+{
+	"wxs":[
+		{
+			"appid":"wx76e1309b01a7b17e",
+			"appsecret":"dd00e71cb2370432d9de848b674eb8e7"
+		}
+	],
+	"redis":"sso=192.168.3.128:1712"
+}

src/jfw/modules/wxtoken/src/main.go

@@ -0,0 +1,306 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"qfw/util"
+	"qfw/util/redis"
+	"strings"
+	"time"
+
+	redigo "github.com/garyburd/redigo/redis"
+)
+
+const (
+	weixinHost           = "https://api.weixin.qq.com/cgi-bin"
+	weixinJsApiTicketURL = "https://api.weixin.qq.com/cgi-bin/ticket/getticket"
+	WxTokenKey           = "WxToken_%s"
+	WxTicketKey          = "WxTicket_%s"
+	RedisCode            = "sso"
+	MaxExpSec            = 5400
+	retryMaxN            = 3
+)
+
+var (
+	Config    *config
+	AllCountT = map[string]*countTocken{}
+	AllWxs    = map[string]*wx{}
+)
+
+type wx struct {
+	Appid      string
+	Appsecret  string
+	TokenChan  chan accessToken
+	TicketChan chan jsApiTicket
+}
+type config struct {
+	Wxs   []*wx
+	Redis string
+}
+type accessToken struct {
+	Token    string
+	Expires  time.Time
+	Lasttime *int64
+}
+
+type wxToken struct {
+	AccessToken string `json:"access_token"`
+	ExpiresIn   int64  `json:"expires_in"`
+	LastTime    int64  `json:"last_time"`
+}
+
+type wxTicket struct {
+	Ticket    string `json:"ticket"`
+	ExpiresIn int64  `json:"expires_in"`
+}
+
+type jsApiTicket struct {
+	Ticket  string
+	Expires time.Time
+}
+type countTocken struct {
+	allTimes int64
+	dayTimes int64
+	day      int
+}
+
+func init() {
+	util.ReadConfig(&Config)
+	redis.InitRedis(Config.Redis)
+	for _, v := range Config.Wxs {
+		v.TokenChan = make(chan accessToken)
+		v.TicketChan = make(chan jsApiTicket)
+		AllWxs[v.Appid] = v
+		AllCountT[v.Appid] = &countTocken{0, 0, time.Now().Day()}
+	}
+	go func() {
+		for {
+			for CountK, CountT := range AllCountT {
+				if time.Now().Day() != CountT.day {
+					CountT.dayTimes = 0
+					CountT.day = time.Now().Day()
+					log.Println(CountK, "wx-tocken改变统计时间：", CountT.day, ",次数：", CountT.dayTimes)
+				}
+			}
+			time.Sleep(30 * time.Second)
+		}
+	}()
+	go func() {
+		for {
+			defer util.Catch()
+		L:
+			for {
+				conn := redis.RedisPool[RedisCode].Get()
+				if conn.Err() == nil {
+					psc := redigo.PubSubConn{Conn: conn}
+					if err := psc.Subscribe(redigo.Args{}.AddFlat("WxRefreshSub")...); err != nil {
+						log.Println(err)
+					}
+					for {
+						msg := psc.Receive()
+						switch n := msg.(type) {
+						case error:
+							log.Println("redis receive error", msg)
+							conn.Close()
+							break L
+						case redigo.Message:
+							res := string(n.Data)
+							if flag := "token_"; strings.HasPrefix(res, flag) {
+								params := strings.Split(strings.TrimLeft(res, flag), " ")
+								appid := params[0]
+								c := AllWxs[appid]
+								if c != nil {
+									token := <-c.TokenChan
+									//意外情况，导致token失效，需要重新获取token
+									if token.Token == params[1] {
+										refreshAccessToken(appid)
+									}
+								} else {
+									log.Println("appid", appid, "不存在")
+								}
+							} else if flag := "ticket_"; strings.HasPrefix(res, flag) {
+								appid := strings.TrimLeft(res, flag)
+								c := AllWxs[appid]
+								if c != nil {
+									<-c.TicketChan
+								} else {
+									log.Println("appid", appid, "不存在")
+								}
+							}
+						}
+					}
+				}
+				time.Sleep(1 * time.Second)
+			}
+			time.Sleep(500 * time.Millisecond)
+		}
+	}()
+}
+func main() {
+	log.Println("程序启动。。。")
+	for _, v := range AllWxs {
+		log.Println("初始化 token ticket", v.Appid)
+		go createAccessToken(v.Appid)
+		go createJsApiTicket(v.Appid)
+	}
+	<-chan bool(nil)
+}
+func createAccessToken(appid string) {
+	lt := int64(0)
+	token := accessToken{"", time.Now(), &lt}
+	AllWxs[appid].TokenChan <- token
+	tokenKey := fmt.Sprintf("WxToken_%s", appid)
+	for {
+		ret, err := redis.GetNewBytes(RedisCode, tokenKey)
+		if err == nil {
+			if ret != nil {
+				var res wxToken
+				err := json.Unmarshal(*ret, &res)
+				if err != nil {
+					log.Println(appid, "从redis加载token出错", err)
+				} else {
+					token.Token = res.AccessToken
+					token.Expires = time.Now().Add(time.Duration(res.ExpiresIn * 1000 * 1000 * 1000))
+					*token.Lasttime = res.LastTime
+				}
+			} else {
+				*token.Lasttime = 0
+				token.Expires = time.Now()
+			}
+		}
+		if (time.Now().Unix()-*token.Lasttime) > MaxExpSec || time.Since(token.Expires).Seconds() >= 0 {
+			AllCountT[appid].allTimes++
+			AllCountT[appid].dayTimes++
+			resp, err := http.Get(weixinHost + "/token?grant_type=client_credential&appid=" + appid + "&secret=" + AllWxs[appid].Appsecret)
+			if err != nil {
+				log.Println(appid, "Get access token failed: ", err)
+			} else {
+				defer resp.Body.Close()
+				body, err := ioutil.ReadAll(resp.Body)
+				if err != nil {
+					log.Println(appid, "Read access token failed: ", err)
+				} else {
+					var res wxToken
+					if err := json.Unmarshal(body, &res); err != nil {
+						log.Println(appid, "Parse access token failed: ", err)
+					} else {
+						token.Token = res.AccessToken
+						token.Expires = time.Now().Add(time.Duration(res.ExpiresIn * 1000 * 1000 * 1000))
+						*token.Lasttime = time.Now().Unix()
+						redis.Put(RedisCode, tokenKey, map[string]interface{}{
+							"access_token": res.AccessToken,
+							"expires_in":   token.Expires.Unix(),
+							"last_time":    token.Lasttime,
+						}, MaxExpSec)
+						log.Println(appid, "获取tocken:", AllCountT[appid].dayTimes, "token", res.AccessToken, "expires_in", res.ExpiresIn, "expires", token.Expires.Unix())
+					}
+				}
+			}
+		}
+		AllWxs[appid].TokenChan <- token
+		if AllCountT[appid].dayTimes > 100 {
+			log.Println(appid, ">>-->>获取tocken...<<--<<", AllCountT[appid].dayTimes, "次")
+			time.Sleep(1 * time.Minute)
+		}
+	}
+}
+
+//刷新token
+func refreshAccessToken(appid string) {
+	redis.Del(RedisCode, fmt.Sprintf(WxTokenKey, appid))
+	<-AllWxs[appid].TokenChan
+}
+
+func createJsApiTicket(appid string) {
+	ticket := jsApiTicket{"", time.Now()}
+	AllWxs[appid].TicketChan <- ticket
+	ticketKey := fmt.Sprintf(WxTicketKey, appid)
+	for {
+		ret, err := redis.GetNewBytes(RedisCode, ticketKey)
+		if err == nil {
+			if ret != nil {
+				var res wxTicket
+				err := json.Unmarshal(*ret, &res)
+				if err != nil {
+					log.Println(appid, "从redis加载ticket出错", err)
+				} else {
+					ticket.Ticket = res.Ticket
+					ticket.Expires = time.Now().Add(time.Duration(res.ExpiresIn * 1000 * 1000 * 1000))
+				}
+			} else {
+				ticket.Expires = time.Now()
+			}
+		}
+		if time.Since(ticket.Expires).Seconds() >= 0 {
+			t, err := getJsApiTicket(appid, ticketKey)
+			if err == nil {
+				ticket = *t
+			}
+		}
+		AllWxs[appid].TicketChan <- ticket
+	}
+}
+
+func getJsApiTicket(appid, ticketKey string) (*jsApiTicket, error) {
+	reply, err := sendGetRequest(appid, weixinJsApiTicketURL+"?type=jsapi&access_token=")
+	if err != nil {
+		log.Println(appid, "getJsApiTicket error", err)
+		return nil, err
+	}
+	var res wxTicket
+	if err := json.Unmarshal(reply, &res); err != nil {
+		log.Println(appid, "getJsApiTicket json.Unmarshal error", err)
+		return nil, err
+	}
+	var ticket jsApiTicket
+	ticket.Ticket = res.Ticket
+	ticket.Expires = time.Now().Add(time.Duration(res.ExpiresIn * 1000 * 1000 * 1000))
+	redis.Put(RedisCode, ticketKey, map[string]interface{}{
+		"ticket":     res.Ticket,
+		"expires_in": ticket.Expires.Unix(),
+	}, int(res.ExpiresIn))
+	log.Println(appid, "获取ticket:", "ticket", res.Ticket, "expires_in", res.ExpiresIn, "expires", ticket.Expires.Unix())
+	return &ticket, nil
+}
+
+func sendGetRequest(appid, reqURL string) ([]byte, error) {
+	for i := 0; i < retryMaxN; i++ {
+		token := <-AllWxs[appid].TokenChan
+		if time.Since(token.Expires).Seconds() < 0 {
+			r, err := http.Get(reqURL + token.Token)
+			if err != nil {
+				log.Println(appid, "get JsApiTicket error", err)
+				continue
+			}
+			defer r.Body.Close()
+			reply, err := ioutil.ReadAll(r.Body)
+			if err != nil {
+				log.Println(appid, "Read access token failed: ", err)
+				return nil, err
+			}
+			var result struct {
+				ErrorCode    int    `json:"errcode,omitempty"`
+				ErrorMessage string `json:"errmsg,omitempty"`
+			}
+			if err := json.Unmarshal(reply, &result); err != nil {
+				log.Println(appid, "Parse JsApiTicket error", err)
+				return nil, err
+			}
+			switch result.ErrorCode {
+			case 0:
+				return reply, nil
+			case 42001, 40001:
+				refreshAccessToken(appid)
+				log.Println(appid, "reget the tocken by ", result.ErrorCode, result.ErrorMessage)
+				continue
+			default:
+				return nil, errors.New(fmt.Sprintf("WeiXin send get request reply[%d]: %s", result.ErrorCode, result.ErrorMessage))
+			}
+		}
+	}
+	return nil, errors.New("WeiXin post request too many times:" + reqURL)
+}