fix: handle self-referencing types to avoid infinite recursion

tccli/cli_unfold_argument.py

@@ -16,14 +16,26 @@ class CliUnfoldArgument(CustomArgument):
     def __init__(self):
         super(CliUnfoldArgument, self).__init__(**self.ARG_DATA)
 
-    def build_action_parameters(self, args):
+    def build_action_parameters(self, args, extra_unfold_args=None):
+        """从 argparse Namespace 构造请求体。
+
+        :param args: argparse Namespace，扁平展开参数。
+        :param extra_unfold_args: 可选的额外扁平 ``{key: value}``，与 Namespace 合并后
+            一同走 ``convert_to_dict`` + ``handle_array``；为 ``None`` 或空时等同改动前。
+        :return: 嵌套 dict / list 形式的请求体。
+        """
         parsed_args = vars(args)
         for key in list(parsed_args.keys()):
             if parsed_args[key] is None:
                 del parsed_args[key]
         params_set = {}
         for key, value in parsed_args.items():
             self.convert_to_dict(params_set, key, value)
+        if extra_unfold_args:
+            for key, value in extra_unfold_args.items():
+                if value is None:
+                    continue
+                self.convert_to_dict(params_set, key, value)
         return self.handle_array(params_set, "--")
 
     def convert_to_dict(self, params_set, key, value):

tccli/document_handler.py

@@ -175,6 +175,14 @@ def _json_format(self, param_info):
                     self.doc.write('[%s ...]' % (param_info["members"]))
                 else:
                     self.doc.doc_description('[%s ...]' % (param_info["members"]))
+            elif isinstance(param_info["members"], str):
+                # 自引用截断占位：members 为类型名字符串（如 "AllocationRuleExpression"），
+                # 直接渲染 recursive 提示占位，不再向下展开
+                placeholder = '[<recursive: fill with a JSON object of type %s, same shape as parent> ...]' % param_info["members"]
+                if self.doc.style.indentation > 2:
+                    self.doc.write(placeholder)
+                else:
+                    self.doc.doc_description(placeholder)
             else:
                 self.doc.write('[') if self.doc.style.indentation > 2 \
                     else self.doc.doc_description('[')
@@ -187,8 +195,15 @@ def _json_format(self, param_info):
                 self.doc.style.new_line()
                 self.doc.doc_description(']')
         else:
-            if param_info["members"] not in BASE_TYPE:
-                self._handle_object_members(param_info["members"], param_info["type"])
+            if param_info["members"] in BASE_TYPE:
+                return
+            if isinstance(param_info["members"], str):
+                # 自引用截断占位（非 Array 形态）：members 为类型名字符串，
+                # 直接渲染 recursive 提示占位
+                self.doc.doc_description(
+                    '<recursive: fill with a JSON object of type %s, same shape as parent>' % param_info["members"])
+                return
+            self._handle_object_members(param_info["members"], param_info["type"])
 
     def _handle_object_members(self, param_info, param_type):
         if param_type == "Array" or self.doc.style.indentation == 2:
@@ -216,6 +231,9 @@ def _handle_object_members(self, param_info, param_type):
     def _unfold_complex_object(self, param_info):
         if not param_info["type"] == "Array" and param_info["members"] in BASE_TYPE:
             return
+        # 自引用截断占位：非 Array 且 members 为类型名字符串，无可展开内容，直接跳过
+        if not param_info["type"] == "Array" and isinstance(param_info["members"], str):
+            return
 
         self.doc.style.new_line()
         self.doc.doc_title('JSON Syntax')
@@ -232,13 +250,18 @@ def _param_type(self, param):
     def _complex_object_doc(self, param_info, option):
         if param_info["members"] in BASE_TYPE:
             return
+        # 自引用截断占位：members 为类型名字符串，无子成员可遍历，直接返回
+        if not isinstance(param_info["members"], dict):
+            return
         self.doc.style.indent()
         for sub_param, sub_param_info in param_info["members"].items():
             self.doc.style.new_line()
             self._doc_title(option, sub_param, sub_param_info)
             self.doc.doc_description('%s' % sub_param_info["document"])
             self.doc.style.new_line()
-            if sub_param_info["members"] not in BASE_TYPE:
+            # 仅当子成员的 members 仍然是 dict（即未被截断且非基础类型）时才继续递归
+            if sub_param_info["members"] not in BASE_TYPE \
+                    and isinstance(sub_param_info["members"], dict):
                 self._complex_object_doc(sub_param_info, option)
         self.doc.style.dedent()
         self.doc.style.new_line()

tccli/loaders.py

@@ -14,6 +14,17 @@
 BASE_TYPE = ["int64", "uint64", "string", "float", "bool", "date", "datetime", "datetime_iso", "binary"]
 CLI_BASE_TYPE = ["Integer", "String", "Float", "Timestamp", "Boolean", "Binary"]
 
+# --cli-unfold-argument 模式下扁平 key 的最大点号段数（含数字下标段），超过则报错。
+MAX_INPUT_DEPTH = 30
+
+# 自引用截断点 / 超限输入的统一替代写法提示文案。
+RECURSIVE_HINT_FILE_OPTION = (
+    "Use --cli-input-json file://<path/to/request.json> to provide the entire "
+    "request as a JSON file (the value must begin with 'file://'; raw JSON "
+    "strings are not accepted; run with --generate-cli-skeleton to get a JSON "
+    "template)."
+)
+
 PARAM_TYPE_MAP = {
     'int64': 'Integer',
     'uint64': 'Integer',
@@ -333,24 +344,41 @@ def _filling_param_info(self, param_info, para, param_type, member):
         param_info[para["name"]]["members"] = member
         return param_info
 
-    def _get_param_info(self, param_model, object_model):
+    def _get_param_info(self, param_model, object_model, visited=None):
+        # visited 沿当前 DFS 路径记录已展开的复合类型名，命中即截断
+        if visited is None:
+            visited = frozenset()
         param_info = {}
         for para in param_model:
+            member = para["member"]
+            recursive_hit = member not in BASE_TYPE and member in visited
             if para["type"] == "list":
-                if para["member"] not in BASE_TYPE:
-                    self._filling_param_info(
-                        param_info, para, "list",
-                        [self._get_param_info(object_model[para["member"]]["members"], object_model)])
+                if member not in BASE_TYPE:
+                    if recursive_hit:
+                        self._filling_param_info(
+                            param_info, para, "list", [member])
+                    else:
+                        self._filling_param_info(
+                            param_info, para, "list",
+                            [self._get_param_info(
+                                object_model[member]["members"], object_model,
+                                visited | {member})])
                 else:
                     self._filling_param_info(
-                        param_info, para, "list", [para["member"]])
+                        param_info, para, "list", [member])
             else:
-                if para["member"] not in BASE_TYPE:
-                    param_info = self._filling_param_info(
-                        param_info, para, para["member"],
-                        self._get_param_info(object_model[para["member"]]["members"], object_model))
+                if member not in BASE_TYPE:
+                    if recursive_hit:
+                        param_info = self._filling_param_info(
+                            param_info, para, member, member)
+                    else:
+                        param_info = self._filling_param_info(
+                            param_info, para, member,
+                            self._get_param_info(
+                                object_model[member]["members"], object_model,
+                                visited | {member}))
                 else:
-                    self._filling_param_info(param_info, para, para["member"], para["member"])
+                    self._filling_param_info(param_info, para, member, member)
         return param_info
 
     def get_param_info(self, service, version, action):
@@ -363,21 +391,40 @@ def get_output_param_info(self, service, version, action):
         param_model = service_model["objects"]
         return self._get_param_info(param_model[action + "Response"]["members"], param_model)
 
-    def _generate_param_skeleton(self, param_model, name):
+    def _generate_param_skeleton(self, param_model, name, visited=None):
+        # visited 沿路径记录已展开的复合类型名，命中即以字符串占位表示自引用
+        if visited is None:
+            visited = frozenset()
         param_skeleton = {}
         for para in param_model:
+            member = para["member"]
+            recursive_hit = member not in BASE_TYPE and member in visited
             if para["type"] == "list":
-                if para["member"] not in BASE_TYPE:
-                    param_skeleton[para["name"]] = \
-                        [self._generate_param_skeleton(name[para["member"]]["members"], name)]
+                if member not in BASE_TYPE:
+                    if recursive_hit:
+                        param_skeleton[para["name"]] = [
+                            "<recursive: fill '%s' with a JSON object of type %s (self-referenced)>"
+                            % (para["name"], member)]
+                    else:
+                        param_skeleton[para["name"]] = \
+                            [self._generate_param_skeleton(
+                                name[member]["members"], name,
+                                visited | {member})]
                 else:
-                    param_skeleton[para["name"]] = [PARAM_TYPE_MAP[para["member"]]]
+                    param_skeleton[para["name"]] = [PARAM_TYPE_MAP[member]]
             else:
-                if para["member"] not in BASE_TYPE:
-                    param_skeleton[para["name"]] = \
-                        self._generate_param_skeleton(name[para["member"]]["members"], name)
+                if member not in BASE_TYPE:
+                    if recursive_hit:
+                        param_skeleton[para["name"]] = \
+                            "<recursive: fill '%s' with a JSON object of type %s (self-referenced)>" \
+                            % (para["name"], member)
+                    else:
+                        param_skeleton[para["name"]] = \
+                            self._generate_param_skeleton(
+                                name[member]["members"], name,
+                                visited | {member})
                 else:
-                    param_skeleton[para["name"]] = PARAM_TYPE_MAP[para["member"]]
+                    param_skeleton[para["name"]] = PARAM_TYPE_MAP[member]
         return param_skeleton
 
     def generate_param_skeleton(self, service, version, action):
@@ -396,7 +443,7 @@ def get_unfold_param_info(self, service, version, action, profile="default", par
         if param_array:
             all_param_list = self._add_array_item(all_param_list, profile)
 
-        return self._filling_unfold_param_info(all_param_list, service, version, action)
+        return self._filling_unfold_param_info(all_param_list, service, version, action, object_model)
 
     def _add_array_item(self, param_list, profile):
         is_conf_exist, conf_path = Utils.file_existed(os.path.join(os.path.expanduser("~"), ".tccli"),
@@ -415,27 +462,39 @@ def _add_array_item(self, param_list, profile):
                         all_param_list.append(tmp)
         return all_param_list
 
-    def _recur_get_unfold_param_info(self, param_model, object_model, return_param_list, param_list):
+    def _recur_get_unfold_param_info(self, param_model, object_model, return_param_list, param_list,
+                                     visited=None):
         for para in param_model:
-            self._get_unfold_param_info(object_model, return_param_list, param_list, para)
+            self._get_unfold_param_info(object_model, return_param_list, param_list, para, visited)
         if param_list.pop().isdigit():
             param_list.pop()
 
-    def _get_unfold_param_info(self, object_model, return_param_list, param_list, para):
+    def _get_unfold_param_info(self, object_model, return_param_list, param_list, para, visited=None):
+        # visited 沿路径维护，识别自引用类型（如 AllocationRuleExpression.Children）
+        if visited is None:
+            visited = frozenset()
         param_list.append(para["name"])
         if para["type"] == "list" and para["member"] not in BASE_TYPE:
             param_list.append('0')
-        if para["member"] not in BASE_TYPE:
-            self._recur_get_unfold_param_info(object_model[para["member"]]["members"],
-                                              object_model, return_param_list, param_list)
+        member = para["member"]
+        if member not in BASE_TYPE:
+            if member in visited:
+                tmp = copy.deepcopy(param_list)
+                return_param_list.append(tmp)
+                if param_list.pop().isdigit():
+                    param_list.pop()
+                return
+            self._recur_get_unfold_param_info(object_model[member]["members"],
+                                              object_model, return_param_list, param_list,
+                                              visited | {member})
         else:
             tmp = copy.deepcopy(param_list)
             return_param_list.append(tmp)
 
             if param_list.pop().isdigit():
                 param_list.pop()
 
-    def _filling_unfold_param_info(self, param_list, service, version, action):
+    def _filling_unfold_param_info(self, param_list, service, version, action, object_model=None):
         unfold_param = {}
         param_info = self.get_param_info(service, version, action)
         for param in param_list:
@@ -448,12 +507,22 @@ def _filling_unfold_param_info(self, param_list, service, version, action):
             type_name = res["type_name"]
             required = res.get("required")
             document = res["document"]
+            recursive_truncated = False
+            recursive_type = None
 
             for idx, item in enumerate(tmp_param[1:]):
+                # 命中自引用截断：当前 res 的 members 是占位字符串（类型名）而非 dict
                 if res["type"] == "Array":
-                    res = res["members"][0][item]
+                    members_container = res["members"][0]
                 else:
-                    res = res["members"][item]
+                    members_container = res["members"]
+                if not isinstance(members_container, dict) or item not in members_container:
+                    # 该 leaf 是被环检测截断的占位项，不再向下钻取
+                    recursive_truncated = True
+                    recursive_type = members_container if isinstance(members_container, str) \
+                        else (res.get("type_name") or "")
+                    break
+                res = members_container[item]
 
                 # ?? seriously ??
                 if required == "Required" and res["required"] == "Optional":
@@ -466,13 +535,44 @@ def _filling_unfold_param_info(self, param_list, service, version, action):
                     document = res["document"]
                     break
 
+            # 二次判定：路径走完后，若该 leaf 自身是被环检测截断的复合类型（members 为占位）
+            if not recursive_truncated:
+                final_members = res.get("members")
+                if isinstance(final_members, list) and len(final_members) == 1 \
+                        and isinstance(final_members[0], str) \
+                        and final_members[0] not in BASE_TYPE \
+                        and final_members[0] not in CLI_BASE_TYPE:
+                    recursive_truncated = True
+                    recursive_type = final_members[0]
+                elif isinstance(final_members, str) \
+                        and final_members not in BASE_TYPE \
+                        and final_members not in CLI_BASE_TYPE:
+                    recursive_truncated = True
+                    recursive_type = final_members
+
             if len([item for item in param if item.isdigit() and int(item) > 0]) > 0:
                 required = "Optional"
 
+            if recursive_truncated:
+                # 自引用截断点统一标记为 Object，提示用户用 JSON 整体传入
+                param_type = "Object"
+                type_name = recursive_type or "Object"
+                required = "Optional"
+                document = (document or "") + \
+                    ("\nNote: this field is a self-referencing type %s. "
+                     "--cli-unfold-argument only expands the first level. "
+                     "For deeper nesting:\n  %s"
+                     % (recursive_type or "", RECURSIVE_HINT_FILE_OPTION))
+
             unfold_param[".".join(param)]["type"] = param_type
             unfold_param[".".join(param)]["type_name"] = type_name
             unfold_param[".".join(param)]["required"] = required
             unfold_param[".".join(param)]["document"] = document
+            # 稳定字段：供上层（如 command.py）在客户深入自引用路径报 Unknown options 时
+            # 给出针对性提示，无需依赖 document 文案
+            if recursive_truncated:
+                unfold_param[".".join(param)]["recursive_truncated"] = True
+                unfold_param[".".join(param)]["recursive_type"] = recursive_type or ""
         return unfold_param
 
     def get_action_example_model(self, service, version, action):