python发送消息到Teams以及阿里云的上传与删除-526互联

  1 import urllib
  2 import time
  3 import oss2
  4 import os
  5 import random
  6 from pathlib import Path
  7 import requests
  8 import datetime
  9 from decouple import config
 10 
 11 from urllib.parse import unquote
 12 
 13 
 14 # ? Collision ? Party Popper ? Partying Face ? Police Car Light ? Waving Hand ?? 鼓掌
 15 
 16 # 发送消息到Teams
 17 def post_message_to_teams(text):
 18     webhook_url = config('TEAMS_WEBHOOK_URL')
 19     current_time = (datetime.datetime.now()).strftime("%Y-%B-%d %H:%M:%S %p")
 20     print(current_time, text)
 21     screen_cap_img_paths = list(Path('./ScreenCap').iterdir())
 22     image_extensions = ['.jpg', '.jpeg', '.png', '.gif']  # 可根据需要添加其他图片文件扩展名
 23     # 筛选有效的图片文件
 24     image_paths = [str(path) for path in screen_cap_img_paths if path.suffix.lower(
 25     ) in image_extensions]
 26     if image_paths:
 27         screen_cap_img_path = image_paths[0]
 28         filename = os.path.basename(screen_cap_img_path)
 29     else:
 30         screen_cap_img_path = ''
 31         print("No valid image file found.")
 32 
 33     if screen_cap_img_path != '':
 34         # 提取文件名
 35         filename = os.path.basename(screen_cap_img_path)
 36         # 上传图片到阿里云，接收返回的图片URL
 37         image_url = upload_image_to_aliyun(screen_cap_img_path)
 38         print('image_url:' + image_url)
 39         headers = {'Content-Type': 'application/json'}
 40         data = {
 41             "@type": "MessageCard",
 42             "@context": "http://schema.org/extensions",
 43             "themeColor": "0076D7",
 44             "summary": "Summary",
 45             "title": "??? Error Arrrrr!",
 46             "sections": [{
 47                 "text": f"{current_time} => {text} <br><img src=\"{image_url}\" alt=\"{filename}\">",
 48             }],
 49         }
 50         response = requests.post(webhook_url, json=data, headers=headers)
 51         if response.status_code == 200:
 52             print("Image sent successfully!")
 53         else:
 54             print("Failed to send image to Teams.")
 55 
 56         for f in list(Path('./ScreenCap').iterdir()):
 57             os.remove(f)
 58     else:
 59         data = {
 60             "text": f"{current_time} => {text}"
 61         }
 62         response = requests.post(webhook_url, json=data)
 63 
 64 
 65 # 上传图片到阿里云
 66 def upload_image_to_aliyun(image_path):
 67     access_key_id = config('ALI_ACCESS_KEY_ID')
 68     access_key_secret = config('ALI_ACCESS_KEY_SECRET')
 69     endpoint = config('ALI_ENDPOINT')
 70     bucket_name = config('ALI_BUCKET_NAME')
 71     auth = oss2.Auth(access_key_id, access_key_secret)
 72     bucket = oss2.Bucket(auth, endpoint, bucket_name)
 73     # 设置存储桶的访问控制策略为公共读写
 74     bucket.put_bucket_acl(oss2.BUCKET_ACL_PUBLIC_READ_WRITE)
 75     # 生成图片上传路径
 76     image_name = os.path.basename(image_path)
 77     random_number = str(random.randint(100000, 999999))
 78     image_upload_path = f'cainiao_images/{datetime.datetime.now ().strftime ("%Y%m%d")}/{random_number}-{image_name}'
 79     # 上传图片
 80     with open(image_path, 'rb') as file:
 81         image_data = file.read()
 82         bucket.put_object(image_upload_path, image_data)
 83     # 生成签名URL
 84     signed_url = bucket.sign_url(
 85         'GET', image_upload_path, 60 * 60 * 24 * 365)  # 有效期为一年
 86     # 获取图片的URL
 87     image_url = signed_url.split('?')[0]
 88     # 返回图片的URL
 89     return image_url
 90 
 91 
 92 # 删除阿里云的图片
 93 def delete_image_from_aliyun(image_path):
 94     access_key_id = config('ALI_ACCESS_KEY_ID')
 95     access_key_secret = config('ALI_ACCESS_KEY_SECRET')
 96     endpoint = config('ALI_ENDPOINT')
 97     bucket_name = config('ALI_BUCKET_NAME')
 98     # 创建存储空间连接
 99     auth = oss2.Auth(access_key_id, access_key_secret)
100     bucket = oss2.Bucket(auth, endpoint, bucket_name)
101     # 删除文件
102     object_name = image_path.split('/')[-1]
103     decoded_image_path = unquote(object_name)
104     try:
105         # 彻底删除文件
106         bucket.delete_object(decoded_image_path, params={
107             'x-oss-process': 'image/resize,w_0'})
108         print("文件删除成功")
109     except oss2.exceptions.NoSuchKey:
110         print("文件不存在")
111     except oss2.exceptions.OssError as e:
112         print("文件删除失败:", e)
113 
114 
115 # 删除阿里云整个文件夹的文件
116 def delete_folder_files():
117     access_key_id = config('ALI_ACCESS_KEY_ID')
118     access_key_secret = config('ALI_ACCESS_KEY_SECRET')
119     endpoint = config('ALI_ENDPOINT')
120     bucket_name = config('ALI_BUCKET_NAME')
121     # 创建存储空间连接
122     auth = oss2.Auth(access_key_id, access_key_secret)
123     bucket = oss2.Bucket(auth, endpoint, bucket_name)
124     folder = f'cainiao_images/{datetime.datetime.now ().strftime ("%Y%m%d")}'
125     # 列举指定前缀的文件
126     for obj in oss2.ObjectIterator(bucket, prefix=folder):
127         bucket.delete_object(obj.key)
128 
129 
130 if __name__ == '__main__':
131     post_message_to_teams("")
View Code
消息
python