飞书批量给openid发图片
#!/usr/bin/env python# --coding:utf-8--import datetimeimport osfrom http.server import BaseHTTPRequestHandler, HTTPServerfrom os import pathimport jsonfrom urllib import request, parseimport logging.c
·
#!/usr/bin/env python
# --coding:utf-8--
import datetime
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
from os import path
import json
from urllib import request, parse
import logging.config
import requests
logging.config.fileConfig("/Logger.conf")
logger = logging.getLogger('main')
def handle_request_url_verify(self, post_obj):
# 原样返回 challenge 字段内容
challenge = post_obj.get("challenge", "")
rsp = {'challenge': challenge}
self.response(json.dumps(rsp))
return
def handle_message(self, event):
# 此处只处理 text 类型消息，其他类型消息忽略
msg_type = event.get("msg_type", "")
if msg_type != "text":
print("unknown msg_type =", msg_type)
self.response("")
return
# 调用发消息 API 之前，先要获取 API 调用凭证：tenant_access_token
access_token = self.get_tenant_access_token()
if access_token == "":
self.response("")
return
# 处理
feishuuser=''
if event.get("open_id")=='ou_9f0e23df7d56ccb06beee7eaa9b4ce13':
feishuuser='lt'
else:
feishuuser="<at user_id=\""+event.get("open_id")+"\"></at>:"
# "恭喜中奖 <at user_id=\"ou_270xxxxxxx\"></at>"
msgfromfeishu=str(event.get("text"))
menu="0、help"+"\n"+"1、killsession"+"\n"+"2、tablespace"+"\n"+"3、AC01查询"
if msgfromfeishu=='0' or msgfromfeishu=="help":
msgtoferishu=menu
elif msgfromfeishu =='1' or msgfromfeishu.startswith('killsession'):
msgtoferishu=r"select inst_id,'ALTER SYSTEM KILL SESSION '''||sid||','||serial#||',@'||inst_id||''' IMMEDIATE;' from gv$session where username is not null and action like 'J_%' and program like '%(J%' order by logon_time, sid "
elif msgfromfeishu == '2' or msgfromfeishu.startswith('tablespace'):
msgtoferishu = r"select a.a1 表空间名称, c.c2 类型, c.c3 区管理, b.b2 / 1024 / 1024 表空间大小M, (b.b2 - a.a2) / 1024 / 1024 已使用M, substr((b.b2 - a.a2) / b.b2 * 100, 1, 5) 利用率 from (select tablespace_name a1, sum(nvl(bytes, 0)) a2 from dba_free_space group by tablespace_name) a, (select tablespace_name b1, sum(bytes) b2 from dba_data_files group by tablespace_name) b, (select tablespace_name c1, contents c2, extent_management c3 from dba_tablespaces) c where a.a1 = b.b1 and c.c1 = b.b1; "
else:
msgtoferishu="<at user_id=\""+event.get("open_id")+"\"></at>:"+msgfromfeishu+"\n可以试试以下指令：\n"+str(menu)
logger.info(feishuuser + ":" + msgfromfeishu)
# 机器人 echo 收到的消息
self.send_message(access_token, event.get("open_id"), str(msgtoferishu))
self.response("")
return
def response(self, body):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(body.encode())
def get_tenant_access_token():
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
headers = {
"Content-Type" : "application/json"
}
req_body = {
"app_id": APP_ID,
"app_secret": APP_SECRET
}
data = bytes(json.dumps(req_body), encoding='utf8')
req = request.Request(url=url, data=data, headers=headers, method='POST')
try:
response = request.urlopen(req)
except Exception as e:
print(e.read().decode())
return ""
rsp_body = response.read().decode('utf-8')
rsp_dict = json.loads(rsp_body)
code = rsp_dict.get("code", -1)
if code != 0:
print("get tenant_access_token error, code =", code)
return ""
return rsp_dict.get("tenant_access_token", "")
def upload_image(image_path):
"""上传图片
Args:
image_path: 文件上传路径
image_type: 图片类型
Return
{
"ok": true,
"image_key": "xxx",
"url": "https://xxx"
}
Raise:
Exception
* file not found
* request error
"""
with open(image_path, 'rb') as f:
image = f.read()
access_token = get_tenant_access_token()
print(access_token)
# headers = {
# "Content-Type": "application/json",
# "Authorization": "Bearer " + access_token
# }
resp = requests.post(
url='https://open.feishu.cn/open-apis/image/v4/put/',
headers={"Authorization": "Bearer " + access_token},
files={
"image": image
},
data={
"image_type": "message"
},
stream=True)
resp.raise_for_status()
content = resp.json()
print(content)
if content.get("code") == 0:
return content['data']['image_key']
else:
raise Exception("Call Api Error, errorCode is %s" % content["code"])
def upload_image1(image_path,access_token):
with open(image_path, 'rb') as f:
image = f.read()
# access_token = self.get_tenant_access_token()
# print(access_token)
# headers = {
# "Content-Type": "application/json",
# "Authorization": "Bearer " + access_token
# }
resp = requests.post(
url='https://open.feishu.cn/open-apis/image/v4/put/',
headers={"Authorization": "Bearer " + access_token},
files={
"image": image
},
data={
"image_type": "message"
},
stream=True)
resp.raise_for_status()
content = resp.json()
print(content)
if content.get("code") == 0:
return content['data']['image_key']
else:
raise Exception("Call Api Error, errorCode is %s" % content["code"])
def send_message( gmtitle, open_id, text):
url = "https://open.feishu.cn/open-apis/message/v4/send/"
# url="https://open.feishu.cn/open-apis/contact/v3/users/"
access_token = get_tenant_access_token()
print(access_token)
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer " + access_token
}
req_body = {
"open_id": open_id,
"msg_type": "text",
"content": {
"text": text
}
}
req_body = {
"open_id": open_id,
"msg_type": "text",
"content": {
"text": text
}
}
req_body = {
"open_id": open_id,
"msg_type": "interactive",
"card": {"config": {"wide_screen_mode": True},
"header": {"title": {"tag": "text", "content":gmtitle},
"template": "red"},
"elements": [{
"tag": "div",
"text": {
"tag": "lark_md",
"content": text
}
}]}
}
# getimg
# https://open.feishu.cn/open-apis/image/v4/get?image_key=24383920-9321-4ecd-8b33-bf8ce74e84c8
data = bytes(json.dumps(req_body), encoding='utf8')
req = request.Request(url=url, data=data, headers=headers, method='POST')
try:
response = request.urlopen(req)
except Exception as e:
print(e.read().decode())
return
rsp_body = response.read().decode('utf-8')
rsp_dict = json.loads(rsp_body)
print(rsp_dict)
code = rsp_dict.get("code", -1)
if code != 0:
print("send message error, code = ", code, ", msg =", rsp_dict.get("msg", ""))
def send_messagewithcmd( access_token, open_id, text,cmdstr,imgkey):
url = "https://open.feishu.cn/open-apis/message/v4/send/"
# access_token = self.get_tenant_access_token()
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer " + access_token
}
req_body = {
"open_id": open_id,
"msg_type": "text",
"content": {
"text": text
}
}
req_body = {
"open_id": open_id,
"msg_type": "interactive",
"card": {"config": {"wide_screen_mode": True},
"header": {"title": {"tag": "plain_text", "content": "<"+cmdstr+">指令结果:"},
"template": "red"},
"elements": [{
"tag": "div",
"text": {
"tag": "lark_md",
"content": text
}
}]}
}
if imgkey!='t-5882dca26fab7effd0e21c819b3fd39976bb480e':
req_body = {
"open_id": open_id,
"msg_type": "interactive",
"card": {"config": {"wide_screen_mode": True},
"header": {"title": {"tag": "plain_text", "content": "<" + cmdstr + ">指令结果:"},
"template": "red"},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": text
}
},
{"alt": {
"content": "",
"tag": "plain_text"
},
"img_key": imgkey,
"tag": "img"
}, {
"actions": [
{
"tag": "button",
"text": {
"content": "csdn学习",
"tag": "plain_text"
},
"type": "primary",
"url": "https://blog.csdn.net/viviliving"
}
],
"tag": "action"
}
]
}
}
else:
req_body = {
"open_id": open_id,
"msg_type": "interactive",
"card": {"config": {"wide_screen_mode": True},
"header": {"title": {"tag": "plain_text", "content": "<" + cmdstr + ">指令结果:"},
"template": "red"},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": text
}
},
{
"actions": [
{
"tag": "button",
"text": {
"content": "csdn学习",
"tag": "plain_text"
},
"type": "primary",
"url": "https://blog.csdn.net/viviliving"
}
],
"tag": "action"
}
]
}
}
data = bytes(json.dumps(req_body), encoding='utf8')
req = request.Request(url=url, data=data, headers=headers, method='POST')
try:
response = request.urlopen(req)
except Exception as e:
print(e.read().decode())
return
rsp_body = response.read().decode('utf-8')
rsp_dict = json.loads(rsp_body)
code = rsp_dict.get("code", -1)
if code != 0:
print("send message error, code = ", code, ", msg =", rsp_dict.get("msg", ""))
if __name__ == '__main__':
# run()
today = datetime.date.today().strftime('%Y%m%d')
dircmd = 'dir ' + 'F:\gp\\' + today + '*.png /b'
dircmd.replace('\\\\', '\\')
try:
pngfilename = os.popen(dircmd).read()
# p = subprocess.Popen(sqlloadercmd, stdout=subprocess.PIPE,)
# result = p.communicate()
# cont = result[0].decode('utf-8')
except Exception:
print(dircmd)
pngfilename = ''
# pngfilename = 'F:\\gp\\' + today + '(融资融券)差.png'
# return pngfilename
access_token = get_tenant_access_token()
# open_id = 'ou_8f523205dc3d95c9a1c8ba846f871d67' # xuexi cuixiang
# send_message('',open_id,pngfilename)
imgkey = 't-5882dca26fab7effd0e21c819b3fd39976bb480e'
send_messagewithcmd(access_token, open_id, (pngfilename), dircmd, imgkey)
listpng=pngfilename.split('\n')
for msgreturn in listpng:
# msgreturn='<img>20220120(融资融券)差.png'
if msgreturn=='':
break
imgkey = 't-5882dca26fab7effd0e21c819b3fd39976bb480e'
imgkey = upload_image1('F:\\gp\\'+msgreturn,access_token)
send_messagewithcmd(access_token, open_id, str(msgreturn), msgreturn, imgkey)
# self.send_messagewithcmd(access_token, event.get("open_id"), str(msgreturn), cmdmsg, str(msgreturn))
CSDN联合极客时间，共同打造面向开发者的精品内容学习社区，助力成长！
更多推荐
0
0- 0
已为社区贡献17条内容
所有评论(0)