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

Python Programming 2021-09-15 10:34:56
python teaches build wechat push


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.
版权声明
本文为[Python Programming]所创,转载请带上原文链接,感谢
https://pythonmana.com/2021/09/20210909152354290U.html

  1. Realizing the function of sending e-mail automatically with Python
  2. Smtpauthenticationerror in Python: solution
  3. 8 steps to teach you how to solve Sudoku in Python! (including source code)
  4. Python change la vie | identifier facilement des centaines de numéros de livraison
  5. Python change life | utilisation de modèles reconnus par ocr
  6. Bibliothèques Python utiles et intéressantes - - psutil
  7. 3. Traitement des données pandas
  8. 【Python编程基础】控制流之链式比较运算符
  9. MFC uses Python scripting language
  10. 【Python編程基礎】控制流之鏈式比較運算符
  11. 【 base de programmation python】 opérateur de comparaison de chaîne pour le flux de contrôle
  12. Python game development, pyGame module, python implementation of Xiaole games
  13. Mise en œuvre du Code de vérification unique (OTP) avec le cadre de repos Django
  14. Python - eval ()
  15. Python - Programmation orientée objet - _Rapport()
  16. Différence entre python - rep (), Str ()
  17. Python - Programmation orientée objet - _Appel()
  18. Python calling matlab script
  19. Python - Programmation orientée objet - _Nouveau() et mode Singleton
  20. Python - Programmation orientée objet - méthode magique (méthode de double soulignement)
  21. Python - Programmation orientée objet - pratique (6)
  22. Python - Programmation orientée objet - réflexion hasattr, GetAttr, GetAttr, delattr
  23. Python - Programmation orientée objet - _Dict
  24. Python - pydantic (2) Modèle imbriqué
  25. Non-ASCII character ‘\xe5‘ in file kf1.py on line 4, but no encoding declared; see http://python.or
  26. python笔记(一)
  27. Non - ASCII character 'xe5' in file kf1.py on Line 4, but no Encoding declared;Voirhttp://python.or
  28. Notes Python (1)
  29. Talk about how JMeter executes Python scripts concurrently
  30. In Beijing, you can't see the moon in the Mid Autumn Festival. Draw a blood red moon in Python
  31. Un des pandas crée un objet
  32. Machine learning | unitary regression model Python practical case
  33. Draw a "Lollipop chart" with Excel and python
  34. Python uses scikit learn to calculate TF-IDF
  35. Getting started with Python Basics_ 3 conditional statements and iterative loops
  36. Python dynamic properties and features
  37. 云计算开发:Python内置函数-min()函数详解
  38. [Python skill] how to speed up loop operation and numpy array operation
  39. 雲計算開發:Python內置函數-min()函數詳解
  40. Développement de l'informatique en nuage: explication détaillée de la fonction intégrée python - min ()
  41. 从0起步学Python(附程序实例讲解)第1讲
  42. 从0起步学Python(附程序实例讲解)第1讲
  43. Apprendre Python à partir de 0 (avec des exemples de programme) leçon 1
  44. Apprendre Python à partir de 0 (avec des exemples de programme) leçon 1
  45. With Python, I'll take you to enjoy it for a month when the Mid Autumn Festival is coming
  46. You can't write interface software in Python! Which software on sale has no UI?
  47. Python国内外原题解析及源码1~15
  48. Python实现长篇英文自动纠错~
  49. Python implémente la correction automatique des erreurs en anglais long
  50. Analyse des problèmes originaux et code source de Python au pays et à l'étranger 1 ~ 15
  51. 一张思维导图学Python之表白
  52. Python教学中课程思政建设的研究探索2
  53. Recherche sur la construction idéologique et politique du Programme d'études dans l'enseignement Python 2
  54. Une présentation de la cartographie mentale Python
  55. Python高级用法总结(8)-函数式编程
  56. Python + Mirai development QQ robot starting tutorial (2021.9.9 test is valid)
  57. Python Advanced use Summary (8) - functional Programming
  58. How to get started with Python and share learning methods for free. All you want to know is here
  59. Python + Mirai development QQ robot starting tutorial (2021.9.9 test is valid)
  60. Python基础第1讲(含代码、Python最新安装包、父与子的编程之旅:与小卡特一起学Python中文版)