Generally speaking, if you want to add more functions to the wechat public number, you need to have a server to build the background service of the public number. So in Serverless architecture, is there a more convenient way to achieve such a public number background? Shall we try?

Preliminary build

Serverless native development

First of all, of course, we must have a wechat public number!

Next, we need to apply a fixed IP for our function computing service:

After clicking the whitelist, we can fill in the form to complete the application of fixed public network egress IP.

Next comes code development.

  1. The function is bound to the public, the background, reference documentation: developers.weixin.qq.com/doc/offiacc… We can start with a basic authentication function in the document:
def checkSignature(param):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Access_Overview.html: param param: : return: ' ' '
    signature = param['signature']
    timestamp = param['timestamp']
    nonce = param["nonce"]
    tmparr = [wxtoken, timestamp, nonce]
    tmparr.sort()
    tmpstr = ' '.join(tmparr)
    tmpstr = hashlib.sha1(tmpstr.encode("utf-8")).hexdigest()
    return tmpstr == signature
Copy the code

Define a basic reply method:

def response(body, status=200):
    return {
        "isBase64Encoded": False."statusCode": status,
        "headers": {"Content-Type": "text/html"},
        "body": body
    }
Copy the code

Then at the function entry:

def main_handler(event, context):    
    if 'echostr' in event['queryString'] :# Verification during access
        return response(event['queryString'] ['echostr'] if checkSignature(event['queryString']) else False)
Copy the code

Configure our Yaml:

# serverless.yml
Weixin_GoServerless:
  component: "@serverless/tencent-scf"
  inputs:
    name: Weixin_GoServerless
    codeUri: ./Admin
    handler: index.main_handler
    runtime: Python3.6
    region: ap-shanghai
    description: Wechat public account background server configuration
    memorySize: 128
    timeout: 20
    environment:
      variables:
        wxtoken: Customize a string
        appid: Temporarily don't write
        secret: Temporarily don't write
    events:
      - apigw:
          name: Weixin_GoServerless
          parameters:
            protocols:
              - https
            environment: release
            endpoints:
              - path: /
                method: ANY
                function:
                  isIntegratedResponse: TRUE
Copy the code

Execute code to complete deployment:

Next, in the background of the public account, select basic configuration:

Select modify configuration:

Note here:

  • URL, write the address that was just returned to us after deployment, and add a/at the end

  • Token, write our Yaml wxtoken, keep the same string in both places

  • EncodingAESKey, you can hit random generation

  • The message encryption method can choose plaintext

Once done, we can click Submit:

When we see a successful commit, we have completed the binding of the first step. Next, we go to the background of the function:

Open the fixed egress IP address and copy the IP address:

Click View -> Modify, and copy and paste the IP address in, save. We also look at the developer ID and password:

Copy and paste these two contents into our environment variable:

At this point, we have completed the binding of a public account background service.

To facilitate subsequent operations, get the global variable first:

wxtoken = os.environ.get('wxtoken')
appid = os.environ.get('appid')
secret = os.environ.get('secret')
Copy the code
  1. Next, edit each module (this paper only provides some simple and basic modules, and for more functions, you can refer to the wechat official account documentation for implementation)
  • Get the AccessToken module:
def getAccessToken(a):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Get_access_token.html returned to normal: {"access_token":" access_token"," expiRES_IN ":7200} Return: {"errcode":40013,"errmsg":"invalid APPID "} :return: ""
    url = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s" % (appid, secret)
    accessToken = json.loads(urllib.request.urlopen(url).read().decode("utf-8"))
    print(accessToken)
    return None if "errcode" in accessToken else accessToken["access_token"]
Copy the code
  • Create a custom menu module:
def setMenu(menu):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Custom_Menus/Creating_Custom-Defined_Menu.html right back: {"errcode":0,"errmsg":" OK "} {"errcode":40018,"errmsg":"invalid button name size"} :return: ""
    accessToken = getAccessToken()
    if not accessToken:
        return "Get Access Token Error"

    url = "https://api.weixin.qq.com/cgi-bin/menu/create?access_token=%s" % accessToken
    postData = urllib.parse.urlencode(menu).encode("utf-8")
    requestAttr = urllib.request.Request(url=url, data=postData)
    responseAttr = urllib.request.urlopen(requestAttr)
    responseData = json.loads(responseAttr.read())
    return responseData['errmsg'] if "errcode" in responseData else "success"
Copy the code
  • Common message reply module:
def textXML(body, event):
    ": param event: :return: "": param event: :return:" ": param event: :return: "": param event: :return:"
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
       
        
       
      """.format(toUser=event["FromUserName"],
                                                                   fromUser=event["ToUserName"],
                                                                   time=int(time.time()),
                                                                   msg=body["msg"])


def pictureXML(body, event):
    ": param body: {"media_id": 123} media_id: this parameter is mandatory. Media_id is the ID obtained when a multimedia file is uploaded through the interface in the material management. :param event: :return: '''
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
        
       
        
        
      """.format(toUser=event["FromUserName"],
                                       fromUser=event["ToUserName"],
                                       time=int(time.time()),
                                       media_id=body["media_id"])


def voiceXML(body, event):
    ": param body: {"media_id": 123} media_id: mandatory, using the interface in the material management to upload multimedia files, the obtained id: param event: :return:"
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
        
        
         
         
       
      """.format(toUser=event["FromUserName"],
                                       fromUser=event["ToUserName"],
                                       time=int(time.time()),
                                       media_id=body["media_id"])


def videoXML(body, event):
    ''' :param body: {"media_id": 123, "title": "test", "description": "test} media_id: Description: Optional, description of video message: Param event: :return: ""
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
       
      """.format(toUser=event["FromUserName"],
                                       fromUser=event["ToUserName"],
                                       time=int(time.time()),
                                       media_id=body["media_id"],
                                       title=body.get('title'.' '),
                                       description=body.get('description'.' '))


def musicXML(body, event):
    ''' :param body: {"media_id": 123, "title": "test", "description": "Test} media_id: mandatory, the media ID of the thumbnail, obtained by uploading the multimedia file through the interface in the Material management. Title: Optional, music title description: Optional, music description URL: Optional, music link Hq_URL: Mandatory Param Event: : Return: "param Event: : Return:"
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
        
        <! [CDATA[{title}]]> 
        
         
         
        
         
         
        
         
         
        
         
         
       
      """.format(toUser=event["FromUserName"],
                                       fromUser=event["ToUserName"],
                                       time=int(time.time()),
                                       media_id=body["media_id"],
                                       title=body.get('title'.' '),
                                       url=body.get('url'.' '),
                                       hq_url=body.get('hq_url'.' '),
                                       description=body.get('description'.' '))


def articlesXML(body, event):
    "' : param body: a list [{" title" : "test", "description" : "test", "picUrl" : "test", "url" : "Test "}] picUrl: mandatory, image link, support JPG, PNG format, better effect is 360*200, 200*200 url: Param Event: :return: ""
    if len(body["articles") >8:  A maximum of 8 returns is allowed
        body["articles"] = body["articles"] [0:8]
    tempArticle = """
       
       <! [CDATA[{title}]]> 
       
        
        
       
        
        
       
        
        
      """
    return """
      
       
        
        
       
        
        
       
        {time}
        
       
        
        
       
        {count}
        
       
         {articles} 
       
      """.format(toUser=event["FromUserName"],
                                          fromUser=event["ToUserName"],
                                          time=int(time.time()),
                                          count=len(body["articles"]),
                                          articles="".join([tempArticle.format(
                                              title=eveArticle['title'],
                                              description=eveArticle['description'],
                                              picurl=eveArticle['picurl'],
                                              url=eveArticle['url'])for eveArticle in body["articles"]]))
Copy the code
  • Modify main_handler to make it:

    • Identify binding capabilities

    • Identify basic information

    • Recognize special additional requests (such as triggering custom menu updates via URLS)

Overall code:

def main_handler(event, context):
    print('event: ', event)

    if event["path"] = ='/setMenu':  Set menu interface
        menu = {
            "button": [{"type": "view"."name": "Excellent article"."url": "https://mp.weixin.qq.com/mp/homepage?__biz=Mzg2NzE4MDExNw==&hid=2&sn=168bd0620ee79cd35d0a80cddb9f2487"
                },
                {
                    "type": "view"."name": "Open Source project"."url": "https://mp.weixin.qq.com/mp/homepage?__biz=Mzg2NzE4MDExNw==&hid=1&sn=69444401c5ed9746aeb1384fa6a9a201"
                },
                {
                    "type": "miniprogram"."name": "Online programming"."appid": "wx453cb539f9f963b2"."pagepath": "/page/index"}}]return response(setMenu(menu))

    if 'echostr' in event['queryString'] :# Verification during access
        return response(event['queryString'] ['echostr'] if checkSignature(event['queryString']) else False)
    else:  # User messages/events
        event = getEvent(event)
        if event["MsgType"] = ="text":
            # text message
            return response(body=textXML({"msg": "This is a text message."}, event))
        elif event["MsgType"] = ="image":
            # picture message
            return response(body=textXML({"msg": "This is a picture message."}, event))
        elif event["MsgType"] = ="voice":
            # Voice message
            pass
        elif event["MsgType"] = ="video":
            # Video message
            pass
        elif event["MsgType"] = ="shortvideo":
            # Small video messages
            pass
        elif event["MsgType"] = ="location":
            # Geolocation messages
            pass
        elif event["MsgType"] = ="link":
            # link message
            pass
        elif event["MsgType"] = ="event":
            # Event message
            if event["Event"] = ="subscribe":
                # subscribe to events
                if event.get('EventKey'.None) :# If the user does not follow, the event push after following (TWO-DIMENSIONAL code with parameters)
                    pass
                else:
                    # General attention
                    pass
            elif event["Event"] = ="unsubscribe":
                # unsubscribe events
                pass
            elif event["Event"] = ="SCAN":
                # Event push when users have paid attention (QR code with parameters)
                pass
            elif event["Event"] = ="LOCATION":
                # Reporting geolocation events
                pass
            elif event["Event"] = ="CLICK":
                # Click the menu to pull the event push message
                pass
            elif event["Event"] = ="VIEW":
                Event push when clicking the menu jump link
                pass
Copy the code

You can see it in the code above:

if event["MsgType"] = ="text":
    # text message
    return response(body=textXML({"msg": "This is a text message."}, event))
elif event["MsgType"] = ="image":
    # picture message
    return response(body=textXML({"msg": "This is a picture message."}, event))
Copy the code

This means that when the user sends a text message, we reply the user with a text message: this is a text message. When the user sends an image, we return an image message to the user, testing the connectivity of our background with these two functions:

As you can see, the system can return normally.

Someone asked, what’s the point of such a simple Demo? We can tell you that we can be very light through a function to achieve the back-end service of wechat public number; Here are the basic abilities, and we can add innovation to these basic abilities. For example:

  1. The user sends a picture message. We can tell the user what the picture contains through some image recognition API.

  2. The message sent by the user is a text message, so we can first set some help information/retrieval information for comparison, and then open the chat function for the user if it is not found (this involves natural language processing in artificial intelligence, such as text similarity detection).

  3. If the user is sending a voice message, we can also convert it into text, generate a conversation message, and then convert it into voice and return it to the user

  4. If the user sends the location information, we can return the street view information of the user’s latitude and longitude, or the information of nearby life services

  5. Leave it to your imagination!

Use the Werobot framework

The above method, is through the Serverless native development method for docking. In addition, we can also choose some existing frameworks such as Werobot.

WeRoBot is a wechat public account development framework. Quickly deploy the framework with the 0700-Werobot Component in Serverless Component:

Weixin_Werobot:
  component: "@serverless/tencent-werobot"
  inputs:
    functionName: Weixin_Werobot
    code: ./test
    werobotProjectName: app
    werobotAttrName: robot
    functionConf:
      timeout: 10
      memorySize: 256
      environment:
        variables:
          wxtoken: Your token
      apigatewayConf:
        protocols:
          - http
        environment: release
Copy the code

Then create a new code:

import os
import werobot

robot = werobot.WeRoBot(token=os.environ.get('wxtoken'))

robot.config['SESSION_STORAGE'] = False
robot.config["APP_ID"] = os.environ.get('appid')
robot.config["APP_SECRET"] = os.environ.get('secret')

# @robot.handler Handles all messages
@robot.handler
def hello(message):
    return 'Hello World! '

if __name__ == "__main__":
    Let the server listen at 0.0.0.0:80
    robot.config['HOST'] = '0.0.0.0'
    robot.config['PORT'] = 80
    robot.run()


Copy the code

Install the werobot dependencies locally, then deploy:

Copy the following address to the public background, open the call.

Git: github.com/serverless-… Note that we must close the Session or change the Session to the cloud database, do not use local files, etc., for example, close the Session configuration:

robot.config['SESSION_STORAGE'] = False
Copy the code

Text similarity realizes text and text retrieval

Sometimes users do not know what we sent an article, also do not know the specific content of each article, he may only need simple keywords, to see whether the public number has what he wants.

For example, he searches: How do I upload files? Or search: How to develop Component? The text search function can quickly push the most relevant historical articles to the user, which will be a very convenient thing. The renderings are as follows:

Through simple question description, find the target result, this is the article search function we do. Of course, we can also expand it into a “customer service system,” which is a story later.

To get back to the point, we add two new functions to the previous code:

  • Function 1: index creation function

Main function: By triggering this function, the existing public account data can be sorted out, and the appropriate index file can be established, and stored in COS.


# -*- coding: utf8 -*-
import os
import re
import json
import random
from snownlp import SnowNLP
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client

bucket = os.environ.get('bucket')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
region = os.environ.get('region')
client = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))


def main_handler(event, context):
    response = client.get_object(
        Bucket=bucket,
        Key=event["key"],
    )
    response['Body'].get_stream_to_file('/tmp/output.txt')

    with open('/tmp/output.txt') as f:
        data = json.loads(f.read())

    articlesIndex = []
    articles = {}
    tempContentList = [
        "_"."&nbsp;",]for eveItem in data:
        for i in range(0, len(eveItem['content'] ['news_item'])):
            content = eveItem['content'] ['news_item'][i]['content']
            content = re.sub(r'
      
       '
      (.*?)>.'_', content)
            content = re.sub(r'<.*? > '.' ', content)
            for eve in tempContentList:
                content = content.replace(eve, "")
            desc = "% s. % s. %s" % (
                eveItem['content'] ['news_item'][i]['title'],
                eveItem['content'] ['news_item'][i]['digest']."。".join(SnowNLP(content).summary(3))
            )
            tempKey = "".join(random.sample('zyxwvutsrqponmlkjihgfedcba'.5))
            articlesIndex.append(
                {
                    "media_id": tempKey,
                    "description": desc
                }
            )
            articles[tempKey] = eveItem['content'] ['news_item'][i]

    client.put_object(
        Bucket=bucket,
        Body=json.dumps(articlesIndex).encode("utf-8"),
        Key=event['index_key'],
        EnableMD5=False
    )
    client.put_object(
        Bucket=bucket,
        Body=json.dumps(articles).encode("utf-8"),
        Key=event['key'],
        EnableMD5=False
    )
Copy the code

This part is probably a little bit more customized. The first variable is the tempContentList variable, which can be used to write things that may occur in the public account but are not important. For example, the guide at the end of the public account focuses on the text, which is generally not involved in the search, so it is best to replace and remove the text during index building. Then we also removed the contents of the code tag by using the above code, because code can also affect the results. I also removed the HTML tag.

The original document looked something like this:

Processed file (abstracted by title + Description +SnowNLP) :

These files will be stored in COS.

The core of this part is to correctly let the extracted description describe the content of the article as accurately as possible. In general, the title is the heart of the article, but the title may have lost some information.

For example, the article “Use Tencent Cloud Serverless to know the difference between the two” actually describes the difference between Plugin and Component. The title knows two things, but the core goal is missing, so add the following description: What is the Serverless Framework Plugin? What is Component? What’s the difference between Plugin and Component? To get started with Serverless CLI, the two products must be distinct, and this article will share the differences and corresponding features and functions.

Of course, after adding the description, the content becomes quite accurate, but in the text, there may be more accurate description or additional content, so the title + description + abstract is used (the first three sentences extracted by textRank are extracted text).

  • Function 2: The search function

Main function: When the user sends the specified keyword to the wechat signal, the result can be obtained through this function.

Think: function 1 and function 2 can be integrated into the previous function, why to separate these two functions to make a separate function exist? Wouldn’t it be good to put them in the same function?

Reason is – trigger frequency is one of the most main function, and the function itself does not need too much resource allocation (64 m will be enough), and the functions 1 and 2, may need to consume more resources, if three functions merge together, may need to overall adjustment function of the memory size is big, meet the demand of three functions. That might consume more resources,

For example, the main function fires 10 times (64M, 1S each), function 1 fires twice (512m, 5S each), and function 2 fires four times (384M, 3S each).

If the three functions are put together, the resource consumption is:

If it is executed as three functions, the resource consumption is:

Total resource consumption for the former is 13,308 and 10,432 for the latter. The more calls you make, the greater the percentage of calls to the main function, so the more resources you save. Therefore, it is recommended that modules with large resource consumption differences be deployed in different functions.

import os
import json
import jieba
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
from collections import defaultdict
from gensim import corpora, models, similarities

bucket = os.environ.get('bucket')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
region = os.environ.get('region')
client = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))


def main_handler(event, context):
    response = client.get_object(
        Bucket=bucket,
        Key=event["key"],
    )
    response['Body'].get_stream_to_file('/tmp/output.txt')

    with open('/tmp/output.txt') as f:
        data = json.loads(f.read())

    articles = []
    articlesDict = {}
    for eve in data:
        articles.append(eve['description'])
        articlesDict[eve['description']] = eve['media_id']

    sentence = event["sentence"]

    documents = []
    for eve_sentence in articles:
        tempData = "".join(jieba.cut(eve_sentence))
        documents.append(tempData)
    texts = [[word for word in document.split()] for document in documents]
    frequency = defaultdict(int)
    for text in texts:
        for word in text:
            frequency[word] += 1
    dictionary = corpora.Dictionary(texts)
    new_xs = dictionary.doc2bow(jieba.cut(sentence))
    corpus = [dictionary.doc2bow(text) for text in texts]
    tfidf = models.TfidfModel(corpus)
    featurenum = len(dictionary.token2id.keys())
    sim = similarities.SparseMatrixSimilarity(
        tfidf[corpus],
        num_features=featurenum
    )[tfidf[new_xs]]
    answer_list = [(sim[i], articles[i]) for i in range(1, len(articles))]
    answer_list.sort(key=lambda x: x[0], reverse=True)
    result = []
    print(answer_list)
    for eve in answer_list:
        if eve[0] > 0.10:
            result.append(articlesDict[eve[1]])
    if len(result) >= 8:
        result = result[0:8]
    return {"result": json.dumps(result)}

Copy the code

This part of the code is also very simple, mainly through the similarity of the text to score each text, and then rank according to the score from high to low, given a threshold value (threshold set here is 0.1), output the data before the threshold value.

In addition, the two dependencies quoted here are Jieba and Gensim, both of which may involve binary files, so it is strongly recommended to pack in CentOS.

Next is the call to the main function. To implement the above function, we need to add a new method to the main function:

  1. Get the full text message
def getTheTotalOfAllMaterials(a):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Asset_Management/Get_the_total_of_all_materials.html: return: ' ' '
    accessToken = getAccessToken()
    if not accessToken:
        return "Get Access Token Error"
    url = "https://api.weixin.qq.com/cgi-bin/material/get_materialcount?access_token=%s" % accessToken
    responseAttr = urllib.request.urlopen(url=url)
    return json.loads(responseAttr.read())


def getMaterialsList(listType, count):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Asset_Management/Get_materials_list.html: return: ' ' '
    accessToken = getAccessToken()
    if not accessToken:
        return "Get Access Token Error"

    url = "https://api.weixin.qq.com/cgi-bin/material/batchget_material?access_token=%s" % accessToken
    materialsList = []
    for i in range(1, int(count / 20) + 2):
        requestAttr = urllib.request.Request(url=url, data=json.dumps({
            "type": listType,
            "offset": 20 * (i - 1),
            "count": 20
        }).encode("utf-8"), headers={
            "Content-Type": "application/json"
        })
        responseAttr = urllib.request.urlopen(requestAttr)
        responseData = json.loads(responseAttr.read().decode("utf-8"))
        materialsList = materialsList + responseData["item"]
    return materialsList
Copy the code

This can be called with the following code:

rticlesList = getMaterialsList("news", getTheTotalOfAllMaterials()['news_count'])
Copy the code
  1. Store graphic messages to COS, and implement interfunction calls via the Invoke interface of the function:
def saveNewsToCos(a):
    global articlesList
    articlesList = getMaterialsList("news", getTheTotalOfAllMaterials()['news_count'])
    try:
        cosClient.put_object(
            Bucket=bucket,
            Body=json.dumps(articlesList).encode("utf-8"),
            Key=key,
            EnableMD5=False
        )
        req = models.InvokeRequest()
        params = '{"FunctionName":"Weixin_GoServerless_GetIndexFile", "ClientContext":"{\\"key\\": \\"%s\\", \\"index_key\\": \\"%s\\"}"}' % (
            key, indexKey)
        req.from_json_string(params)
        resp = scfClient.Invoke(req)
        resp.to_json_string()
        response = cosClient.get_object(
            Bucket=bucket,
            Key=key,
        )
        response['Body'].get_stream_to_file('/tmp/content.json')
        with open('/tmp/content.json') as f:
            articlesList = json.loads(f.read())
        return True
    except Exception as e:
        print(e)
        return False
Copy the code
  1. According to the search feedback back Key to achieve the corresponding article content
def searchNews(sentence):
    req = models.InvokeRequest()
    params = '{"FunctionName":"Weixin_GoServerless_SearchNews", "ClientContext":"{\\"sentence\\": \\"%s\\", \\"key\\": \\"%s\\"}"}' % (
        sentence, indexKey)
    req.from_json_string(params)
    resp = scfClient.Invoke(req)
    print(json.loads(json.loads(resp.to_json_string())['Result'] ["RetMsg"]))
    media_id = json.loads(json.loads(json.loads(resp.to_json_string())['Result'] ["RetMsg"[])"result"])
    return media_id if media_id else None
Copy the code

Finally, add the use logic to main_handler:

Logic is very simple answer, is according to the user sent a message, to find the corresponding result. After getting the results, judge the number of results. If there is 1 similar content, return a text, if there are more than one, return the text with links.

The other logic is to build indexes, which are triggered directly by the API gateway. Of course, if it is not safe or necessary, you can add the permission firm parameter:

Additional optimizations:

In the interface list, we can see that the interfaces to access accessToken are actually limited in number of times, with each access valid for two hours. So, we’re going to persist this in our function. For this little thing, it wasn’t cost-effective to get a MySQL, so I decided to use COS:

def getAccessToken(a):
    "' document address: https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Get_access_token.html returned to normal: {"access_token":" access_token"," expiRES_IN ":7200} Return: {"errcode":40013,"errmsg":"invalid APPID "} :return: ""
    global accessToken

    The first judgment is to determine whether there is already accessToken locally, taking into account container reuse
    if accessToken:
        if int(time.time()) - int(accessToken["time"< =])7000:
            return accessToken["access_token"]

    If you don't have accessToken locally, go to cos
    try:
        response = cosClient.get_object(
            Bucket=bucket,
            Key=accessTokenKey,
        )
        response['Body'].get_stream_to_file('/tmp/token.json')
        with open('/tmp/token.json') as f:
            accessToken = json.loads(f.read())
    except:
        pass

    # this time to see if there is in the cosine, if there is in the cosine, again to judge the paragraph
    if accessToken:
        if int(time.time()) - int(accessToken["time"< =])7000:
            return accessToken["access_token"]

    If the process has not stopped, then accessToken has not been obtained, it needs to be obtained from the interface and synchronized to cos
    url = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s" % (appid, secret)
    accessTokenResult = json.loads(urllib.request.urlopen(url).read().decode("utf-8"))
    accessToken = {"time": int(time.time()), "access_token": accessTokenResult["access_token"]}
    print(accessToken)
    response = cosClient.put_object(
        Bucket=bucket,
        Body=json.dumps(accessToken).encode("utf-8"),
        Key=accessTokenKey,
        EnableMD5=False
    )
    return None if "errcode" in accessToken else accessToken["access_token"]
Copy the code

Of course this code can continue to be optimized, but this is just an idea.

In addition to the use of text similarity for text retrieval, we can also use the AI capabilities provided by cloud manufacturers to increase the robot function for the public number, limited to the length of the first paragraph, interested readers can explore.

conclusion

So far, we have completed a simple public number development. Through the Serverless native development ideas (you can also use Werobot and other public account development framework), the public account background service is deployed to the Serverless architecture. A text retrieval function is realized by using natural language processing technology (especially text similarity).

Serverless architecture has great advantages in the development of event-driven triggered scenarios such as wechat public accounts. This paper is only a small exploration. More functions and applications, capabilities and values still depend on specific business. Hopefully, readers will gain a deeper understanding of the Serverless architecture through this article.

Serverless Framework 30-day trial plan

We invite you to experience the most convenient way to develop and deploy Serverless. During the trial period, all related products and services provide free resources and professional technical support to help your business quickly and easily achieve Serverless!

For details, see: Serverless Framework Trial Plan

One More Thing

What can you do in 3 seconds? Take a sip of water, read an email, or — deploy a full Serverless application?

Copy the link to PC browser to access: serverless.cloud.tencent.com/deploy/expr…

Fast deployment in 3 seconds, experience the fastest Serverless HTTP development in history immediately!

Portal:

  • GitHub: github.com/serverless
  • Website: serverless.com

Welcome to Serverless Chinese, you can experience more about Serverless application development in the best practices!


Recommended reading: Serverless Architecture: From Principle, Design to Project Practice