Skip to content

Commit 8d24b9a

Browse files
kaburagiseckaburagisec
committed
fix(zeep): Enhance zeep xsd:any parser to support attribute parsing
1 parent 60e3496 commit 8d24b9a

1 file changed

Lines changed: 72 additions & 22 deletions

File tree

‎onvif/utils/zeep.py‎

Lines changed: 72 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -72,25 +72,47 @@ def flatten_xsd_any_fields(obj, _visited=None):
7272
value_data = values[value_key]
7373

7474
# Only process if _value_N is a dict (from our patched parser)
75-
if isinstance(value_data, dict):
75+
if isinstance(value_data, dict) and "__original_elements__" in value_data:
7676
# Extract original elements first
7777
original_elements = value_data.get("__original_elements__")
7878

79-
# Copy all non-private fields from _value_N dict to parent
80-
for key, val in list(value_data.items()):
81-
# Skip __original_elements__ and other private keys
82-
if key.startswith("_"):
83-
continue
84-
85-
# Only copy if field exists in parent and is None
86-
if key in values and values[key] is None:
87-
values[key] = val
79+
# Check if this is a single-tag wrapper (like {"Capabilities": {...}})
80+
# If so, flatten one level by extracting the inner content
81+
non_private_keys = [k for k in value_data.keys() if not k.startswith("_")]
82+
83+
if len(non_private_keys) == 1:
84+
# Single tag wrapper - flatten it
85+
tag_name = non_private_keys[0]
86+
inner_content = value_data[tag_name]
87+
88+
if isinstance(inner_content, dict):
89+
# Copy fields from inner dict to parent
90+
for key, val in inner_content.items():
91+
if key in values and values[key] is None:
92+
values[key] = val
93+
elif key not in values:
94+
values[key] = val
95+
else:
96+
# If inner content is not a dict, just copy it directly
97+
if tag_name in values and values[tag_name] is None:
98+
values[tag_name] = inner_content
99+
elif tag_name not in values:
100+
values[tag_name] = inner_content
101+
else:
102+
# Multiple tags or no tags - copy all non-private fields
103+
for key, val in list(value_data.items()):
104+
if key.startswith("_"):
105+
continue
106+
107+
if key in values and values[key] is None:
108+
values[key] = val
109+
elif key not in values:
110+
values[key] = val
88111

89112
# Replace _value_N with ONLY the original elements list
90113
if original_elements is not None:
91114
values[value_key] = original_elements
92115
else:
93-
# If no original elements, set to None
94116
values[value_key] = None
95117

96118
value_n += 1
@@ -102,19 +124,39 @@ def flatten_xsd_any_fields(obj, _visited=None):
102124
value_key = f"_value_{value_n}"
103125
value_data = getattr(obj, value_key)
104126

105-
if isinstance(value_data, dict):
127+
if isinstance(value_data, dict) and "__original_elements__" in value_data:
106128
# Extract original elements first (before any modification)
107129
original_elements = value_data.get("__original_elements__")
108130

109-
# Copy all non-private fields from _value_N to parent
110-
for key, val in value_data.items():
111-
# Skip __original_elements__ and other private keys
112-
if key.startswith("_"):
113-
continue
114-
115-
# Only copy if field exists in parent and is None
116-
if hasattr(obj, key) and getattr(obj, key) is None:
117-
setattr(obj, key, val)
131+
# Check if this is a single-tag wrapper
132+
non_private_keys = [k for k in value_data.keys() if not k.startswith("_")]
133+
134+
if len(non_private_keys) == 1:
135+
# Single tag wrapper - flatten it
136+
tag_name = non_private_keys[0]
137+
inner_content = value_data[tag_name]
138+
139+
if isinstance(inner_content, dict):
140+
for key, val in inner_content.items():
141+
if hasattr(obj, key) and getattr(obj, key) is None:
142+
setattr(obj, key, val)
143+
elif not hasattr(obj, key):
144+
setattr(obj, key, val)
145+
else:
146+
if hasattr(obj, tag_name) and getattr(obj, tag_name) is None:
147+
setattr(obj, tag_name, inner_content)
148+
elif not hasattr(obj, tag_name):
149+
setattr(obj, tag_name, inner_content)
150+
else:
151+
# Multiple tags - copy all non-private fields
152+
for key, val in value_data.items():
153+
if key.startswith("_"):
154+
continue
155+
156+
if hasattr(obj, key) and getattr(obj, key) is None:
157+
setattr(obj, key, val)
158+
elif not hasattr(obj, key):
159+
setattr(obj, key, val)
118160

119161
# Replace _value_N with ONLY the original elements list
120162
if original_elements is not None:
@@ -172,13 +214,21 @@ def patched_parse_xmlelements(self, xmlelements, schema, name=None, context=None
172214
)
173215
child_result[child_qname.localname] = val
174216
except Exception:
175-
if list(child):
217+
# If schema lookup fails, try to parse manually
218+
# Check if child has attributes (common in ONVIF capabilities)
219+
if child.attrib:
220+
# Parse attributes as a dict
221+
attr_dict = {k: parse_text_value(v) for k, v in child.attrib.items()}
222+
child_result[child_qname.localname] = attr_dict
223+
elif list(child):
224+
# Has nested children
176225
nested = {
177226
QName(sub.tag).localname: parse_text_value(sub.text)
178227
for sub in child
179228
}
180229
child_result[child_qname.localname] = nested
181230
else:
231+
# Only has text content
182232
child_result[child_qname.localname] = parse_text_value(
183233
child.text
184234
)

0 commit comments

Comments
 (0)