(1)首先需要在项目的根路径下,创建utils包,用于存放项目的其他数据接口;
在安仁等地区,都构建了全面的区域性战略布局,加强发展的系统性、市场前瞻性、产品创新能力,以专注、极致的服务理念,为客户提供成都网站制作、网站设计 网站设计制作按需规划网站,公司网站建设,企业网站建设,成都品牌网站建设,成都营销网站建设,成都外贸网站建设公司,安仁网站建设费用合理。(2)然后在utils包下,创建一个
wechat_send.py
模块, 用于给用户封装企业微信的接口,提供给外界调用;
import requests
import json
class Wechat_Info:
def __init__(self):
self.partyID = '1'
self.corpID = 'ww5cfabaf35ce8cd7b'
self.secret = 'uiwvmNj8f1IVy3QYrZ62WePGFKA_BsIPmHigq3TRydM'
self.agentID = '1000002'
self.token = None
def __get_token(self, corpid, secret):
Url = "/tupian/20230522/gettoken"
Data = {
"corpid": corpid,
"corpsecret": secret
}
r = requests.get(url=Url, params=Data)
token = r.json()['access_token']
return token
def send_message(self, message):
url = "/tupian/20230522/send
self.__get_token(self.corpID, self.secret))
data = {
"toparty": self.partyID,
"msgtype": "text",
"agentid": self.agentID,
"text": {
"content": message
},
"safe": "0"
}
result = requests.post(url=url, data=json.dumps(data))
return result.text
if __name__ == '__main__':
wechat_info = Wechat_Info()
result = wechat_info.send_message('微信测试')
print(result)
partyID
、
corpID
、
secret
、
agentID
都是配置信息,需要引入一个读取配置文件的库
configparser
;
安装:
pip install configparser
xkd.conf
文件;
[Wechat]
party_id = 1
corp_id = ww5cfabaf35ce8cd7b
secret = uiwvmNj8f1IVy3QYrZ62WePGFKA_BsIPmHigq3TRydM
agent_id = 1000002
const_file.py
文件,依赖作为项目中全局常量的引用;
import os
PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CONFIG_DIR = os.path.join(PROJECT_DIR, 'cfg')
# 指定配置文件的路径
CONFIG_FILE = os.path.join(CONFIG_DIR, 'xkd.conf')
wechat_send.py
模块接口(配置中的信息都是明文的,在项目开发中,最好将重要的配置信息进行加密,可以使用base64进行加密和解密);
import requests
import json
from configparser import ConfigParser
from utils.const_file import CONFIG_FILE
class Wechat_Info:
def __init__(self):
self.wechat_info = self.__get_config_info()
def __get_config_info(self):
parser = ConfigParser()
parser.read(CONFIG_FILE)
wechat_info = {}
wechat_info['party_id'] = parser.get('Wechat', 'party_id')
wechat_info['corp_id'] = parser.get('Wechat', 'corp_id')
wechat_info['secret'] = parser.get('Wechat', 'secret')
wechat_info['agent_id'] = parser.get('Wechat', 'agent_id')
return wechat_info
def __get_token(self, corpid, secret):
Url = "/tupian/20230522/gettoken"
Data = {
"corpid": corpid,
"corpsecret": secret
}
r = requests.get(url=Url, params=Data)
token = r.json()['access_token']
return token
def send_message(self, message):
url = "/tupian/20230522/send
self.__get_token(self.wechat_info.get('corp_id'), self.wechat_info.get('secret')))
data = {
"toparty": self.wechat_info.get('party_id'),
"msgtype": "text",
"agentid": self.wechat_info.get('agent_id'),
"text": {
"content": message
},
"safe": "0"
}
result = requests.post(url=url, data=json.dumps(data))
return result.text
if __name__ == '__main__':
wechat_info = Wechat_Info()
result = wechat_info.send_message('微信测试')
print(result)
(7)在utils包下,创建
aliyun_monitor.py
文件,封装阿里云的云监控数据的接口;
(8)然后编辑xkd.conf配置文件,添加阿里云
RAM access key
和
access key secret
、
region id
;
[Wechat]
party_id = 1
corp_id = wwabdfbc8feeee095a
secret = NY2nQE1cavlug_IuyMoRlIZDIOdsDS4KMpo_81XWslo
agent_id = 1000002
[ALIYUN]
access_key = LTAIxdvbuz2MD74Z
access_key_secret = I2b7O1UJmkjfgHVC4i7Atb8gGXwAbU
region_id = cn-shenzhen
aliyun_monitor.py
文件;
from aliyunsdkcore import client
from aliyunsdkcms.request.v20180308 import QueryMetricListRequest
from datetime import datetime, timedelta
import json
from configparser import ConfigParser
from utils.const_file import CONFIG_FILE
class CloudMonitor:
def __init__(self):
self.ram_info = self.__get_ram_info()
self.clt = client.AcsClient(self.ram_info.get('access_key'), self.ram_info.get('access_key_secret'), self.ram_info.get('region_id'))
self.request = QueryMetricListRequest.QueryMetricListRequest()
self.last_average = None
def __get_ram_info(self):
parser = ConfigParser()
parser.read(CONFIG_FILE)
ram_info = {}
ram_info['access_key'] = parser.get('ALIYUN', 'access_key')
ram_info['access_key_secret'] = parser.get('ALIYUN', 'access_key_secret')
ram_info['region_id'] = parser.get('ALIYUN', 'region_id')
return ram_info
def __get_start_timestamp(self):
start_datetime = datetime.now() - timedelta(minutes=3)
start_timestamp = int(start_datetime.timestamp() * 1000)
return start_timestamp
def get_last_metric(self,instance_id, monitor_key):
'''
:param instance_id: ECS是实例ID
:param monitor_key: 需要监控的指标
:return:
'''
self.request.set_accept_format('json')
self.request.set_Project('acs_ecs_dashboard')
self.request.set_Metric('{}'.format(monitor_key))
start_time = self.__get_start_timestamp()
self.request.set_StartTime(start_time)
self.request.set_EndTime(int(datetime.now().timestamp() * 1000))
self.request.set_Dimensions("{'instanceId': '%s'}" % (instance_id))
result = self.clt.do_action_with_exception(self.request)
result = json.loads(result)
data_str = result.get('Datapoints')
data_list = json.loads(data_str)
return data_list[-1]['Average']
if __name__ == '__main__':
monitor = CloudMonitor()
result = monitor.get_last_metric('i-wz98bynewgl7gu3jqqb3', 'CPUUtilization')
print(result)
pip install django-crontab
django-crontab
到INSTALLED_APPS;
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'users',
'courses',
'teachers',
'schools',
'rest_framework',
'django_crontab',
]
import CloudMonitor
和
Wechat_Info
;
from utils.aliyun_monitor import CloudMonitor
from utils.wechat_send import Wechat_Info
def send_aliyun_metric_to_wechat():
monitor = CloudMonitor()
wechat_info = Wechat_Info()
result = monitor.get_last_metric('i-wz98bynewgl7gu3jqqb3', 'CPUUtilization')
wechat_info.send_message('当前服务器CPU使用率是:{}'.format(result))
CRONJOBS = [
('* * * * *', 'users.cron.send_aliyun_metric_to_wechat')
]
python manage.py crontab add
;
安装命令行:
pip install configparser
;
ConfigParser是用来读取配置文件的模块,配置文件的格式跟windows下的ini配置文件相似,可以包含一个或多个section,每个section都有自己的option;
section 用 [sect_name] 表示,每个option是一个键值对,使用分隔符
=
或
:
隔开;
在 option 中分隔符两端的空格会被忽略掉;
配置文件使用可以使用
#
进行注释;
安装命令:
pip install django-crontab
django-crontab使用前需要添加到settings文件中的INSTALLED_APPS中;
添加定时任务:
python manage.py crontab add
,注意如果修改了任务,也需要再次运行此命令;
清除定时任务:
python manage.py crontab remove
;
显示定时任务:
python manage.py crontab show
;
注意
django-crontab
必须在Linux的crontab开启的情况下方可使用,不然会出现不执行的情况;
参考: /tupian/20230522/login.html
新闻标题:将获取的指标封装成接口提供API请求调用-创新互联
本文链接:https://www.cdcxhl.com/article16/dhojgg.html
成都网站建设公司_创新互联,为您提供微信公众号、建站公司、App开发、商城网站、用户体验、网站设计公司
声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联