Python teaches you to build wechat push live Betta reminder from 0 (single room simplified version)

Python Programming 2021-09-15 10:34:56

Re launch remaking , Single room push reminder version ( The following text has the idea of expanding multi room ), Suitable for small partners without server to learn to use .

Gitee Document synchronization update :https://gitee.com/LGW_space/dy-live-lintener
The language bird documentation is more detailed ( Including finished product cases ):https://www.yuque.com/books/share/fdc7c120-e4eb-47d5-93de-dcf5d347e5a8?# 《 Betta live reminder service 》


Betta live reminder —— Simplified edition

Function is introduced :

 Of course, I'm learning Python It's going to be difficult , There is no good learning material , How to learn ?
Study Python I don't know how to recommend to join the communication Q Group number :928946953
There are like-minded partners in the group , Help each other , There are good video tutorials and PDF!
And Daniel's answer !

 

  • After the anchor starts broadcasting or changes the room title, push the reminder to wechat ( Single room )

    > In theory, multiple rooms can be configured , It can achieve the same effect as the whole room push of Betta .

Project screenshots :

<ignore_js_op>

 

[TOC]

One 、 Project preparation

1. Technical content used

  • Cloud functions
  • Object storage ( Why use object storage , As described below )
  • python(3.6/3.7)
  • WxPusher

2. Tools needed

  • Tencent cloud ( Cloud functions and cloud object storage )

Two 、 Open object storage

Here we use object storage, which is equivalent to a function of the database , Why use object storage ?
> Object storage adopts a flat file organization , So when the number of documents rises to ten million 、 Billion level , Capacity in PB At the level , The performance advantage of this file organization is shown , The file no longer has the problem of directory tree depth , Historical and near line data have the same access efficiency . in addition , Object storage mostly adopts distributed architecture ,

In short, it is cheap and efficient

Official entry steps

1. Sign up for Tencent cloud

The official registration method

2. Opening COS service

stay Tencent cloud console in , choice 【 Cloud products 】>【 Object storage 】, Get into COS Console , Open according to the interface prompt COS service .( If you have opened , Please skip this step .)

3. Create buckets

We need to create a bucket for storing objects :

  1. stay Object Storage Console Click... In the left navigation bar 【 Bucket list 】, Enter the bucket management page .

  2. single click 【 Create buckets 】, Enter the following configuration information , Other configurations can be left as default .

    • name : Enter bucket name . The name cannot be modified after setting . Here, for example, enter examplebucket.
    • The region : Where the bucket belongs , Select the region closest to your business , For example, Guangzhou .
    • Access right : Bucket access , Here we leave the default as “ Private reading and writing ”.
  3. single click 【 determine 】, You can create it .

3、 ... and 、 Development of cloud functions

Cloud function we use Tencent cloud function as an example , The same goes for Alibaba cloud .( It seems that Alibaba cloud does not allow trigger cycles to be lower than 1 minute )

1. Sign up for Tencent cloud

The official registration method

2. Create cloud functions

  • Log in to Tencent cloud , Find the cloud function in the cloud product , stay 【 Function service 】 Create a new cloud function in
  • choice 【 Custom creation 】
  • modify 【 The name of the function 】、【 regional 】、【 Running environment (Python3.6)】、【 Submission method ( Online editing )】、【 Execution method (index.main_handler)】 Don't change the default code for now .
  • Click finish

3. Perfect function

Before you write the code , We need to use these key information :

  • appid : Yours APPID( There is... In the account information )
  • secret_id : Yours SecretId (API Get from key management )
  • secret_key : Yours SecretKey(API Get from key management )
  • region : The area where the bucket is located
  • BUCKET: Bucket name
  • token : No can be empty
  • FILE_NAME : The file name you want to upload Such as room_info
  • WxPusherToken :WxPusher Users of appToken

3.1 Introduce the required package , Complete the basic configuration

 Copy code Hidden code
# -*- coding: utf8 -*-
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
from qcloud_cos_v5 import CosServiceError
from qcloud_cos_v5 import CosClientError
import datetime
import json
import logging
import requests
from urllib.parse import parse_qs
logger = logging.getLogger()
logger.setLevel(logging.ERROR) # Default printing INFO The level of log , It can be adjusted to DEBUG、WARNING、ERROR、CRITICAL Level log 

appid = 123123 # Please replace with your APPID. Please replace it with your APPID
secret_id = u'xxx' # Please replace with your SecretId. Please replace it with your SecretId
secret_key = u'xxx' # Please replace with your SecretKey. Please replace it with your SecretKey
region = u'ap-shanghai' # Please replace with the region where COS bucket located. Please replace it with you bucket Where it's located 
BUCKET = 'xxx'
FILE_NAME = 'xxx'
token = ''
WxPusherToken = 'xxxxxx'
# Configure buckets 
config = CosConfig(Secret_id=secret_id, Secret_key=secret_key, Region=region, Token=token)
client = CosS3Client(config)

 

notes :qcloud_cos_v5  by COS Of SDK

3.2 Create method Get the live broadcast status according to the room number

Betta third party API, Get room information through room number , Get it from Baidu .
Finally back to Room information

 Copy code Hidden code
def getRoomInfo():
url = ' Betta official API/ Your room number, '
try:
r = requests.get(url, timeout=5)
except requests.exceptions.RequestException as e:
print(str(e))
return 'timeout'
else:
if r.status_code == 404:
return '404'
json_str = json.loads(r.text)
if isinstance(json_str, dict):
if 'error' in json_str.keys() and json_str['error'] == 0:
room_info = json_str['data']
return room_info

 

3.3 Create method send out WxPusher news

 Copy code Hidden code
def send_WxPusher_msg(c):
headers = {
'Content-Type': 'application/json'
}
url = 'http://wxpusher.zjiecode.com/api/send/message'
parameter = {
"appToken": WxPusherToken,
"content": c,
"contentType": 1, # Content type 1 It means words 2 Express html( Send only body The data inside the tag can , barring body label ) 3 Express markdown 
"topicIds":[
Your subject number
]
}
r = requests.post(url=url, headers=headers,
data=json.dumps(parameter, ensure_ascii=False).encode('utf-8'))
json_str = json.loads(r.text)
datas = json_str['data']
for data in datas:
if data['code'] == 1001:
logger.error(data['status'])

 

3.4 Create method to update ( Upload ) file

Set the update time field , Used to lock updates , Updates are not allowed within the specified time / Push repeatedly ( I guess it's the problem of Tencent cloud multithreading , Will result in repeated updates 、 Push repeatedly )

 Copy code Hidden code
def updateLocalInfo(roominfo):
DIC = {
"room_id": roominfo["room_id"],
"room_status": roominfo["room_status"],
"room_name": roominfo["room_name"],
"update_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
response = client.put_object(
Bucket=BUCKET,
Body=DIC,
Key=FILE_NAME
)
print(" The update is successful :",response['ETag'])

 

3.5 Create method get files

url obtain : After uploading the file , Click the... Of the file in the bucket details , essential information Of Object address Information
Such as  https:// Bucket name .cos. regional .myqcloud.com/ file name

 Copy code Hidden code
def getLocalInfo():
url = ' The object address of the file '
r = requests.get(url)
info_str = r.text
params = parse_qs(info_str)
result = {key: params[key][0] for k
ey in params} return result

 


3.6 Create method Information Integration

 Copy code Hidden code
def time_last(time_str): # Calculate the time difference 
t_r = datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
now_time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
t_n = datetime.datetime.strptime(now_time_str, "%Y-%m-%d %H:%M:%S")
return (t_n - t_r).seconds
def checkRoomInfo():
room = getLocalInfo() # Get historical room information 
curr_info = getRoomInfo() # Get the current real room information 
time_l = time_last(room['update_time']) # Calculate the time difference between the current time and the last update time 
# Live broadcast reminder 
if curr_info['room_status'] == "1": # Live broadcast in progress 
if room['room_status'] == "2": # The previous status was Not on the air 
if time_l > 300: # 300 Cannot push repeatedly within seconds 
content = '【 premiere 】 The anchor you follow :' + curr_info['owner_name'] + ' It's live ' + \
'\n Room title :' + curr_info['room_name'] + \
'\n Live broadcast in progress :' + curr_info['cate_name']
send_WxPusher_msg(content)
updateLocalInfo(roominfo=curr_info)
else:
logger.error(
'【 premiere 】 Your room number, :' + room['room_id'] + '\n Last update :' + room['update_time'] + '\n Time difference :' + str(
time_l))
else:
if room['room_status'] == "1": # Turned off the live broadcast 
updateLocalInfo(roominfo=curr_info)
# Change room Title reminder 
if curr_info['room_name'] != room['room_name']:
if time_l > 300:
content1 = '【 Change title 】 The anchor you follow :' + curr_info['owner_name'] + ' Changed the room title ' + \
'\n Current room title :' + curr_info['room_name']
send_WxPusher_msg(content1)
updateLocalInfo(roominfo=curr_info)
else:
logger.error(
'【 Change title 】 Your room number, :' + room['room_id'] + '\n Last update :' + room['update_time'] + '\n Time difference :' + str(
time_l))

 

By this time, all the work will be basically completed ~

Development and extension ideas

Multi room support : If you support one room, you must support multiple rooms .

  • Download the file : The content of the file is Dictionary /json Array , Every dictionary records information about the room .
  • Traverse : After obtaining the room information in the storage bucket , With betta API Traverse to get the real information of each room .
  • to update / Upload : By comparing real information , Update dictionary array , Finally upload to COS.
Please bring the original link to reprint ,thank
Similar articles