Generates Python code for a given node in the graph based on its attributes.
(node_id: str, graph: nx.DiGraph)
| 136 | |
| 137 | |
| 138 | def generate_code(node_id: str, graph: nx.DiGraph) -> str: |
| 139 | """ |
| 140 | Generates Python code for a given node in the graph based on its attributes. |
| 141 | """ |
| 142 | |
| 143 | node_attrs = graph.nodes[node_id] |
| 144 | |
| 145 | if node_attrs.get("node_type", "") == "cookie": |
| 146 | cookie_value = node_attrs.get('content', {}).get('value', '') |
| 147 | cookie_key = node_attrs.get('content', {}).get('key', '') |
| 148 | return f"{cookie_value} = cookie_dict['{cookie_key}']" |
| 149 | |
| 150 | content = node_attrs.get("content", {}) |
| 151 | curl = content.get("key", "") |
| 152 | response = content.get("value", {}) |
| 153 | response_type = response.get("type", "") |
| 154 | response_text = response.get("text", "") |
| 155 | |
| 156 | dynamic_parts = node_attrs.get("dynamic_parts", "") |
| 157 | extracted_parts = node_attrs.get("extracted_parts", "") |
| 158 | input_variables = node_attrs.get("input_variables", "") |
| 159 | to_parse_response = True |
| 160 | |
| 161 | parse_response_prompt = "" |
| 162 | |
| 163 | if response_type in ["application/octet-stream", "application/pdf", "application/zip", "image/jpeg", "image/png"]: |
| 164 | parse_response_prompt = f""" |
| 165 | The response is a downloadable file of type {response_type}. |
| 166 | Include code to save the response content to a file with an appropriate extension. |
| 167 | """ |
| 168 | |
| 169 | if "application/json" in response_type: |
| 170 | key_paths = [] |
| 171 | for extracted_part in extracted_parts: |
| 172 | key_path = find_json_path(json.loads(response_text), extracted_part) |
| 173 | key_paths.append(key_path) |
| 174 | |
| 175 | parse_response_prompt = f""" |
| 176 | Response: |
| 177 | {response_text} |
| 178 | |
| 179 | Parse out the following variables from the response using JSON keys: |
| 180 | {key_paths} |
| 181 | |
| 182 | Through your judgement from analyzing the response, if polling is required to retrieve the variables above from the response. If so, implement polling else dont. |
| 183 | """ |
| 184 | |
| 185 | if "text/html" in response_type or "application/javascript" in response_type: |
| 186 | if len(response_text) > 100000: |
| 187 | context_snippets = [] |
| 188 | for part in extracted_parts: |
| 189 | index = response_text.find(part) |
| 190 | if index != -1: |
| 191 | start = max(0, index - 50) |
| 192 | end = min(len(response_text), index + len(part) + 50) |
| 193 | snippet = response_text[start:end] |
| 194 | context_snippets.append(f"{part}: {snippet}") |
| 195 |
no test coverage detected