Files
jianxin 8fe6822545
continuous-integration/drone/push Build is passing
add check activity help link
2024-10-22 19:36:08 +08:00

659 lines
22 KiB
Python

import os
import requests
import dotenv
from urllib import parse
import os
import json
import dotenv
import random
import zipfile
import requests
import uuid
import sys
import hashlib
import shutil
import time
import requests, warnings
from requests.packages import urllib3
# 关闭警告
urllib3.disable_warnings()
warnings.filterwarnings("ignore")
from urllib import parse
from requests_toolbelt.multipart.encoder import MultipartEncoder
# 关闭警告
urllib3.disable_warnings()
warnings.filterwarnings("ignore")
from urllib import parse
from requests_toolbelt.multipart.encoder import MultipartEncoder
SAVE_ROBOT_TEMPLATE = {
"packageMd5": "2f260ae05e7aca32ba77f7e2b2b0785c",
"robotPackage": {
"activities": [],
"appIcon": "",
"authority": "use",
"customItems": {
"gifUrl": None,
"imageUrl": "",
"uiaType": "pc",
"videoUrl": "",
},
"description": "",
"externalDependencies": [],
"instruction": "",
"internalDependencies": [],
"isBrief": False,
"name": "测试",
"packageName": "",
"robotType": "app",
"software": False,
"softwareTitle": False,
"statistics": {"blockCount": 1, "flowCount": 1, "sourceLineCount": 0},
"uiTags": "",
},
"userCustomTypeUuid": None,
"uuid": "4cdf6f8b-185d-463f-bd37-ad207e25be59",
}
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_activity_info(token, page=1, size=200):
"""
获取活动信息列表
:param token: 认证用的 token
:param page: 分页页码,默认为 1
:param size: 分页大小,默认为 10
:return: 所有页的数据列表
"""
url = "https://api.winrobot360.com/api/v1/robot/activity/package/list"
all_data = []
while True:
payload = {
"source": "market",
"page": page,
"size": size,
"groupInfoUuid": "22222222-2222-2222-2222-222222222222",
"marketUiTag": "",
"uiaType": "pc",
"order": "desc",
}
headers = {
"Authorization": f"bearer {token}",
"Accept": "application/json",
"Accept-Encoding": "json",
}
response = requests.get(url, headers=headers, params=payload)
response_json = response.json()
if not response.ok or not response_json.get("success"):
raise ValueError(response_json.get("msg"))
data = response_json.get("data")
all_data.extend(data)
if not data or len(data) < size:
break
page += 1
return all_data
def get_activity_detail(token, package_name, package_version):
"""
获取自定义指令的相关信息
:param token: token
:param sdk: 自定义指令
:return: package_name, package_url
"""
url = f"https://api.winrobot360.com/api/v1/robot/use/activity/package/detail?packageName={package_name}&version={package_version}"
payload = {}
headers = {
"Authorization": f"bearer {token}",
"Accept": "application/json",
"Accept-Encoding": "json",
}
response = requests.request("GET", url, headers=headers, data=payload)
return response.json().get("data")
def get_office_app_uuids() -> list:
"""
获取官方应用的 uuid,
:return: list, 官方应用 uuid
"""
uuids = []
tmp_list = []
page = 1
url = "https://api.winrobot360.com/api/noauth/v1/group/official/queryOfficialRobotList?page=%s&size=200"
page_response = requests.request("GET", url % page)
page_info = page_response.json().get("page").get("pages")
for item in range(1, page_info + 1):
response = requests.request("GET", url % item)
for item in response.json().get("data"):
# if item.get("robotGradeName") == "简单应用":
# continue
if item.get("uiTags") == "极速":
continue
print(item.get("name"), item.get("robotGradeName"))
uuids.append(item.get("uuid"))
name = item.get("name")
uuid = item.get("uuid")
return uuids
class Client:
"""初始化一个客户端"""
def __init__(self, username, password, host="https://api.winrobot360.com") -> None:
self.username = username
self.password = password
self.host = host
self.token = self.get_token()
self.headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json",
"Accept-Encoding": "json",
}
self.path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "download"
)
def get_token(self) -> str:
"""利用网页端获取 token"""
api = "/oauth/token"
url = self.host + api
data = {
"username": self.username,
"password": self.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, verify=False
).json()
if response.get("success") == "false":
return response.get("msg")
token = response.get("access_token")
if not token:
raise Exception("获取 token 失败, 请检查专有云账号和密码,")
return token
def download(self, url, save_path, filename):
"""根据链接下载文件"""
url = self.host + url
res = requests.get(url, verify=False)
with open(os.path.join(save_path, filename), "wb") as f:
f.write(res.content)
return os.path.join(save_path, filename)
def get_robot_detail(self, roobt_uuid):
"""获取机器人详细信息"""
route = f"/api/v1/robot/develop/detailByRobotUuid?robotUuid={roobt_uuid}"
url = self.host + route
payload = {}
response = requests.request(
"POST", url, headers=self.headers, data=payload, verify=False
)
return response.json().get("data")
def get_robot_uuids(self):
route = "/api/v1/robot/develop/robotList?userCustomTypeUuid=&page=1&size=30&robotType=app"
url = self.host + route
pages = (
requests.get(url, headers=self.headers, verify=False)
.json()
.get("page")
.get("pages")
)
robot_uuids = []
for page in range(1, pages + 1):
page_url = (
self.host
+ f"/api/v1/robot/develop/robotList?userCustomTypeUuid=&page={page}&size=30&robotType=app"
)
page_info = (
requests.get(page_url, headers=self.headers, verify=False)
.json()
.get("data")
)
robot_uuids.extend([item.get("uuid") for item in page_info])
return robot_uuids
def check_public_robot(self, robot_uuid):
route = f"/api/v1/robot/develop/detailByRobotUuid?robotUuid={robot_uuid}&checkBenefit=False "
url = self.host + route
result = requests.post(url, headers=self.headers, verify=False).json()
return result.get("msg") == "未找到应用"
def assign_upload_url(
self, robot_uuid, file_origin_scene="robot_package_schema", robot_type="app"
):
route = f"/api/v1/robot/develop/assignUploadUrl?fileOriginScene={file_origin_scene}&robotType={robot_type}&robotUuid={robot_uuid}"
url = self.host + route
result = requests.post(url, headers=self.headers, verify=False)
return result.json().get("data")
def save_public(self, robot_detail):
route = "/api/v1/robot/develop/save"
url = self.host + route
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json",
"Content-Type": "application/json; charset=utf-8",
}
response = requests.request(
"POST", url, headers=headers, data=json.dumps(robot_detail), verify=False
)
return response.json()
def upload_plublic(self, upload_detail, file_path):
filesize = os.path.getsize(file_path)
headers = {"Content-Length": str(filesize)}
result = requests.put(
upload_detail.get("uploadUrl"),
data=open(file_path, "rb"),
verify=False,
headers=headers,
)
def _unzip(self, file_path):
save_path, _ = os.path.splitext(file_path)
with zipfile.ZipFile(file_path) as zf:
zf.extractall(save_path)
return save_path
def _zip(self, path):
save_path = path + ".botx"
z = zipfile.ZipFile(
save_path,
"w",
compression=zipfile.ZIP_STORED,
allowZip64=True,
)
for dirpath, dirnames, filenames in os.walk(path):
fpath = dirpath.replace(path, "")
fpath = fpath and fpath + os.sep or ""
for filename in filenames:
z.write(os.path.join(dirpath, filename), fpath + filename)
z.close()
return save_path
def md5(self, file_path):
with open(file_path, "rb") as f:
bytes = f.read() # read file as bytes
readable_hash = hashlib.md5(bytes).hexdigest()
return readable_hash
def check_create_benefit(self):
route = "/api/v1/robot/develop/checkCreateBenefit"
url = self.host + route
result = requests.get(url, headers=self.headers, verify=False)
return result.json().get("data")
def generate_packagename(self):
route = "/api/v1/robot/activity/generate/packageName"
url = self.host + route
result = requests.get(url, headers=self.headers, verify=False)
return result.json().get("data")
def get_activity_detail(self, activity_name):
"""
获取自定义指令的相关信息
:param token: token
:param sdk: 自定义指令
:return: package_name, package_url
"""
name, version = activity_name.split("==")
router = f"/api/v1/robot/use/activity/package/detail?packageName={name}&version={version}"
url = self.host + router
response = requests.request(
"GET", url, headers=self.headers, data={}, verify=False
)
data = response.json().get("data")
if not data:
raise Exception("获取指令详情异常")
return data
def publish(self, robot_uuid):
"""发版指令"""
url = f"https://api.winrobot360.com/api/v1/robot/develop/publish?robotUuid={robot_uuid}&versionLog="
payload = {}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
}
response = requests.request("POST", url, headers=headers, data=payload)
return response.json()
def share_enterprise_market(self, robot_uuid, enterprise_uuid):
"""分享至企业市场"""
url = f"https://api.winrobot360.com/api/v1/enterpriseRobot/share/activity/market?robotUuid={robot_uuid}&enterpriseUuid={enterprise_uuid}"
payload = {}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
}
response = requests.request("POST", url, headers=headers, data=payload)
return response.json()
def get_user_info(self):
"""获取用户的基础信息"""
url = "https://api.winrobot360.com/api/v1/user/info/rich"
payload = {}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
}
response = requests.request("GET", url, headers=headers, data=payload)
# data enterprise ownerUuid=
return response.json()
def get_activity_package_name_by_type(self, query_type="market"):
"""获取市场, 企业市场, 个人指令的编码
:param query_type: market, enterprise, personal
"""
url = f"https://api.winrobot360.com/api/v1/robot/activity/package/list?key=&source={query_type}&page=1&size=200&groupInfoUuid=22222222-2222-2222-2222-222222222222&marketUiTag=&uiaType=pc"
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json",
}
response = requests.request("GET", url, headers=headers, verify=False)
activities = response.json().get("data")
return {
activity.get("packageName"): activity.get("packageVersion")
for activity in activities
}
def func(self, robot_uuid):
url = f"https://api.winrobot360.com/api/v1/robot/develop/detailByRobotUuid?robotUuid={robot_uuid}&checkBenefit=False"
payload = {}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
}
response = requests.request(
"POST", url, headers=headers, data=payload, verify=False
)
def get_robot_list(self):
router = "/api/v1/robot/develop/robotList?page=1&size=200&robotType=app"
url = self.host + router
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json",
}
response = requests.request("GET", url, headers=headers, verify=False)
return [item.get("robotUuid") for item in response.json().get("data")]
def logout(self):
url = "https://api.winrobot360.com/api/v1/user/logout"
payload = {}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json, text/json, text/x-json, text/javascript, application/xml, text/xml",
}
response = requests.request("POST", url, headers=headers, data=payload)
res = response.json()
def add_developer(self, app_id, user_id):
url = (
"https://api.winrobot360.com/api/client/user/relation/addDeveloperRelation"
)
payload = {"appId": app_id, "userId": user_id}
headers = {
"Authorization": f"bearer {self.token}",
"Accept": "application/json",
}
response = requests.request("POST", url, headers=headers, json=payload)
print(response.json())
def new_blank_robot(name="", user_id="80808032853078016"):
"""新建一个应用"""
public_username = os.environ["XBOT_USERNAME"]
public_password = os.environ["XBOT_PASSWORD"]
public = Client(public_username, public_password)
# 1. 设置文件上传模板文件路径
schema_save_path = os.path.join("/code/app/flow_template/robot", "package.json")
package_save_path = os.path.join("/code/app/flow_template/robot", "blank.bot")
# 2. 创建新公有云新 UUID
new_uuid = str(uuid.uuid4())
print(new_uuid)
# 3. 修改应用信息
# 3.1 修改 package.json 文件
with open(schema_save_path, "r", encoding="utf-8") as f:
package_json_object = json.load(f)
package_json_object["uuid"] = new_uuid
package_json_object["name"] = name
package_json_object["internaldependencies"] = []
with open(schema_save_path, "w", encoding="utf8") as f:
json.dump(package_json_object, f, ensure_ascii=False, indent=2)
# 3.1 修改 package.bot 文件
with zipfile.ZipFile(package_save_path, mode="r") as zip_in, zipfile.ZipFile(
package_save_path + "x", mode="w"
) as zip_out, open(schema_save_path, "r", encoding="utf8") as f:
for item in zip_in.infolist():
if item.filename == "package.json":
continue
data = zip_in.read(item.filename)
zip_out.writestr(item, data)
updated_json_content = json.dumps(json.load(f), ensure_ascii=False, indent=2)
zip_info = zipfile.ZipInfo("package.json")
zip_info.compress_type = zipfile.ZIP_STORED
zip_out.writestr(zip_info, updated_json_content)
os.remove(package_save_path)
os.rename(package_save_path + "x", package_save_path)
new_md5 = public.md5(package_save_path)
# 4. 分配上传链接
robot_package_bot_json = public.assign_upload_url(new_uuid, "robot_package_bot")
robot_package_schema_json = public.assign_upload_url(
new_uuid, "robot_package_schema"
)
# 6. 保存应用
public.upload_plublic(robot_package_bot_json, package_save_path)
public.upload_plublic(robot_package_schema_json, schema_save_path)
SAVE_ROBOT_TEMPLATE["packageMd5"] = new_md5
SAVE_ROBOT_TEMPLATE["uuid"] = new_uuid
SAVE_ROBOT_TEMPLATE["robotPackage"]["name"] = name
SAVE_ROBOT_TEMPLATE["robotPackage"]["internalDependencies"] = []
SAVE_ROBOT_TEMPLATE["robotPackage"]["externalDependencies"] = []
res = public.save_public(SAVE_ROBOT_TEMPLATE)
print(res)
time.sleep(10)
print(f"set developer: {user_id}")
public.add_developer(new_uuid, user_id)
return new_uuid
def new_blank_activity(name="", user_id=""):
"""新建一个指令"""
public_username = os.environ["XBOT_USERNAME"]
public_password = os.environ["XBOT_PASSWORD"]
public = Client(public_username, public_password)
# 1. 设置文件上传模板文件路径
schema_save_path = os.path.join("/code/app/flow_template/activity", "package.json")
package_save_path = os.path.join("/code/app/flow_template/activity", "activity.bot")
# 3. 公有云生成指令编码和 UUID
activity_package_code = public.generate_packagename()
new_robot_uuid = str(uuid.uuid4())
# 4. 更新 package.json 和 package.bot
with open(schema_save_path, "r", encoding="utf-8") as f:
package_json_object = json.load(f)
package_json_object["uuid"] = new_robot_uuid
package_json_object["activity_code"] = activity_package_code
package_json_object["name"] = name
with open(schema_save_path, "w") as f:
json.dump(package_json_object, f, ensure_ascii=False, indent=2)
with zipfile.ZipFile(package_save_path, mode="r") as zip_in, zipfile.ZipFile(
package_save_path + "x", mode="w"
) as zip_out, open(schema_save_path, "r") as f:
for item in zip_in.infolist():
if item.filename == "package.json":
continue
data = zip_in.read(item.filename)
zip_out.writestr(item, data)
updated_json_content = json.dumps(json.load(f), ensure_ascii=False, indent=2)
zip_info = zipfile.ZipInfo("package.json")
zip_info.compress_type = zipfile.ZIP_STORED
zip_out.writestr(zip_info, updated_json_content)
os.remove(package_save_path)
os.rename(package_save_path + "x", package_save_path)
new_md5 = public.md5(package_save_path)
# 4. 公有云生成上传链接
robot_package_bot_json = public.assign_upload_url(
new_robot_uuid, "robot_package_bot", "activity"
)
robot_package_schema_json = public.assign_upload_url(
new_robot_uuid, "robot_package_schema", "activity"
)
# 5. 公有云保存指令
public.upload_plublic(robot_package_bot_json, package_save_path)
public.upload_plublic(robot_package_schema_json, schema_save_path)
# 6. 公有云保存指令
SAVE_ROBOT_TEMPLATE["packageMd5"] = new_md5
SAVE_ROBOT_TEMPLATE["uuid"] = new_robot_uuid
SAVE_ROBOT_TEMPLATE["robotPackage"]["name"] = name
SAVE_ROBOT_TEMPLATE["robotPackage"]["packageName"] = activity_package_code
SAVE_ROBOT_TEMPLATE["robotPackage"]["robotType"] = "activity"
SAVE_ROBOT_TEMPLATE["robotPackage"]["uiTags"] = None
SAVE_ROBOT_TEMPLATE["robotPackage"]["software"] = None
SAVE_ROBOT_TEMPLATE["robotPackage"]["softwareTitle"] = None
SAVE_ROBOT_TEMPLATE["robotPackage"]["activities"].append(
{"description": None, "helpLink": None, "name": "test"},
)
SAVE_ROBOT_TEMPLATE["robotPackage"]["internalDependencies"] = []
SAVE_ROBOT_TEMPLATE["robotPackage"]["externalDependencies"] = []
public.save_public(SAVE_ROBOT_TEMPLATE)
time.sleep(10)
public.add_developer(new_robot_uuid, user_id)
return new_robot_uuid
if __name__ == "__main__":
dotenv.load_dotenv()
xbot_username = os.environ["XBOT_USERNAME"]
xbot_password = os.environ["XBOT_PASSWORD"]
token = get_token(xbot_username, xbot_password)
activities_info = get_activity_info(token)
messages = []
messages.append("==========市场指令缺失文档==========")
for activity_info in activities_info:
package_viewN_name = activity_info.get("packageViewName")
activities = activity_info.get("activities", "空")
for activity in activities:
name = activity.get("name")
help_link = activity.get("helpLink", "")
if not help_link:
# print(f"{package_viewN_name}: {name} 缺失文档")
messages.append(f"{package_viewN_name}: {name}")
print("\n".join(messages))
# for item in office_activity_uuids:
# print(item.get("packageViewName"), item.get("uuid"))
# new_blank_robot("测试-根据需求生成应用", "620156920605310976")
# new_blank_activity()