hy/chatapp-deploy-platform
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy fix bug

Browse Source
This commit is contained in:
ZuoZuo 2026-04-07 17:01:26 +08:00
parent 4cd2d4e31c
commit e69c17592e
1 changed files with 12 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
ops-scripts/tencent_operator.py
View File

@ -45,6 +45,13 @@ def decode_output(encoded: str | None) -> str:
return "<failed to decode output>"
def sdk_body(resp: Any) -> dict[str, Any]:
body = json.loads(resp.to_json_string())
if not isinstance(body, dict):
raise RuntimeError(f"unexpected sdk response type: {type(body).__name__}")
return body
class CloudOperator:
def __init__(self, config: dict[str, Any], service_name: str, instance_id: str, release_id: str | None) -> None:
self.config = config
@ -197,7 +204,7 @@ class CloudOperator:
req = tat_models.RunCommandRequest()
req.from_json_string(json.dumps(payload))
resp = self.tat.RunCommand(req)
body = json.loads(resp.to_json_string())["Response"]
body = sdk_body(resp)
return str(body["InvocationId"])
def wait_for_tat(self, invocation_id: str) -> dict[str, Any]:
@ -214,7 +221,7 @@ class CloudOperator:
)
)
resp = self.tat.DescribeInvocationTasks(req)
body = json.loads(resp.to_json_string())["Response"]
body = sdk_body(resp)
tasks = body.get("InvocationTaskSet", [])
task = next((item for item in tasks if str(item["InstanceId"]) == self.instance_id), None)
if not task:
@ -240,7 +247,7 @@ class CloudOperator:
req = tat_models.DescribeAutomationAgentStatusRequest()
req.from_json_string(json.dumps({"InstanceIds": [self.instance_id], "Limit": 1}))
resp = self.tat.DescribeAutomationAgentStatus(req)
body = json.loads(resp.to_json_string())["Response"]
body = sdk_body(resp)
online = any(item.get("AgentStatus") == "Online" for item in body.get("AutomationAgentSet", []))
if not online:
raise RuntimeError(f"TAT agent is not online for {self.instance_id}")
@ -279,7 +286,7 @@ class CloudOperator:
req = clb_models.ModifyTargetWeightRequest()
req.from_json_string(json.dumps(payload))
resp = self.clb.ModifyTargetWeight(req)
request_id = json.loads(resp.to_json_string())["Response"]["RequestId"]
request_id = sdk_body(resp)["RequestId"]
self.wait_for_clb_task(request_id)
def wait_for_clb_task(self, request_id: str) -> None:
@ -288,7 +295,7 @@ class CloudOperator:
req = clb_models.DescribeTaskStatusRequest()
req.from_json_string(json.dumps({"TaskId": request_id}))
resp = self.clb.DescribeTaskStatus(req)
body = json.loads(resp.to_json_string())["Response"]
body = sdk_body(resp)
status = int(body["Status"])
if status == 0:
return