ttt

GitlabTools.py

+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

+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

+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

+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

+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()