Skip to content

T
test_webhook
Commit 9d153fa2 authored by 时海鑫's avatar 时海鑫
Browse files
Options

Merge branch 'dev' into 'master' 

ttt

See merge request !13
parents 8a07ecfc 68f57af1
Hide whitespace changes
Inline Side-by-side
Showing with 517 additions and 0 deletions
GitlabTools.py 0 → 100644
View file @ 9d153fa2
import requests
GITLAB_URL = "http://tyw-server.synology.me:12345"
GITLAB_API_TOKEN = "3WxJ46oPKLTeEmDQAPp8"
GITLAB_PROJECT_ID = "1556"
BUILD_FILE = "files/output.zip"
LOCAL_REPO_PATH = "dist"
TAG_NAME = "v1.0.1"
BRANCH_NAME = "xxx"
# 你的配置
PROJECT_ID = 1556 # GitLab 项目 ID
PRIVATE_TOKEN = "3WxJ46oPKLTeEmDQAPp8"
# API URL
url = f"{GITLAB_URL}/api/v4/projects/{PROJECT_ID}/members/all"
headers = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
# 获取成员信息
response = requests.get(url, headers=headers)
members = response.json()
for member in members:
user_id = member["id"]
username = member["username"]
# 再查每个成员的用户详情（才能获取邮箱）
user_url = f"{GITLAB_URL}/api/v4/users/{user_id}"
user_res = requests.get(user_url, headers=headers)
if user_res.status_code == 200:
user_data = user_res.json()
email = user_data.get("email", "未知")
print(f"{username}: {email}")
else:
print(f"{username}: ❌ 无法获取邮箱")
IARBuildTools.py 0 → 100644
View file @ 9d153fa2
import subprocess
from pathlib import Path
# 定义不同版本的 iarbuild.exe 路径
IARBUILD_VERSIONS = {
1: r"C:\Program Files (x86)\IAR Systems\Embedded Workbench 9.0\common\bin\iarbuild.exe",
2: r"C:\Program Files (x86)\IAR Systems\Embedded Workbench 8.4\common\bin\iarbuild.exe",
3: r"C:\Program Files\IAR Systems\Embedded Workbench 9.1\common\bin\iarbuild.exe"
}
def build_iar_project(ewp_path, build_type, build_mode="Debug"):
"""
使用指定版本的 iarbuild.exe 编译项目
参数:
ewp_path: IAR 项目文件路径 (str 或 Path)
build_type: 编译器版本 (1, 2, 3 等)
build_mode: 编译模式 (Debug/Release)
返回:
tuple: (success: bool, output: str, error: str)
"""
# 获取对应版本的编译器路径
iarbuild_path = IARBUILD_VERSIONS.get(build_type)
if not iarbuild_path:
return False, "", f"Invalid build type: {build_type}"
if not Path(iarbuild_path).exists():
return False, "", f"iarbuild.exe not found at: {iarbuild_path}"
if not Path(ewp_path).exists():
return False, "", f"Project file not found: {ewp_path}"
# 执行编译命令
result = subprocess.run(
[iarbuild_path, str(ewp_path), "-build", build_mode, "-log", "errors"],
capture_output=True,
text=True
)
# 返回编译结果
return (
result.returncode == 0,
result.stdout,
result.stderr
)
# 示例调用
if __name__ == "__main__":
# 示例参数
project_file = r".\C655_V0.01\IARProject\C655.ewp"
selected_version = 1 # 1 对应 IAR 9.0
# 调用编译函数
success, output, error = build_iar_project(project_file, selected_version)
# 处理返回结果（这部分由调用者决定）
print(f"Success: {success}")
if output:
print(f"Output:\n{output}")
if error:
print(f"Error:\n{error}")
# 可以根据 success 决定后续操作
if not success:
exit(1)
\ No newline at end of file
SendEmailTools.py 0 → 100644
View file @ 9d153fa2
import base64
import datetime
import smtplib
from email.header import Header
from email.mime.text import MIMEText
# def sendEmailnew(log, text, info, receiver_email):
# smtp_server = 'smtp.qq.com'
# port = 465
# sender_email = '842463468@qq.com'
# receiver_email = receiver_email
# password = 'kswtirrkzidibbeg'
# message = MIMEText(str(datetime.date.today()) + '银行招聘信息更新有更新请查看' + text + '\n' + info, 'plain',
# 'utf-8')
# message['From'] = '=?UTF-8?B?' + base64.b64encode(text.encode()).decode() + '?= <842463468@qq.com>'
# message['Subject'] = Header(str(datetime.date.today()) + ' - ' + text, 'utf-8')
# try:
# with smtplib.SMTP_SSL(smtp_server, port) as server:
# server.login(sender_email, password)
# server.sendmail(sender_email, receiver_email, message.as_string())
# log.write(str(datetime.date.today()) +" - "+receiver_email+ ' - :邮件发送成功' + '\n')
# except Exception as e:
# log.write(str(datetime.date.today()) +" - "+receiver_email+ ' - :' + "邮件发送失败: " + str(e) + '\n')
def sendEmail2(text, info, receiver_email):
smtp_server = 'smtp.hljtyw.com'
port = 465
sender_email = 'haixin.shi@hljtyw.com'
receiver_email = receiver_email
password = 'lGN9lD8pDtoatlca'
message = MIMEText(str(datetime.date.today()) + '产品升级更新' + text + '\n' + info, 'plain',
'utf-8')
message['From'] = '=?UTF-8?B?' + base64.b64encode(text.encode()).decode() + '?= <haixin.shi@hljtyw.com>'
message['Subject'] = Header(str(datetime.date.today()) + ' - ' + text, 'utf-8')
try:
with smtplib.SMTP_SSL(smtp_server, port) as server:
server.login(sender_email, password)
server.sendmail(sender_email, receiver_email, message.as_string())
print(str(datetime.date.today()) +" - "+receiver_email+ ' - :邮件发送成功' + '\n')
except Exception as e:
print(str(datetime.date.today()) +" - "+receiver_email+ ' - :' + "邮件发送失败: " + str(e) + '\n')
if __name__ == '__main__':
sendEmail2(text='长安汽车',info="版本号升级为vv1.0.0",receiver_email="842463468@qq.com")
\ No newline at end of file
UploadFile2git.py 0 → 100644
View file @ 9d153fa2
import os
import shutil
import stat
import subprocess
import requests
from model.RecvData import GitlabPayload
def bk():
# ======== 配置信息 ========
GITLAB_URL = "http://tyw-server.synology.me:12345"
GITLAB_API_TOKEN = "3WxJ46oPKLTeEmDQAPp8"
GITLAB_PROJECT_ID = "1556"
BUILD_FILE = "files/output.zip"
LOCAL_REPO_PATH = "dist"
TAG_NAME = "v1.0.1"
BRANCH_NAME = "xxx"
# 带用户名和 token 的 clone 地址（推荐 token 代替密码）
repo_url = f"http://oauth2:{GITLAB_API_TOKEN}@tyw-server.synology.me:12345/shihaixin/autocompliecode.git"
# ======== 工具函数 ========
def remove_readonly(func, path, _):
os.chmod(path, stat.S_IWRITE)
func(path)
def tag_exists(tag_name):
result = subprocess.run(["git", "-C", LOCAL_REPO_PATH, "tag"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
tag_output = result.stdout.decode()
return tag_name in tag_output.splitlines()
# ======== 步骤 1: 拉取代码 ========
if os.path.exists(LOCAL_REPO_PATH):
shutil.rmtree(LOCAL_REPO_PATH, onerror=remove_readonly)
print(f"🧹 已删除目录: {LOCAL_REPO_PATH}")
os.makedirs(LOCAL_REPO_PATH)
print(f"✅ 已重新创建目录: {LOCAL_REPO_PATH}")
subprocess.run(["git", "clone", "-b", BRANCH_NAME, repo_url, LOCAL_REPO_PATH], check=True)
print(f"✅ 仓库克隆完成: {repo_url} (分支: {BRANCH_NAME})")
# ======== 步骤 2: 模拟构建产物 ========
print("🚧 模拟编译中...")
if not os.path.exists(BUILD_FILE):
with open(BUILD_FILE, "w") as f:
f.write("This is a mock build file.\n")
print(f"✅ 编译完成，生成产物: {BUILD_FILE}")
# ======== 步骤 3: 创建 tag 并推送 ========
if tag_exists(TAG_NAME):
print(f"⚠️ Git tag '{TAG_NAME}' 已存在，跳过创建。")
else:
subprocess.run(["git", "-C", LOCAL_REPO_PATH, "tag", TAG_NAME], check=True)
subprocess.run(["git", "-C", LOCAL_REPO_PATH, "push", "origin", TAG_NAME], check=True)
print(f"🏷️ Git tag '{TAG_NAME}' created and pushed.")
# ======== 步骤 4: 上传构建产物 ========
with open(BUILD_FILE, "rb") as f:
files = {"file": (os.path.basename(BUILD_FILE), f)}
headers = {"PRIVATE-TOKEN": GITLAB_API_TOKEN}
upload_url = f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}/uploads"
response = requests.post(upload_url, headers=headers, files=files)
if response.status_code != 201:
print(f"❌ 上传构建产物失败: {response.text}")
exit(1)
file_info = response.json()
file_url = file_info["url"]
print(f"✅ 文件上传成功: {file_url}")
# ======== 步骤 5: 创建或更新 Release ========
release_payload = {
"name": f"Release {TAG_NAME}",
"tag_name": TAG_NAME,
"description": f"Auto released for tag {TAG_NAME}",
"assets": {
"links": [
{
"name": os.path.basename(BUILD_FILE),
"url": f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}{file_url}"
}
]
}
}
release_url = f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}/releases"
release_res = requests.post(release_url, headers=headers, json=release_payload)
if release_res.status_code == 409:
# 已存在 Release，更新之
update_url = f"{release_url}/{TAG_NAME}"
release_res = requests.put(update_url, headers=headers, json=release_payload)
if release_res.ok:
print(f"🚀 Release 发布成功: {TAG_NAME}")
else:
print(f"❌ Release 发布失败: {release_res.status_code} - {release_res.text}")
def creatTagAndRelease(recv_data: GitlabPayload, confDict: dict, url: str):
# url=shihaixin/autocompliecode.git
# ======== 配置信息 ========
GITLAB_URL = "http://tyw-server.synology.me:12345" # 地址可以写死，暂时不换
GITLAB_API_TOKEN = "3WxJ46oPKLTeEmDQAPp8" # 需要替换，当前是我的token
GITLAB_PROJECT_ID = recv_data.metadata.project_id
BUILD_FILE = confDict['jar_project_url']
LOCAL_REPO_PATH = confDict['build_dist_url'] + "//dist"
TAG_NAME = confDict['tag_name']
BRANCH_NAME = recv_data.commit.branch
# 带用户名和 token 的 clone 地址（推荐 token 代替密码）
repo_url = f"http://oauth2:{GITLAB_API_TOKEN}@tyw-server.synology.me:12345/{url}"
# ======== 工具函数 ========
def remove_readonly(func, path, _):
os.chmod(path, stat.S_IWRITE)
func(path)
def tag_exists(tag_name):
result = subprocess.run(["git", "-C", LOCAL_REPO_PATH, "tag"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
tag_output = result.stdout.decode()
return tag_name in tag_output.splitlines()
# ======== 步骤 1: 拉取代码 ========
if os.path.exists(LOCAL_REPO_PATH):
shutil.rmtree(LOCAL_REPO_PATH, onerror=remove_readonly)
print(f"🧹 已删除目录: {LOCAL_REPO_PATH}")
os.makedirs(LOCAL_REPO_PATH)
print(f"✅ 已重新创建目录: {LOCAL_REPO_PATH}")
subprocess.run(["git", "clone", "-b", BRANCH_NAME, repo_url, LOCAL_REPO_PATH], check=True)
print(f"✅ 仓库克隆完成: {repo_url} (分支: {BRANCH_NAME})")
# ======== 步骤 2: 模拟构建产物 ========
print("🚧 模拟编译中...")
if not os.path.exists(BUILD_FILE):
with open(BUILD_FILE, "w") as f:
f.write("This is a mock build file.\n")
print(f"✅ 编译完成，生成产物: {BUILD_FILE}")
if confDict.get('tag') == 1:
# ======== 步骤 3: 创建 tag 并推送 ========
if tag_exists(TAG_NAME):
print(f"⚠️ Git tag '{TAG_NAME}' 已存在，跳过创建。")
else:
subprocess.run(["git", "-C", LOCAL_REPO_PATH, "tag", TAG_NAME], check=True)
subprocess.run(["git", "-C", LOCAL_REPO_PATH, "push", "origin", TAG_NAME], check=True)
print(f"🏷️ Git tag '{TAG_NAME}' created and pushed.")
release_status_code = 0
release_status_text = ""
if confDict.get('release') == '1':
# ======== 步骤 4: 上传构建产物 ========
with open(BUILD_FILE, "rb") as f:
files = {"file": (os.path.basename(BUILD_FILE), f)}
headers = {"PRIVATE-TOKEN": GITLAB_API_TOKEN}
upload_url = f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}/uploads"
response = requests.post(upload_url, headers=headers, files=files)
if response.status_code != 201:
print(f"❌ 上传构建产物失败: {response.text}")
exit(1)
file_info = response.json()
file_url = file_info["url"]
print(f"✅ 文件上传成功: {file_url}")
# ======== 步骤 5: 创建或更新 Release ========
release_payload = {
"name": f"Release {TAG_NAME}",
"tag_name": TAG_NAME,
"description": f"Auto released for tag {TAG_NAME}",
"assets": {
"links": [
{
"name": os.path.basename(BUILD_FILE),
"url": f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}{file_url}"
}
]
}
}
release_url = f"{GITLAB_URL}/api/v4/projects/{GITLAB_PROJECT_ID}/releases"
release_res = requests.post(release_url, headers=headers, json=release_payload)
if release_res.status_code == 409:
# 已存在 Release，更新之
update_url = f"{release_url}/{TAG_NAME}"
release_res = requests.put(update_url, headers=headers, json=release_payload)
release_status_code = release_res.ok
release_status_text = release_res.text
if release_status_code:
print(f"🚀 Release 发布成功: {TAG_NAME}")
return {
'code': 0,
'message': "消息发送成功",
'error_message': ""
}
else:
print(f"❌ Release 发布失败: {release_status_code} - {release_status_text}")
return {
'code': release_status_code,
'message': "消息发送失败",
'error_message': f"{release_status_text}"
}
if __name__ == '__main__':
creatTagAndRelease()
WxBootTools.py 0 → 100644
View file @ 9d153fa2
import requests
import json
from model import RecvData
from model.RecvData import GitlabPayload
def test():
webhook_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=752b9124-5c68-41ed-99b4-3607eab8349c'
data = {
"msgtype": "template_card",
"template_card": {
"card_type": "text_notice",
"source": {
"icon_url": "https://wework.qpic.cn/wwpic/252813_jOfDHtcISzuodLa_1629280209/0",
"desc": "软件二部",
"desc_color": 0
},
"main_title": {
"title": "长安655项目",
"desc": "master 分支"
},
"emphasis_content": {
"title": "V1.0.0",
"desc": "当前版本"
},
"quote_area": {
"type": 1,
"url": "https://work.weixin.qq.com/?from=openApi",
"appid": "APPID",
"pagepath": "PAGEPATH",
"title": "本次升级内容",
"quote_text": "无"
},
"horizontal_content_list": [{
"keyname": "打包人",
"value": "时先生"
},
{
"keyname": "项目地址",
"value": "点击访问",
"type": 1,
"url": "http://tyw-server.synology.me:12345/shihaixin/autocompliecode"
},
{
"keyname": "Jobs地址",
"value": "点击访问",
"type": 1,
"url": "http://tyw-server.synology.me:12345/shihaixin/autocompliecode/-/jobs"
}
],
"jump_list": [{
"type": 1,
"url": "http://tyw-server.synology.me:12345/shihaixin/autocompliecode",
"title": "项目地址"
}
],
"card_action": {
"type": 1,
"url": "http://tyw-server.synology.me:12345/shihaixin/autocompliecode"
}
}
}
headers = {'Content-Type': 'application/json'}
response = requests.post(webhook_url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
print("✅ 消息发送成功")
else:
print(f"❌ 消息发送失败: {response.status_code}, {response.text}")
def sendWx(recv_data: GitlabPayload, confDict: dict, url: str):
webhook_url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={confDict['wechat_webhook']}"
data = {
"msgtype": "template_card",
"template_card": {
"card_type": "text_notice",
"source": {
"icon_url": "https://wework.qpic.cn/wwpic/252813_jOfDHtcISzuodLa_1629280209/0",
"desc": "软件二部",
"desc_color": 0
},
"main_title": {
"title": f"{recv_data.metadata.project_name}",
"desc": f"{recv_data.commit.branch} 分支"
},
"emphasis_content": {
"title": f"{confDict['version_name']}",
"desc": "当前版本"
},
"quote_area": {
"type": 1,
"url": "https://work.weixin.qq.com/?from=openApi",
"appid": "APPID",
"pagepath": "PAGEPATH",
"title": "本次升级内容",
"quote_text": f"{recv_data.commit.message}"
},
"horizontal_content_list": [{
"keyname": "打包人",
"value": f"{recv_data.metadata.user_login}"
},
{
"keyname": "项目地址",
"value": "点击访问",
"type": 1,
"url": f"{url.replace('.git','')}"
},
{
"keyname": "Jobs地址",
"value": "点击访问",
"type": 1,
"url": f"{url.replace('.git','')+'/-/jobs'}"
}
],
"jump_list": [{
"type": 1,
"url": f"{url.replace('.git','')}",
"title": "项目地址"
}
],
"card_action": {
"type": 1,
"url":f"{url.replace('.git','')}"
}
}
}
headers = {'Content-Type': 'application/json'}
response = requests.post(webhook_url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
print("✅ 消息发送成功")
return {
'code': 0,
'message': "消息发送成功",
'error_message': ""
}
else:
print(f"❌ 消息发送失败: {response.status_code}, {response.text}")
return {
'code': response.status_code,
'message': "消息发送失败",
'error_message': f"{response.text}"
}
if __name__ == '__main__':
# sendWx()
test()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment