351 lines
11 KiB
Python
351 lines
11 KiB
Python
# import requests
|
|
import logging
|
|
import os
|
|
import re
|
|
import requests
|
|
import zipfile
|
|
import shutil
|
|
import uuid
|
|
|
|
import os
|
|
import json
|
|
|
|
from urllib import parse
|
|
|
|
APPS_COUNT = 0
|
|
SDKS_COUNT = 0
|
|
|
|
import json
|
|
import os
|
|
import lark_oapi as lark
|
|
from lark_oapi.api.im.v1 import *
|
|
|
|
sdk_external_dependencies = []
|
|
|
|
|
|
class CustomFormatter(logging.Formatter):
|
|
grey = "\x1b[38;20m"
|
|
yellow = "\x1b[33;20m"
|
|
red = "\x1b[31;20m"
|
|
bold_red = "\x1b[31;1m"
|
|
reset = "\x1b[0m"
|
|
green = "\x1b[32m"
|
|
|
|
format = "%(asctime)s: %(message)s"
|
|
|
|
FORMATS = {
|
|
logging.DEBUG: grey + format + reset,
|
|
logging.INFO: green + format + reset,
|
|
logging.WARNING: yellow + format + reset,
|
|
logging.ERROR: red + format + reset,
|
|
logging.CRITICAL: bold_red + format + reset,
|
|
}
|
|
date_format = "%Y-%m-%d %H:%M:%S"
|
|
|
|
def format(self, record):
|
|
log_fmt = self.FORMATS.get(record.levelno)
|
|
formatter = logging.Formatter(log_fmt, self.date_format)
|
|
return formatter.format(record)
|
|
|
|
|
|
# 初始化日志
|
|
logger = logging.getLogger("App")
|
|
logger.setLevel(logging.DEBUG)
|
|
ch = logging.StreamHandler()
|
|
ch.setFormatter(CustomFormatter())
|
|
logger.addHandler(ch)
|
|
|
|
|
|
def sanitize_filename(filename, replace_char="_"):
|
|
"""
|
|
去除 Windows 文件名中的非法字符
|
|
|
|
参数:
|
|
filename (str): 要处理的原始文件名
|
|
replace_char (str): 替换非法字符的字符，默认为下划线
|
|
|
|
返回:
|
|
str: 处理后的合法文件名
|
|
"""
|
|
# Windows 文件系统不允许的字符
|
|
illegal_chars = r'[<>:"/\\|?*\x00-\x1f]'
|
|
|
|
# 替换非法字符
|
|
sanitized = re.sub(illegal_chars, replace_char, filename)
|
|
|
|
# 同时去除文件名开头和结尾的空格和点
|
|
sanitized = sanitized.strip(" .")
|
|
|
|
# 保留文件扩展名中的点
|
|
return sanitized
|
|
|
|
|
|
def get_token(username, password):
|
|
"""
|
|
利用网页端获取 token
|
|
:param username: 用户名
|
|
:param password: 密码
|
|
:return: token
|
|
"""
|
|
|
|
url = "https://api.winrobot360.com/oauth/token"
|
|
data = {
|
|
"username": username,
|
|
"password": password,
|
|
"grant_type": "password",
|
|
"scope": "all",
|
|
}
|
|
|
|
payload = parse.urlencode(data)
|
|
headers = {
|
|
"accept": "application/json",
|
|
"authorization": "Basic Y29uc29sZTpzVVdpeTZzTmlYZDBlNUxo",
|
|
"content-type": "application/x-www-form-urlencoded;charset=UTF-8",
|
|
}
|
|
response = requests.request("POST", url, headers=headers, data=payload).json()
|
|
|
|
if response.get("success") == "false":
|
|
raise ValueError(response.get("msg"))
|
|
return response.get("access_token")
|
|
|
|
|
|
def get_client_version(token):
|
|
"""获取当期登录账号客户端的版本"""
|
|
url = "https://api.winrobot360.com/api/v1/user/info/rich"
|
|
|
|
payload = {}
|
|
headers = {
|
|
"Authorization": f"bearer {token}",
|
|
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
|
|
}
|
|
|
|
response = requests.request("GET", url, headers=headers, data=payload)
|
|
return response.json().get("data").get("clientVersion")
|
|
|
|
|
|
def get_activity_detail(token, sdk_uuid):
|
|
"""
|
|
获取自定义指令的相关信息
|
|
:param token: token
|
|
:param sdk: 自定义指令
|
|
:return: package_name, package_url
|
|
"""
|
|
name, version = sdk_uuid.split("==")
|
|
url = f"https://api.winrobot360.com/api/v1/robot/use/activity/package/detail?packageName={name}&version={version}"
|
|
payload = {}
|
|
headers = {
|
|
"Host": "api.winrobot360.com",
|
|
"Authorization": f"bearer {token}",
|
|
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
|
|
"User-Agent": "RestSharp/107.3.0.0",
|
|
"Accept-Encoding": "json",
|
|
}
|
|
|
|
response = requests.request("GET", url, headers=headers, data=payload)
|
|
return response.json().get("data")
|
|
|
|
|
|
def download_file(url, save_path, filename):
|
|
"""
|
|
根据链接下载文件
|
|
:param url: 链接
|
|
:param save_path: 保存路径
|
|
:param filename: 保存文件名
|
|
:return: 保存文件文件名
|
|
"""
|
|
res = requests.get(url)
|
|
with open(os.path.join(save_path, filename), "wb") as f:
|
|
f.write(res.content)
|
|
|
|
return os.path.join(save_path, filename)
|
|
|
|
|
|
def zip_file(startdir, file_news=None):
|
|
"""
|
|
压缩文件
|
|
:param startdir: 要压缩的文件夹路径
|
|
:param file_news: 压缩后文件名
|
|
"""
|
|
file_news = file_news if file_news else startdir + ".bot"
|
|
z = zipfile.ZipFile(file_news, "w", zipfile.ZIP_DEFLATED)
|
|
for dirpath, dirnames, filenames in os.walk(startdir):
|
|
fpath = dirpath.replace(startdir, "")
|
|
fpath = fpath and fpath + os.sep or ""
|
|
for filename in filenames:
|
|
z.write(os.path.join(dirpath, filename), fpath + filename)
|
|
z.close()
|
|
return file_news
|
|
|
|
|
|
def unzip_file(zip_file_path, unzip_path):
|
|
"""
|
|
解压文件
|
|
:param zip_file_path: 解压的文件路径
|
|
:param unzip_path: 解压（输出）的路径
|
|
:return:
|
|
"""
|
|
zip_file = zipfile.ZipFile(zip_file_path)
|
|
for file in zip_file.namelist():
|
|
zip_file.extract(file, unzip_path)
|
|
zip_file.close()
|
|
|
|
|
|
def modify_package_version(package_dirctory: str, version: str):
|
|
"""
|
|
修改 package.json 文件中的版本号
|
|
"""
|
|
package_json_path = os.path.join(package_dirctory, "package.json")
|
|
with open(package_json_path, encoding="utf8") as f:
|
|
package_json = json.load(f)
|
|
package_json["version"] = version
|
|
with open(package_json_path, "w") as f:
|
|
json.dump(package_json, f, indent=4, ensure_ascii=False)
|
|
|
|
|
|
def package_activity(token, robot_uuid, basedir):
|
|
"""
|
|
根据 uuid 将应用打包成 bot 文件, 并存到指定的文件中
|
|
:param token:
|
|
:param roobt_uuid:
|
|
:param basedir: 文件存放路径
|
|
"""
|
|
# 1. 根据机器人 uuid 获取机器人详细信息
|
|
activity_detail = get_activity_detail(token=token, sdk_uuid=robot_uuid)
|
|
activity_name = activity_detail.get("name")
|
|
version = activity_detail.get("releaseVersion")
|
|
packageUrl = activity_detail.get("packageUrl")
|
|
internal_dependencies = activity_detail.get("flatInternalDependencies")
|
|
|
|
logger.info(f"打包指令: {activity_name}")
|
|
|
|
# 去除不能作为windows文件名的字符
|
|
activity_name = sanitize_filename(activity_name, "")
|
|
|
|
# 2. 下载 activity 文件
|
|
activity_path = os.path.join(basedir, activity_name)
|
|
os.makedirs(activity_path)
|
|
download_name = download_file(packageUrl, activity_path, "xbot_robot.zip")
|
|
xbot_robot_path = os.path.join(activity_path, "xbot_robot")
|
|
os.makedirs(xbot_robot_path)
|
|
unzip_file(download_name, xbot_robot_path)
|
|
os.remove(download_name)
|
|
modify_package_version(xbot_robot_path, version)
|
|
|
|
# 3. 下载 activity 的 internal dependence
|
|
xbot_extensions_path = os.path.join(activity_path, "xbot_extensions")
|
|
os.makedirs(xbot_extensions_path)
|
|
|
|
with open(os.path.join(xbot_extensions_path, "requirement.txt"), "w") as f:
|
|
f.write("\n".join(internal_dependencies))
|
|
|
|
with open(os.path.join(xbot_extensions_path, "__init__.py"), "w") as f:
|
|
f.write("\n".join([f"from . import {item}" for item in internal_dependencies]))
|
|
|
|
for dependence in internal_dependencies:
|
|
dependence_detail = get_activity_detail(token, dependence)
|
|
name = dependence_detail.get("packageName")
|
|
version = dependence_detail.get("releaseVersion")
|
|
package_url = dependence_detail.get("packageUrl")
|
|
dependence_save_path = os.path.join(xbot_extensions_path, name)
|
|
os.makedirs(dependence_save_path)
|
|
download_name = download_file(package_url, xbot_extensions_path, "package.zip")
|
|
unzip_file(download_name, dependence_save_path)
|
|
modify_package_version(dependence_save_path, version)
|
|
os.remove(download_name)
|
|
|
|
activity_save_path = zip_file(activity_path)
|
|
|
|
shutil.rmtree(activity_path)
|
|
return activity_save_path
|
|
|
|
|
|
def send_message(file_path, open_id):
|
|
# 创建client
|
|
|
|
YOUR_APP_ID = "cli_a5110cd1da70900b"
|
|
YOUR_APP_SECRET = "XwTUlUq7dGK1XexUS7ljShJJ7t1MIwXF"
|
|
|
|
client = (
|
|
lark.Client.builder()
|
|
.app_id(YOUR_APP_ID)
|
|
.app_secret(YOUR_APP_SECRET)
|
|
.log_level(lark.LogLevel.DEBUG)
|
|
.build()
|
|
)
|
|
|
|
# 构造请求对象
|
|
file = open(file_path, "rb")
|
|
_, filename = os.path.split(file_path)
|
|
request: CreateFileRequest = (
|
|
CreateFileRequest.builder()
|
|
.request_body(
|
|
CreateFileRequestBody.builder()
|
|
.file_type("stream")
|
|
.file_name(filename)
|
|
.file(file)
|
|
.build()
|
|
)
|
|
.build()
|
|
)
|
|
|
|
# 发起请求
|
|
response: CreateFileResponse = client.im.v1.file.create(request)
|
|
|
|
# 处理失败返回
|
|
if not response.success():
|
|
lark.logger.error(
|
|
f"client.im.v1.file.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}"
|
|
)
|
|
|
|
content = {"file_key": response.data.file_key}
|
|
|
|
uuid_str = str(uuid.uuid4())
|
|
|
|
request: CreateMessageRequest = (
|
|
CreateMessageRequest.builder()
|
|
.receive_id_type("open_id")
|
|
.request_body(
|
|
CreateMessageRequestBody.builder()
|
|
.receive_id(open_id)
|
|
.msg_type("file")
|
|
.content(json.dumps(content))
|
|
.build()
|
|
)
|
|
.build()
|
|
)
|
|
# 发起请求
|
|
response: CreateMessageResponse = client.im.v1.message.create(request)
|
|
|
|
# 处理失败返回
|
|
if not response.success():
|
|
lark.logger.error(
|
|
f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}"
|
|
)
|
|
return
|
|
|
|
# 处理业务结果
|
|
lark.logger.info(lark.JSON.marshal(response.data, indent=4))
|
|
|
|
|
|
def push_xbot_activity_bot(activity_code, open_id):
|
|
"""自动推送官方指令bot包到指定用户
|
|
:param activity_code: 指令代码带版本号
|
|
:param open_id: 接收者open_id
|
|
"""
|
|
username = "剑心@fckj"
|
|
password = "320753691love"
|
|
# open_id = "ou_d542891996bf1961f9a94563e6bd4d49"
|
|
token = get_token(username=username, password=password)
|
|
|
|
logger.error(f"使用 {username} 账号登陆")
|
|
version = get_client_version(token=token)
|
|
|
|
logger.error(f"当前账号客户端版本号为: {version}")
|
|
|
|
base_dir = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
activity_path = package_activity(
|
|
token=token, robot_uuid=activity_code, basedir=base_dir
|
|
)
|
|
send_message(activity_path, open_id)
|