Post

python-telegram-bot定义指令并执行shell

前言

项目打包比较麻烦,所以定义一个打包机器人接收指令,然后自动调用打包并将打好的包发送出来

小飞机接收指令并执行shell脚本ios_bot.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
#!/usr/bin/env python

"""
Don't forget to enable inline mode with @BotFather

1、创建机器人
2、设置开启机器人的inline mode
3、将机器人拉入到群里,并且设置为管理员

Usage:
停止执行代码:  Ctrl-C
"""

import logging
from uuid import uuid4
import os

from telegram import InlineQueryResultArticle, InputTextMessageContent, Update
from telegram.ext import Application, CommandHandler, ContextTypes, InlineQueryHandler, MessageHandler, filters

# 开启日志
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
# set higher logging level for httpx to avoid all GET and POST requests being logged
logging.getLogger("httpx").setLevel(logging.WARNING)

logger = logging.getLogger(__name__)

# 定义支持的App列表
suportAppList = [
    ("aa", "scheme1", "xx.xx.x1", "https://xxx/xxx/xx1.png"),
    ("bb", "scheme2", "xx.xx.x1", "https://xxx/xxx/xx2.png"),
    ("cc", "scheme3", "xx.xx.x1", "https://xxx/xxx/xx3.png"),
    ]

# 定义响应/start的指令
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
   
    message = update.message
    chatId = message.chat.id
   
    text = message.text
    # 如果输入的是模版,不处理
    if "/start scheme bundleId" in text:
        await update.message.reply_text("输入的是demo,需要真实的内容")
        return

    # 去掉多余的空格,方便后续使用
    while "  " in text:
        text = text.replace("  ", "")
    print(f"text2: {text}")

    # 查找/start命令的位置
    fromIdx = text.find('/start')
    if fromIdx >= 0:
        # 截取字符串,只去命令之外的内容
        realText = text[fromIdx+7:len(text)]
        print(f"realText: {realText}")
        findTxtList = realText.split(" ")
        print(f"paramsCount: {len(findTxtList)}")

        # 判断参数个数是否正确
        if len(findTxtList) != 2:
            await update.message.reply_text("输入的内容不对,必须是/start scheme bundleId 的格式")
            return
        scheme = findTxtList[0]
        bundleId = findTxtList[1]
        
        # 判断scheme是否正确
        fullSchemeList = []
        demoStr = ""
        # 查找支持的所有scheme
        for x in suportAppList:
            fullSchemeList.append(x[1])
            demoStr = f"{demoStr}\n{x[1]}"

        # 比较scheme是否正确
        if scheme not in fullSchemeList:
            await update.message.reply_text(f"输入的scheme内容: {scheme}不对,必须是如下的一个:\n{demoStr}")
            return
        
        # 找到App的名称
        appName = ""
        for x in suportAppList:
            if scheme in x[1]:
                appName = x[0]

        print(f"scheme: {scheme}")
        print(f"bundleId: {bundleId}")
        print(f"chatId: {chatId}")

        # 提示参数正确,开始准备执行
        await update.message.reply_text(f"Hi! 开始打包,大概耗时15分钟\nApp名称:{appName}\nscheme: {scheme} \nbundleId: {bundleId} \n默认打包分支:develop")
        commandStr = f"sh /xxx/xxx/xxx.sh {scheme} {bundleId} {chatId}"
        # 同步执行shell脚本,执行期间会堵塞其他命令
        os.system(commandStr)

# 响应一遍会话
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Echo the user message."""
    message = update.message
    chat = message.chat
    # if chat.id > 0:
    #     print("群")
    # elif chat.id < 0:
    #     print("单聊")
   
    text = message.text
    print(chat.id)
    print(text)

# 响应help指令
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send a message when the command /help is issued."""
    await update.message.reply_text("Help! \n方式一:/start scheme名 bundleId名,如\n/start qSportProduce com.p.xncsp\n\n方式二:在群里输入@xxxx_bot 选择打包对象")

# 响应inlinemode指令, @机器人的名字时出来
async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle the inline query. This is run when you type: @botusername <query>"""

    print("inline_query")
    query = update.inline_query.query

    findList = []
    print(f"query: {query}")
    if not query:
        # 没有搜索内容,就展示全部内容
        findList = suportAppList
    else:
        # 搜索出名称、scheme、bundleId任何一个包含的结果
        str1 = query.lower()
        for x in suportAppList:
            name = x[0]
            scheme = x[1]
            bundleId = x[2]
            
            # print(cmdStr)
            if (str1 in name) or (str1 in scheme) or (str1 in bundleId):
                findList.append(x)

    # 没有搜索出内容,则使用全部内容
    if len(findList) == 0:
        findList = suportAppList
    
    # 将搜索出来的内容转换为InlineQueryResultArticle
    results = []
    for x in findList:
        name = x[0]
        scheme = x[1]
        bundleId = x[2]
        url = x[3]
        cmdStr = f"/start {scheme} {bundleId}"
        fullTitle=f"{name} {scheme} {bundleId}"

        print(f"url: {url}")
        print(f"fullTitle: {fullTitle}")
        results.append(
            InlineQueryResultArticle(
                id=str(uuid4()),
                thumbnail_url=url,
                title=fullTitle,
                input_message_content=InputTextMessageContent(cmdStr),
            )
        )
    
    await update.inline_query.answer(results)

# 入口主函数
def main() -> None:
    """Run the bot."""
    # Create the Application and pass it your bot's token.
    current_file_path = __file__
    tokenStr = "xxxxxxxxxx"
    
    application = Application.builder().token(tokenStr).build()

    # on different commands - answer in Telegram
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))

    # on inline queries - show corresponding inline results
    application.add_handler(InlineQueryHandler(inline_query))
    # application.add_handler(ChosenInlineResultHandler(chosen_result))

     # on non command i.e message - echo the message on Telegram
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

    # Run the bot until the user presses Ctrl-C
    application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

# python3 /xxx/xxx/ios_bot.py

shell脚本执行打包archive.sh

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
export LC_ALL=en_US.UTF-8

# 不包含脚本自身
echo "输入的命令:$0 $@"
echo "参数个数:$#+1"
# 便利输入的参数
idx=0
echo "参数位置:$idx 参数内容: $0"
for i in $@; do
    idx=$(($idx+1))
    echo "参数位置:$idx 参数内容: $i"
done
# 0代表命令本身,1-9代表第1-9个参数 $0, $1, ... $9
# 10以上的参数要用大括号,如:${17}, ${10}, ...
# 读取当前脚本的路径
script_path=$0
script_dir_path=$(cd "$(dirname "$0")";pwd)
SCHEME=$1
newBundleId=$2
chat_id=$3
code_path="/xxx/xxx"
PROVISION_Name="xxxxx.mobileprovision"
ERTIFICATE="xxxxxx"
bot_token="xxxxxx"

echo "script_path: $script_path"
echo "script_dir_path: $script_dir_path"
echo "code_path: $code_path"
echo "SCHEME: $SCHEME"
echo "newBundleId: $newBundleId"
echo "chat_id: $chat_id"
echo "PROVISION_Name: $PROVISION_Name"
echo "ERTIFICATE: $ERTIFICATE"

project_name="xxx"
exportOptionsPath="${script_dir_path}/ExportOptions.plist"

DATE="$(date +%Y%m%d-%H%M%S)"
xcworkspace_path="$code_path/${project_name}.xcworkspace"

out_dir="${script_dir_path}/output_${DATE}"
if [ ! -d $out_dir ]; then
   # 没有文件夹,创建文件夹
   mkdir "${out_dir}"
else
    rm -rf "${out_dir}"
    mkdir "${out_dir}"
fi

#要上传的ipa文件路径
archive_xcarchive_path="${out_dir}/${DATE}.xcarchive"
archive_ipa_dir_path="${out_dir}/${DATE}"
archive_ipa_path="${archive_ipa_dir_path}/${SCHEME}.ipa"
echo "archive_xcarchive_path: $archive_xcarchive_path"
echo "archive_ipa_dir_path: $archive_ipa_dir_path"
echo "archive_ipa_path: $archive_ipa_path"

echo "=========config---end=============="

#### Clean ######
# xcodebuild clean -workspace "${xcworkspace_path}" \
# -scheme "${SCHEME}" \
# -configuration "Release"

#### Archive ######
xcodebuild archive -workspace "${xcworkspace_path}" \
-scheme "${SCHEME}" \
-configuration "Release" \
-archivePath "${archive_xcarchive_path}" -destination 'generic/platform=iOS'



### Export ipa #####
xcodebuild -exportArchive \
-archivePath "${archive_xcarchive_path}" \
-exportPath "${archive_ipa_dir_path}" \
-exportOptionsPlist "${exportOptionsPath}" \
-allowProvisioningUpdates

curl -s -X POST "https://api.telegram.org/bot${bot_token}/sendMessage?chat_id=${chat_id}" -d parse_mode="HTML" -d text="打包成功,开始重签名"

### 上传dSYM #####
pods_root_path="$code_path/Pods"

# "${pods_root_path}/FirebaseCrashlytics/upload-symbols" \
# -gsp "${code_path}/xxx/xxx/GoogleService-Info.plist" \
# -p ios \
# "${archive_xcarchive_path}/dSYMs"

# 重签名
userName=`whoami`
PROVISION="/Users/$userName/Library/MobileDevice/Provisioning Profiles/${PROVISION_Name}"

echo "ERTIFICATE: $ERTIFICATE"
echo "userName: $userName"
echo "PROVISION: $PROVISION"

# 
unzip_out="${script_dir_path}/output_${DATE}/tta"
echo "unzip_out: $unzip_out"
if [ ! -d $unzip_out ]; then
   # 没有文件夹,创建文件夹
   mkdir "${unzip_out}"
else
    rm -rf "${unzip_out}"
    mkdir "${unzip_out}"
fi

in_ipa_name="test.ipa"
in_ipa_path="$unzip_out/${in_ipa_name}"
cp "$archive_ipa_path" "$in_ipa_path"
cd $unzip_out
echo `pwd`
unzip -q "$in_ipa_name"

# 定位到 *.app 目录及 info.plist
appDir="$unzip_out/Payload/`ls "$unzip_out/"Payload`"
info_plist="${appDir}/Info.plist"
echo "appDir: $appDir"
echo "info_plist: $info_plist"

/usr/libexec/PlistBuddy -c "Set :CFBundleIdentifier $newBundleId" "$info_plist"
# /usr/libexec/PlistBuddy -c "Set :CFBundleDisplayName 'xyz'" "$info_plist"
appName2=`/usr/libexec/PlistBuddy -c "Print :CFBundleDisplayName" "$info_plist"`
appBundleId2=`/usr/libexec/PlistBuddy -c "Print :CFBundleIdentifier" "$info_plist"`

rm -rf "${appDir}/_CodeSignature/"
cp "$PROVISION" "${appDir}/embedded.mobileprovision"

# Extract entitlements from app
codesign -d --entitlements :entitlements.plist "${appDir}/"

# Re-sign embedded frameworks
codesign -f -s "$CERTIFICATE" --entitlements entitlements.plist "${appDir}/Frameworks/*"

codesign -f -s "$CERTIFICATE" --entitlements entitlements.plist "${appDir}/Plugins/*"

# Re-sign the app (with entitlements)
codesign -f -s "$CERTIFICATE" --entitlements entitlements.plist "${appDir}"

resign_ipa_name="${SCHEME_tail}_${DATE}_resigned.ipa"
echo "resign_ipa_name: $resign_ipa_name"
zip -qr $resign_ipa_name "Payload"

resign_ipa_path="${unzip_out}/${resign_ipa_name}"
echo "resign_ipa_path: $resign_ipa_path"

curl -s -X POST "https://api.telegram.org/bot${bot_token}/sendMessage?chat_id=${chat_id}" -d parse_mode="HTML" -d text="重签成功"

curl -s -X POST "https://api.telegram.org/bot${bot_token}/sendMessage?chat_id=${chat_id}" -d parse_mode="HTML" -d text="开始上传ipa包文件"
# 发送文件
desc="appName: ${appName2} SCHEME: ${SCHEME_tail} bundleId: ${appBundleId2}"
curl -s -X POST "https://api.telegram.org/bot${bot_token}/sendDocument" -F chat_id=$chat_id -H "Content-Type:multipart/form-data" -F document=@"${resign_ipa_path}" -F caption="$desc"

# Cleanup
# rm entitlements.plist
rm -r "${unzip_out}/Payload"

# update bundleId
# sh shellPath scheme newBundleId chatId
# sh /xxxx/xxx/archive.sh xxx xxx xxx

This post is licensed under CC BY 4.0 by the author.