102 lines
4.3 KiB
Python
102 lines
4.3 KiB
Python
import json
|
||
import os
|
||
|
||
def fix_swagger_responses(swagger_data):
|
||
"""
|
||
Modifies Swagger 2.0 responses to include 'application/json' content type
|
||
in an OpenAPI 3.0.x style, by wrapping the existing schema.
|
||
It also removes the '$schema' key from the response schemas.
|
||
|
||
Args:
|
||
swagger_data (dict): The Swagger data as a Python dictionary.
|
||
|
||
Returns:
|
||
dict: The modified Swagger data.
|
||
"""
|
||
if not isinstance(swagger_data, dict):
|
||
return swagger_data
|
||
|
||
paths = swagger_data.get("paths", {})
|
||
for path_item_key, path_item_value in paths.items(): # Iterate with keys for better debugging if needed
|
||
if not isinstance(path_item_value, dict):
|
||
continue
|
||
for operation_method, operation_details in path_item_value.items():
|
||
# Valid HTTP methods according to OpenAPI spec.
|
||
# We are interested in actual operations like get, post, put, delete, etc.
|
||
if operation_method.lower() not in ["get", "post", "put", "delete", "patch", "options", "head", "trace"]:
|
||
continue
|
||
|
||
if not isinstance(operation_details, dict):
|
||
continue
|
||
|
||
responses = operation_details.get("responses")
|
||
if not isinstance(responses, dict):
|
||
continue
|
||
|
||
for status_code, response_definition in responses.items():
|
||
if not isinstance(response_definition, dict):
|
||
continue
|
||
|
||
if "schema" in response_definition:
|
||
# Preserve the original schema and remove it from its current location
|
||
original_schema = response_definition.pop("schema")
|
||
|
||
# Remove "$schema" key from the original schema if it exists and schema is a dict
|
||
if isinstance(original_schema, dict) and "$schema" in original_schema:
|
||
del original_schema["$schema"]
|
||
|
||
# Create the new content structure
|
||
response_definition["content"] = {
|
||
"application/json": {
|
||
"schema": original_schema
|
||
}
|
||
}
|
||
return swagger_data
|
||
|
||
def process_swagger_file(file_path):
|
||
"""
|
||
Reads a Swagger JSON file, applies fixes, and prints the modified JSON.
|
||
|
||
Args:
|
||
file_path (str): The path to the Swagger JSON file.
|
||
"""
|
||
try:
|
||
# Expand user's home directory if ~ is used
|
||
expanded_path = os.path.expanduser(file_path)
|
||
|
||
with open(expanded_path, 'r', encoding='utf-8') as f:
|
||
swagger_data = json.load(f)
|
||
|
||
print(f"Successfully read file: {expanded_path}")
|
||
|
||
modified_swagger_data = fix_swagger_responses(swagger_data)
|
||
|
||
# Print the modified JSON data
|
||
# indent=2 makes the output pretty-printed and readable
|
||
# ensure_ascii=False allows non-ASCII characters (like Chinese) to be printed as is
|
||
print("\nModified Swagger JSON:")
|
||
print(json.dumps(modified_swagger_data, indent=2, ensure_ascii=False))
|
||
|
||
# If you want to save it back to a new file or overwrite:
|
||
output_file_path = os.path.join(os.path.dirname(expanded_path), "井筒API示例swagger_fixed.json")
|
||
with open(output_file_path, 'w', encoding='utf-8') as f:
|
||
json.dump(modified_swagger_data, f, indent=2, ensure_ascii=False)
|
||
print(f"\nModified Swagger JSON saved to: {output_file_path}")
|
||
|
||
except FileNotFoundError:
|
||
print(f"Error: File not found at {file_path} (Expanded: {expanded_path})")
|
||
except json.JSONDecodeError:
|
||
print(f"Error: Could not decode JSON from file {file_path}. Make sure it's a valid JSON.")
|
||
except Exception as e:
|
||
print(f"An unexpected error occurred: {e}")
|
||
|
||
# --- Main execution ---
|
||
if __name__ == "__main__":
|
||
# 使用您提供的文件路径
|
||
# 注意:在Python字符串中,反斜杠 '\' 是转义字符。
|
||
# 在Windows路径中,您可能需要使用双反斜杠 '\\' 或者原始字符串 r"..."
|
||
# 在macOS或Linux上,普通正斜杠 '/' 就可以。
|
||
# 您提供的路径是macOS/Linux风格,所以可以直接使用。
|
||
swagger_file_path = "/Users/zpc01/workspace/zzlh/compliance/assets/doc/井筒API示例swagger.json"
|
||
|
||
process_swagger_file(swagger_file_path) |