auto-mini-program-tool/server/main.py

import os
import requests
import json
import subprocess
import time
from flask import Flask,request,redirect
from flask_cors import CORS 
app = Flask(__name__,static_folder="static")
CORS(app, resources=r'/*')


# 操作文件，读取本地
def set_local_file_json(obj):
    with open("local.json","w",encoding="utf-8") as f:
        f.write(json.dumps(obj,ensure_ascii=False))
        f.close()

def get_local_file_json():
    try:
        with open("local.json","r",encoding="utf-8") as f:
            return json.loads(f.read())
    except:
        return {"status":0,"project":[]}


# current_path
current_path = os.getcwd()
res_json = {"code":0,"msg":"成功"}

global_params ={
    "token":"",
}

# 1.小程序打包
def build_mini_program(branch):
    command_str = (
    "cd "+current_path+ ";"
    "cd longVideoFactory;"
    "rm -rf dist;"
    "git checkout "+branch+";"
    "git pull;"
    "cnpm install && npm install ;"
    "npm run build;"
    )
    ret = subprocess.run(command_str,shell=True)
    if ret.returncode == 0:
        res_json["code"] = 0
        res_json["msg"] = "打包成功"
        
    else:
        res_json["code"] = 1
        res_json["msg"] = "打包失败"

    
# # 3.微信编译图片
def create_preview(project,page):
    # 小程序 cli path for mac
    wx_cli ="/Applications/wechatwebdevtools.app/Contents/MacOS/cli "
    
    dist_path =  os.path.join(current_path,'longVideoFactory/dist/'+project)
    code_img_path =  os.path.join(current_path,'1.png')
    
    path = """ '{"pathName":"%s"}' """ % page

    os.system(wx_cli+ " open --project " + dist_path)
    os.system(wx_cli+ " reset-fileutils --project " + dist_path)
    f_read = os.popen(wx_cli+" preview  --project "+dist_path+"  --compile-condition "+path+"   --qr-output "+code_img_path+" --qr-format image").read()
    os.system(wx_cli+ " quit")
    if "[error]" in f_read:
        res_json["code"] = 3
        res_json["msg"] = f_read
    else:
        res_json["code"] = 0
        res_json["msg"] = "微信小程序编译成功"

# 修改环境文件
def edit_mode_by_file(project,mode):
    path = os.path.join(current_path,"longVideoFactory/dist/"+project+"/network/volatileConfig.js")
    file_lines = []
    with open(path,"r") as f:
        for item in f.readlines():
            if "export const currentMode" in item:
                item = "export const currentMode = MODE."+mode+" \n"

            file_lines.append(item)
        f.close()
    
    with open(path,"w") as f2:
        f2.writelines(file_lines) 
        f2.close()
    res_json["code"] = 0
    res_json["msg"] = "文件修改成功"
    

def upload_img_by_feishu():
    token = global_params["token"]
    key = ""
    url = "https://open.feishu.cn/open-apis/im/v1/images"
    files = {
        "image_type":(None,"message"),
        "image":('1.png',open(os.path.join(current_path,"1.png"),"rb"),"image/png")
    }
    headers = {
        "Authorization":"Bearer "+token,
    }
    res = requests.post(url,headers=headers,files=files).json()
    if res["code"] == 0:
        global_params["img_key"] = res["data"]["image_key"]
        res_json["code"] = 0
    else:
        res_json["code"] = 4
        res_json["msg"] = "飞书上传图片失败"
        

def tenant_access_token_by_feishu():
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    params = {
        "app_id":"cli_a1ec07458838d00c",
        "app_secret":"Ngm0kfaLkZnpM2T2DH7L8cM5hdAqY0hI"
    }
    res = requests.post(url,data=params).json()
    if res["code"] == 0:
        global_params["token"] = res["tenant_access_token"]
        res_json["msg"] = "获取飞书签名成功"
        res_json["code"] = 0
    else:
        res_json["code"] = 4
        res_json["msg"] = "获取飞书签名失败"

def send_msg_by_feishu(chat_id,project_name,mode_name,branch):
    token = global_params["token"]
    img_key = global_params["img_key"]
    url ="https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"
    headers = {
        "Authorization":"Bearer "+token,
        "Content-Type": "application/json;charset=UTF-8"
    }
    params = {
        "receive_id":chat_id,
        "msg_type": "post",
        "content": json.dumps({
            "zh_cn": {
                "title": "小程序预览码",
                "content": [
                    [{
                        "tag":"text",
                        "text":"项目："+project_name
                    }],
                    [{
                        "tag":"text",
                        "text":"环境："+mode_name
                    }],
                    [{
                        "tag":"text",
                        "text":"分支："+branch
                    }],
                    [{
                            "tag": "img",
                            "image_key": img_key,
                            "width": 500,
                            "height": 500
                        }
                    ]]
                }
        })
    }
    res = requests.post(url,headers=headers,data=json.dumps(params)).json()
    if res["code"] == 0:
        res_json["code"] = 0
        res_json["msg"] = "飞书消息发送成功"
    else:
        res_json["code"] = 5
        res_json["msg"] = "飞书消息发送失败"


# 获取群
def get_group_by_feishu():
    token = global_params["token"]
    headers = {
        "Authorization":"Bearer "+token,
        "Content-Type": "application/json;charset=UTF-8"
    }
    url = "https://open.feishu.cn/open-apis/im/v1/chats"
    res = requests.get(url,headers=headers).json()
    data = []
    for item in res["data"]["items"]:
        if item["name"]  != '':
            data.append(item)
    return data



@app.route('/')
def func():    
    # return 'Welcome to auto-mini-program-build-preview server. only can run on MacOS.'
    return redirect("static/dist/index.html")

@app.route('/getProject')
def getProject():
    local_data = get_local_file_json()
    res_json["code"] = 0
    res_json["data"] = local_data["project"]
    return res_json

@app.route('/getGroup')
def getGroup():
    tenant_access_token_by_feishu()
    res_json["code"] = 0
    res_json["data"] = get_group_by_feishu()
    return res_json

@app.route('/send',methods=['POST'])
def send():
    local_data = get_local_file_json()
    try:
        if local_data["status"] == "1":
            res_json["code"] = 9
            res_json["msg"] = "当前有人打包，请稍后尝试"
            return res_json
        
        local_data["status"] = '1'
        set_local_file_json(local_data)
        res_json["code"] = 0
        res_json["msg"] = "成功"

        # 接受参数
        chat_id = request.json["chat_id"]
        branch = request.json["branch"]
        page = request.json["page"]
        project = request.json["project"]
        mode = request.json["mode"]
        project_name = request.json["project_name"]
        mode_name = request.json["mode_name"]

        func_arr = [
            # ------- 微信打包 ------
            # 1.小程序打包
            build_mini_program(branch),
            # 2.修改环境
            edit_mode_by_file(project,mode),
            time.sleep(1),
            create_preview(project,page),
            # -------- 飞书 --------
            # 1.获取签名
            tenant_access_token_by_feishu(),

            upload_img_by_feishu(),
            send_msg_by_feishu(chat_id,project_name,mode_name,branch)
        ]

        # 循环方法判断
        for item in func_arr:
            item
            if res_json["code"] != 0:
                local_data["status"] = '0'
                set_local_file_json(local_data)
                return res_json

        # 正常执行完
        local_data["status"] = '0'
        set_local_file_json(local_data)

    except Exception as ex:
        local_data["status"] = '0'
        set_local_file_json(local_data)
        res_json["code"] = 9
        print(ex)
        res_json["msg"] = ex.args

    return res_json


@app.route("/getBranch")
def get_branch():
    # 查看所有分支
    command_str= (
        "cd "+current_path+ ";"
        "git clone https://git.yishihui.com/weapp/longVideoFactory.git;"
        "cd longVideoFactory;"
        "git pull;"   
    )
    os.system(command_str)

    command_str= (
        "cd "+current_path+ ";"
        "cd longVideoFactory;"
        "git branch -r;"
    )
    f_read = os.popen(command_str).read()

    branch_list = []
    for item in f_read.split():
        if "->" not in item and "HEAD" not in item and "*" not in item:
            branch_list.append(item.replace("origin/",""))

    res = {"code":0,"data":branch_list}
    return res


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=7777, debug=True)