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 stacktrace_start = text.find("Stacktrace:") 19 if stacktrace_start != -1: 20 text = text[:stacktrace_start] 21 webhook_url = config ('TEAMS_WEBHOOK_URL') 22 current_time = (datetime.datetime.now ()).strftime ("%Y-%B-%d %H:%M:%S %p") 23 print (current_time, text) 24 screen_cap_img_paths = list (Path ('./ScreenCap').iterdir ()) 25 image_extensions = ['.jpg', '.jpeg', '.png', '.gif'] # 可根据需要添加其他图片文件扩展名 26 # 筛选有效的图片文件 27 image_paths = [str (path) for path in screen_cap_img_paths if path.suffix.lower ( 28 ) in image_extensions] 29 if image_paths: 30 screen_cap_img_path = image_paths[0] 31 filename = os.path.basename (screen_cap_img_path) 32 else: 33 screen_cap_img_path = '' 34 # print ("No valid image file found.") 35 36 if screen_cap_img_path != '': 37 # 提取文件名 38 filename = os.path.basename (screen_cap_img_path) 39 # 上传图片到阿里云，接收返回的图片URL 40 image_url = upload_image_to_aliyun (screen_cap_img_path) 41 # print ('image_url:' + image_url) 42 headers = {'Content-Type': 'application/json'} 43 data = { 44 "@type": "MessageCard", 45 "@context": "http://schema.org/extensions", 46 "themeColor": "0076D7", 47 "summary": "Summary", 48 "title": "??? Error Arrrrr!", 49 "sections": [{ 50 "text": f"{current_time} => {text} <br><img src=\"{image_url}\" alt=\"{filename}\">", 51 }], 52 } 53 response = requests.post (webhook_url, json=data, headers=headers) 54 if response.status_code == 200: 55 print ("Image sent successfully!") 56 else: 57 print ("Failed to send image to Teams.") 58 59 for f in list (Path ('./ScreenCap').iterdir ()): 60 os.remove (f) 61 else: 62 data = { 63 "text": f"{current_time} => {text}" 64 } 65 response = requests.post (webhook_url, json=data) 66 67 68 # 上传图片到阿里云 69 def upload_image_to_aliyun(image_path): 70 access_key_id = config ('ALI_ACCESS_KEY_ID') 71 access_key_secret = config ('ALI_ACCESS_KEY_SECRET') 72 endpoint = config ('ALI_ENDPOINT') 73 bucket_name = config ('ALI_BUCKET_NAME') 74 auth = oss2.Auth (access_key_id, access_key_secret) 75 bucket = oss2.Bucket (auth, endpoint, bucket_name) 76 # 设置存储桶的访问控制策略为公共读写 77 bucket.put_bucket_acl (oss2.BUCKET_ACL_PUBLIC_READ_WRITE) 78 # 生成图片上传路径 79 image_name = os.path.basename (image_path) 80 random_number = str (random.randint (100000, 999999)) 81 image_upload_path = f'cainiao_images/{datetime.datetime.now ().strftime ("%Y%m%d")}/{random_number}-{image_name}' 82 # 上传图片 83 with open (image_path, 'rb') as file: 84 image_data = file.read () 85 bucket.put_object (image_upload_path, image_data) 86 # 生成签名URL 87 signed_url = bucket.sign_url ( 88 'GET', image_upload_path, 60 * 60 * 24 * 365) # 有效期为一年 89 # 获取图片的URL 90 image_url = signed_url.split ('?')[0] 91 # 返回图片的URL 92 return image_url 93 94 95 # 删除阿里云整个文件夹的文件 96 def delete_folder_files(): 97 access_key_id = config ('ALI_ACCESS_KEY_ID') 98 access_key_secret = config ('ALI_ACCESS_KEY_SECRET') 99 endpoint = config ('ALI_ENDPOINT') 100 bucket_name = config ('ALI_BUCKET_NAME') 101 # 创建存储空间连接 102 auth = oss2.Auth (access_key_id, access_key_secret) 103 bucket = oss2.Bucket (auth, endpoint, bucket_name) 104 folder = f'cainiao_images/{datetime.datetime.now ().strftime ("%Y%m%d")}' 105 # 列举指定前缀的文件 106 for obj in oss2.ObjectIterator (bucket, prefix=folder): 107 bucket.delete_object (obj.key) 108 109 110 if __name__ == '__main__': 111 post_message_to_teams ("")