1269 lines
51 KiB
Python
Executable File
1269 lines
51 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Smart Schema Consolidation
|
|
==========================
|
|
|
|
Intelligent consolidation of duplicate OpenAPI schemas with smart naming.
|
|
|
|
Key improvements over basic consolidation:
|
|
1. Semantic analysis of schema names to find common entity
|
|
2. Priority-based canonical name selection
|
|
3. Automatic detection of Response/Request/List types
|
|
4. No manual rename_map needed
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import re
|
|
from collections import Counter, defaultdict
|
|
from typing import Dict, List, Tuple, Optional
|
|
|
|
|
|
class SchemaStructureAnalyzer:
|
|
"""Analyzes schema structure to find duplicates based on attributes"""
|
|
|
|
@classmethod
|
|
def get_structural_signature(cls, schema: dict, depth: int = 0, include_constraints: bool = True) -> str:
|
|
"""
|
|
Generate a structural signature for a schema, ignoring metadata.
|
|
This allows finding schemas with identical structure but different descriptions.
|
|
|
|
Args:
|
|
schema: The schema to analyze
|
|
depth: Current recursion depth
|
|
include_constraints: If True, include validation constraints (minLength, pattern, etc.)
|
|
If False, only compare type structure
|
|
"""
|
|
if not isinstance(schema, dict):
|
|
return str(schema)
|
|
|
|
sig_parts = []
|
|
|
|
# Type
|
|
if 'type' in schema:
|
|
sig_parts.append(f"type={schema['type']}")
|
|
|
|
# Format
|
|
if 'format' in schema:
|
|
sig_parts.append(f"format={schema['format']}")
|
|
|
|
# $ref
|
|
if '$ref' in schema:
|
|
sig_parts.append(f"ref={schema['$ref']}")
|
|
|
|
# Validation constraints (if enabled)
|
|
if include_constraints:
|
|
for constraint in ['minLength', 'maxLength', 'minimum', 'maximum',
|
|
'pattern', 'minItems', 'maxItems', 'uniqueItems']:
|
|
if constraint in schema:
|
|
sig_parts.append(f"{constraint}={schema[constraint]}")
|
|
|
|
# Properties (sorted by name, recursive)
|
|
if 'properties' in schema:
|
|
props = []
|
|
for prop_name, prop_def in sorted(schema['properties'].items()):
|
|
prop_sig = cls.get_structural_signature(prop_def, depth + 1, include_constraints)
|
|
nullable = prop_def.get('nullable', False)
|
|
props.append(f"{prop_name}:{prop_sig}:n={nullable}")
|
|
sig_parts.append(f"props=[{';'.join(props)}]")
|
|
|
|
# Required fields
|
|
if 'required' in schema:
|
|
sig_parts.append(f"req={sorted(schema['required'])}")
|
|
|
|
# Enum values
|
|
if 'enum' in schema:
|
|
sig_parts.append(f"enum={sorted(str(e) for e in schema['enum'])}")
|
|
|
|
# Array items
|
|
if 'items' in schema:
|
|
sig_parts.append(f"items={cls.get_structural_signature(schema['items'], depth + 1, include_constraints)}")
|
|
|
|
# AllOf/OneOf/AnyOf
|
|
for key in ['allOf', 'oneOf', 'anyOf']:
|
|
if key in schema:
|
|
sigs = sorted(cls.get_structural_signature(s, depth + 1, include_constraints) for s in schema[key])
|
|
sig_parts.append(f"{key}=[{';'.join(sigs)}]")
|
|
|
|
# AdditionalProperties
|
|
if 'additionalProperties' in schema:
|
|
ap = schema['additionalProperties']
|
|
if isinstance(ap, dict):
|
|
sig_parts.append(f"addProps={cls.get_structural_signature(ap, depth + 1, include_constraints)}")
|
|
else:
|
|
sig_parts.append(f"addProps={ap}")
|
|
|
|
return '|'.join(sig_parts)
|
|
|
|
@classmethod
|
|
def get_attribute_set(cls, schema: dict) -> frozenset:
|
|
"""
|
|
Get the set of attribute names in a schema (for quick comparison).
|
|
"""
|
|
if not isinstance(schema, dict):
|
|
return frozenset()
|
|
|
|
attrs = set()
|
|
if 'properties' in schema:
|
|
for prop_name, prop_def in schema['properties'].items():
|
|
prop_type = prop_def.get('type', 'ref' if '$ref' in prop_def else 'unknown')
|
|
attrs.add((prop_name, prop_type))
|
|
|
|
return frozenset(attrs)
|
|
|
|
@classmethod
|
|
def compare_schemas(cls, schema1: dict, schema2: dict) -> dict:
|
|
"""
|
|
Compare two schemas and return similarity metrics.
|
|
|
|
Returns:
|
|
dict with keys:
|
|
- exact_match: bool - identical JSON
|
|
- structural_match: bool - same structure ignoring metadata
|
|
- attribute_match: float - Jaccard similarity of attributes (0-1)
|
|
- missing_in_1: set - attributes in schema2 but not schema1
|
|
- missing_in_2: set - attributes in schema1 but not schema2
|
|
"""
|
|
exact = json.dumps(schema1, sort_keys=True) == json.dumps(schema2, sort_keys=True)
|
|
structural = cls.get_structural_signature(schema1) == cls.get_structural_signature(schema2)
|
|
|
|
attrs1 = cls.get_attribute_set(schema1)
|
|
attrs2 = cls.get_attribute_set(schema2)
|
|
|
|
if attrs1 or attrs2:
|
|
intersection = attrs1 & attrs2
|
|
union = attrs1 | attrs2
|
|
jaccard = len(intersection) / len(union) if union else 1.0
|
|
else:
|
|
jaccard = 1.0 if exact else 0.0
|
|
|
|
return {
|
|
'exact_match': exact,
|
|
'structural_match': structural,
|
|
'attribute_match': jaccard,
|
|
'missing_in_1': attrs2 - attrs1,
|
|
'missing_in_2': attrs1 - attrs2,
|
|
}
|
|
|
|
|
|
class SchemaNameAnalyzer:
|
|
"""Analyzes schema names to extract semantic information"""
|
|
|
|
# Operations that return lists
|
|
LIST_INDICATORS = {'GetAll', 'FindAll', 'List', 'Bulk'}
|
|
|
|
@staticmethod
|
|
def remove_dto_suffix(name: str) -> str:
|
|
"""Remove Dto suffix but keep Response/Request"""
|
|
return re.sub(r'Dto$', '', name)
|
|
|
|
@staticmethod
|
|
def extract_type_suffix(name: str) -> str:
|
|
"""Extract Response/Request type"""
|
|
if 'Response' in name:
|
|
return 'Response'
|
|
elif 'Request' in name:
|
|
return 'Request'
|
|
elif 'Body' in name:
|
|
return 'Request' # Body is usually a request
|
|
return ''
|
|
|
|
@classmethod
|
|
def extract_entity(cls, name: str) -> Tuple[str, str, bool]:
|
|
"""
|
|
Extract entity name, action type, and whether it's a list operation.
|
|
|
|
Returns: (entity, action_type, is_list)
|
|
"""
|
|
clean = cls.remove_dto_suffix(name)
|
|
type_suffix = cls.extract_type_suffix(name)
|
|
|
|
# Check if it's a list operation
|
|
is_list = any(indicator in clean for indicator in cls.LIST_INDICATORS)
|
|
|
|
# Patterns to extract entity (ordered by specificity)
|
|
patterns = [
|
|
# GetUserByXxx -> User
|
|
(r'^Get(\w+?)By\w+$', 1),
|
|
# GetOneUser -> User
|
|
(r'^GetOne(\w+)$', 1),
|
|
# GetAllUsers -> User (remove trailing 's')
|
|
(r'^GetAll(\w+?)s?$', 1),
|
|
# CreateUser -> User
|
|
(r'^Create(\w+)$', 1),
|
|
# UpdateUser -> User
|
|
(r'^Update(\w+)$', 1),
|
|
# DeleteUser -> User
|
|
(r'^Delete(\w+)$', 1),
|
|
# DisableUser -> User
|
|
(r'^Disable(\w+)$', 1),
|
|
# EnableUser -> User
|
|
(r'^Enable(\w+)$', 1),
|
|
# ResetUserTraffic -> User
|
|
(r'^Reset(\w+?)(?:Traffic|Data|State)$', 1),
|
|
# RevokeUserSubscription -> User
|
|
(r'^Revoke(\w+?)Subscription$', 1),
|
|
# RestartNode -> Node
|
|
(r'^Restart(\w+)$', 1),
|
|
# BulkDeleteUsers -> User
|
|
(r'^Bulk(?:All)?(?:Delete|Update|Reset|Revoke|Enable|Disable)(\w+?)s?$', 1),
|
|
# AddUsersToSquad -> Squad (operation on squad)
|
|
(r'^(?:Add|Remove)\w+To(\w+)$', 1),
|
|
(r'^(?:Add|Remove)\w+From(\w+)$', 1),
|
|
# ReorderNodes -> Node
|
|
(r'^Reorder(\w+?)s?$', 1),
|
|
# SetInboundToManyHosts -> Host
|
|
(r'^Set\w+ToMany(\w+?)s?$', 1),
|
|
# VerifyPasskeyAuthentication -> Passkey
|
|
(r'^Verify(\w+?)(?:Authentication|Registration)$', 1),
|
|
# GetPasskeyAuthenticationOptions -> Passkey
|
|
(r'^Get(\w+?)(?:Authentication|Registration)Options$', 1),
|
|
# LoginResponse -> Auth
|
|
(r'^(Login|Register|OAuth2Callback|TelegramCallback)$', 0),
|
|
]
|
|
|
|
for pattern, group in patterns:
|
|
match = re.match(pattern, clean)
|
|
if match:
|
|
if group == 0:
|
|
# Special case: Auth-related
|
|
return 'Token', 'auth', is_list
|
|
entity = match.group(group)
|
|
# Remove trailing 's' for plural
|
|
if entity.endswith('s') and len(entity) > 3:
|
|
entity = entity[:-1]
|
|
return entity, 'crud', is_list
|
|
|
|
# Fallback: try to extract meaningful name
|
|
# Remove common prefixes
|
|
for prefix in ['Get', 'Create', 'Update', 'Delete', 'Find', 'Fetch']:
|
|
if clean.startswith(prefix):
|
|
clean = clean[len(prefix):]
|
|
break
|
|
|
|
return clean, 'unknown', is_list
|
|
|
|
@classmethod
|
|
def analyze_group(cls, names: List[str]) -> Dict:
|
|
"""Analyze a group of duplicate schema names"""
|
|
entities = []
|
|
types = set()
|
|
is_list_group = False
|
|
actions = set()
|
|
|
|
for name in names:
|
|
entity, action, is_list = cls.extract_entity(name)
|
|
entities.append(entity)
|
|
types.add(cls.extract_type_suffix(name))
|
|
is_list_group = is_list_group or is_list
|
|
actions.add(action)
|
|
|
|
# Find most common entity
|
|
entity_counts = defaultdict(int)
|
|
for e in entities:
|
|
entity_counts[e] += 1
|
|
|
|
primary_entity = max(entity_counts.keys(), key=lambda x: (entity_counts[x], -len(x)))
|
|
|
|
# Determine response type
|
|
has_response = 'Response' in types
|
|
has_request = 'Request' in types
|
|
|
|
return {
|
|
'entity': primary_entity,
|
|
'is_response': has_response and not has_request,
|
|
'is_request': has_request and not has_response,
|
|
'is_mixed': has_response and has_request,
|
|
'is_list': is_list_group,
|
|
'actions': actions,
|
|
'all_entities': list(set(entities)),
|
|
}
|
|
|
|
|
|
class SmartConsolidator:
|
|
"""Smart schema consolidation with intelligent naming"""
|
|
|
|
# Schema name pairs that must NEVER be consolidated, even if structurally identical.
|
|
# Each entry is a frozenset of schema names that should remain separate.
|
|
DO_NOT_MERGE = [
|
|
frozenset({'CreateExternalSquadRequestDto', 'CreateSubscriptionPageConfigRequestDto'}),
|
|
]
|
|
|
|
# Known entity groupings (for disambiguation)
|
|
ENTITY_GROUPS = {
|
|
# Auth-related
|
|
'Token': {'Login', 'Register', 'OAuth2Callback', 'TelegramCallback', 'VerifyPasskeyAuthentication'},
|
|
# Settings
|
|
'Settings': {'RemnawaveSettings', 'SubscriptionSettings'},
|
|
}
|
|
|
|
# Special naming rules - ordered by specificity (most specific first)
|
|
# Check if MAJORITY of names in group match the pattern
|
|
SPECIAL_PATTERNS = {
|
|
# === HIGH PRIORITY: Specific patterns ===
|
|
|
|
# Event responses (async operations) - very specific pattern
|
|
'EventResponse': [
|
|
r'^BulkAll\w+Response',
|
|
r'^(Add|Remove)UsersTo\w+Response',
|
|
r'^(Add|Remove)UsersFrom\w+Response',
|
|
r'^Restart(All)?Node\w*Response',
|
|
r'^ResetNodeTrafficResponse',
|
|
],
|
|
# Delete responses (simple success) - only Delete* prefix
|
|
'DeleteResponse': [
|
|
r'^Delete\w+Response',
|
|
],
|
|
# Bulk action results (user operations)
|
|
'BulkActionResponse': [
|
|
r'^Bulk(Delete|Reset|Revoke|Update)Users\w*Response',
|
|
],
|
|
# Bulk UUID requests
|
|
'BulkUuidsRequest': [
|
|
r'^Bulk(Delete|Disable|Enable|Reset|Revoke)\w+Request',
|
|
],
|
|
# Reorder requests
|
|
'ReorderRequest': [
|
|
r'^Reorder\w+Request',
|
|
],
|
|
# Token/Auth responses
|
|
'TokenResponse': [
|
|
r'^(Login|Register|OAuth2Callback|TelegramCallback|VerifyPasskeyAuthentication)Response',
|
|
],
|
|
# Passkey options (mixed request/response with same structure)
|
|
'PasskeyOptions': [
|
|
r'^(Get|Verify)Passkey(Authentication|Registration)(Options)?(Response)?$',
|
|
r'^VerifyPasskey(Authentication|Registration)Request',
|
|
],
|
|
# Tags list response
|
|
'TagsResponse': [
|
|
r'^GetAll\w*Tags\w*Response',
|
|
],
|
|
# Inbounds list
|
|
'InboundsResponse': [
|
|
r'^(GetAllInbounds|GetInboundsBy\w+)Response',
|
|
],
|
|
# Subscription response (single)
|
|
'SubscriptionResponse': [
|
|
r'^GetSubscription(By\w+|Info)\w*Response',
|
|
],
|
|
# Settings responses
|
|
'SettingsResponse': [
|
|
r'^(Get|Update)RemnawaveSettings\w*Response',
|
|
],
|
|
'SubscriptionSettingsResponse': [
|
|
r'^(Get|Update)SubscriptionSettings\w*Response',
|
|
],
|
|
# Passkeys list
|
|
'PasskeysResponse': [
|
|
r'^(GetAllPasskeys|DeletePasskey|UpdatePasskey)\w*Response',
|
|
],
|
|
# Snippet request
|
|
'SnippetRequest': [
|
|
r'^(Create|Update)Snippet\w*Request',
|
|
],
|
|
# Snippets response
|
|
'SnippetsResponse': [
|
|
r'^(Create|Update|Delete|Get)Snippets?\w*Response',
|
|
],
|
|
|
|
# === MEDIUM PRIORITY: Entity-specific patterns ===
|
|
|
|
# Users list response (multiple users) - MUST BE BEFORE UserResponse!
|
|
'UsersResponse': [
|
|
r'^GetUserBy(Email|Tag|TelegramId)\w*Response',
|
|
],
|
|
# User single responses (CRUD on single user)
|
|
'UserResponse': [
|
|
r'^(Create|Update|Disable|Enable)UserResponse',
|
|
r'^GetUserBy(Uuid|Username|ShortUuid)\w*Response',
|
|
r'^(Reset|Revoke)User\w+Response',
|
|
],
|
|
# Host list responses
|
|
'HostListResponse': [
|
|
r'^GetAllHostsResponse',
|
|
r'^Bulk(Delete|Disable|Enable)HostsResponse',
|
|
r'^Set\w+ToManyHostsResponse',
|
|
],
|
|
# Host single response
|
|
'HostResponse': [
|
|
r'^(Create|Update|GetOne)HostResponse',
|
|
],
|
|
# Node responses (single)
|
|
'NodeResponse': [
|
|
r'^(Create|Update|GetOne|Disable|Enable)NodeResponse',
|
|
],
|
|
# Nodes list
|
|
'NodesResponse': [
|
|
r'^(GetAllNodes|ReorderNode)Response',
|
|
],
|
|
# Template single response
|
|
'TemplateResponse': [
|
|
r'^(Create|Update|Get)(?:Subscription)?TemplateResponse',
|
|
],
|
|
# Templates list response
|
|
'TemplatesResponse': [
|
|
r'^(GetTemplates|Reorder\w+Templates)Response',
|
|
],
|
|
# Config profile single
|
|
'ConfigProfileResponse': [
|
|
r'^(Create|Update|Get(Computed)?ConfigProfileBy\w+)Response',
|
|
],
|
|
# Config profiles list
|
|
'ConfigProfilesResponse': [
|
|
r'^(GetConfigProfiles|ReorderConfigProfiles)Response',
|
|
],
|
|
# Internal squad single
|
|
'InternalSquadResponse': [
|
|
r'^(Create|Update|GetInternalSquadBy\w+)Response',
|
|
],
|
|
# Internal squads list
|
|
'InternalSquadsResponse': [
|
|
r'^(GetInternalSquads|ReorderInternalSquads)Response',
|
|
],
|
|
# External squad single
|
|
'ExternalSquadResponse': [
|
|
r'^(Create|Update|GetExternalSquadBy\w+)Response',
|
|
],
|
|
# External squads list
|
|
'ExternalSquadsResponse': [
|
|
r'^(GetExternalSquads|ReorderExternalSquads)Response',
|
|
],
|
|
# Infra provider
|
|
'InfraProviderResponse': [
|
|
r'^(Create|Update|GetInfraProviderBy\w+)Response',
|
|
],
|
|
# Billing nodes
|
|
'BillingNodesResponse': [
|
|
r'^(Create|Update|Delete|Get)InfraBillingNode\w*Response',
|
|
],
|
|
# Billing history
|
|
'BillingHistoryResponse': [
|
|
r'^(Create|Delete|Get)InfraBillingHistoryRecord\w*Response',
|
|
],
|
|
# HWID devices
|
|
'HwidDevicesResponse': [
|
|
r'^(Create|Delete(All)?|Get)UserHwidDevice\w*Response',
|
|
],
|
|
}
|
|
|
|
def __init__(self, spec: dict):
|
|
self.spec = spec
|
|
self.schemas = spec.get('components', {}).get('schemas', {})
|
|
self.name_analyzer = SchemaNameAnalyzer()
|
|
self.structure_analyzer = SchemaStructureAnalyzer()
|
|
|
|
def find_duplicates(self, mode: str = 'exact') -> Dict[str, List[str]]:
|
|
"""
|
|
Find duplicate schemas.
|
|
|
|
Args:
|
|
mode: 'exact' - identical JSON content (default)
|
|
'structural' - same structure ignoring metadata (description, example)
|
|
'attributes' - same attribute names and types
|
|
"""
|
|
if mode == 'exact':
|
|
return self._find_exact_duplicates()
|
|
elif mode == 'structural':
|
|
return self._find_structural_duplicates()
|
|
elif mode == 'attributes':
|
|
return self._find_attribute_duplicates()
|
|
else:
|
|
raise ValueError(f"Unknown mode: {mode}")
|
|
|
|
def _find_exact_duplicates(self) -> Dict[str, List[str]]:
|
|
"""Find duplicate schemas by exact JSON content"""
|
|
content_to_names = defaultdict(list)
|
|
|
|
for name, schema_def in self.schemas.items():
|
|
content = json.dumps(schema_def, sort_keys=True)
|
|
content_to_names[content].append(name)
|
|
|
|
return {k: v for k, v in content_to_names.items() if len(v) > 1}
|
|
|
|
def _find_structural_duplicates(self, include_constraints: bool = True) -> Dict[str, List[str]]:
|
|
"""Find duplicate schemas by structure (ignoring metadata like descriptions)"""
|
|
sig_to_names = defaultdict(list)
|
|
|
|
for name, schema_def in self.schemas.items():
|
|
sig = self.structure_analyzer.get_structural_signature(
|
|
schema_def, include_constraints=include_constraints
|
|
)
|
|
sig_to_names[sig].append(name)
|
|
|
|
return {k: v for k, v in sig_to_names.items() if len(v) > 1}
|
|
|
|
def _find_attribute_duplicates(self) -> Dict[str, List[str]]:
|
|
"""Find duplicate schemas by attribute set (name + type pairs)"""
|
|
attrs_to_names = defaultdict(list)
|
|
|
|
for name, schema_def in self.schemas.items():
|
|
attrs = self.structure_analyzer.get_attribute_set(schema_def)
|
|
attrs_to_names[attrs].append(name)
|
|
|
|
return {k: v for k, v in attrs_to_names.items() if len(v) > 1}
|
|
|
|
def analyze_duplicates(self) -> Dict:
|
|
"""
|
|
Analyze duplicates at all levels and return comprehensive report.
|
|
|
|
Returns dict with:
|
|
- exact: Identical JSON content
|
|
- structural: Same structure with constraints (safe to consolidate)
|
|
- structural_loose: Same structure without constraints (may have different validation)
|
|
- attribute: Same attribute names/types (different structure possible)
|
|
- near_duplicates: Structural matches that aren't exact (metadata differs)
|
|
- constraint_only: Schemas that differ only in validation constraints
|
|
"""
|
|
exact = self._find_exact_duplicates()
|
|
structural = self._find_structural_duplicates(include_constraints=True)
|
|
structural_loose = self._find_structural_duplicates(include_constraints=False)
|
|
attribute = self._find_attribute_duplicates()
|
|
|
|
# Find near-duplicates (structural but not exact)
|
|
exact_groups = set(frozenset(v) for v in exact.values())
|
|
near_duplicates = {}
|
|
for sig, names in structural.items():
|
|
names_set = frozenset(names)
|
|
if names_set not in exact_groups:
|
|
near_duplicates[sig] = names
|
|
|
|
# Find constraint-only differences (structural_loose but not structural)
|
|
structural_groups = set(frozenset(v) for v in structural.values())
|
|
constraint_only = {}
|
|
for sig, names in structural_loose.items():
|
|
names_set = frozenset(names)
|
|
if names_set not in structural_groups and len(names) > 1:
|
|
constraint_only[sig] = names
|
|
|
|
return {
|
|
'exact': {
|
|
'count': len(exact),
|
|
'total_schemas': sum(len(v) for v in exact.values()),
|
|
'groups': exact,
|
|
},
|
|
'structural': {
|
|
'count': len(structural),
|
|
'total_schemas': sum(len(v) for v in structural.values()),
|
|
'groups': structural,
|
|
},
|
|
'structural_loose': {
|
|
'count': len(structural_loose),
|
|
'total_schemas': sum(len(v) for v in structural_loose.values()),
|
|
'groups': structural_loose,
|
|
},
|
|
'attribute': {
|
|
'count': len(attribute),
|
|
'total_schemas': sum(len(v) for v in attribute.values()),
|
|
'groups': attribute,
|
|
},
|
|
'near_duplicates': {
|
|
'count': len(near_duplicates),
|
|
'total_schemas': sum(len(v) for v in near_duplicates.values()),
|
|
'groups': near_duplicates,
|
|
},
|
|
'constraint_only': {
|
|
'count': len(constraint_only),
|
|
'total_schemas': sum(len(v) for v in constraint_only.values()),
|
|
'groups': constraint_only,
|
|
},
|
|
}
|
|
|
|
def generate_canonical_name(self, names: List[str]) -> str:
|
|
"""Generate best canonical name for a group of duplicates"""
|
|
|
|
clean_names = [self.name_analyzer.remove_dto_suffix(n) for n in names]
|
|
|
|
# Count how many names match each pattern
|
|
pattern_scores = {}
|
|
for canonical_name, patterns in self.SPECIAL_PATTERNS.items():
|
|
matches = 0
|
|
for pattern in patterns:
|
|
for clean_name in clean_names:
|
|
if re.match(pattern, clean_name):
|
|
matches += 1
|
|
if matches > 0:
|
|
# Score = percentage of names that match
|
|
pattern_scores[canonical_name] = matches / len(names)
|
|
|
|
# Choose pattern with highest coverage (if > 50% match)
|
|
if pattern_scores:
|
|
best_pattern = max(pattern_scores.keys(), key=lambda x: pattern_scores[x])
|
|
if pattern_scores[best_pattern] >= 0.5:
|
|
return best_pattern
|
|
|
|
# Fallback: analyze the group semantically
|
|
analysis = self.name_analyzer.analyze_group(names)
|
|
entity = analysis['entity']
|
|
|
|
# Build canonical name
|
|
if analysis['is_request']:
|
|
if analysis['is_list'] or 'Bulk' in str(names):
|
|
return f'{entity}BulkRequest'
|
|
return f'{entity}Request'
|
|
elif analysis['is_response']:
|
|
if analysis['is_list']:
|
|
return f'{entity}ListResponse'
|
|
return f'{entity}Response'
|
|
else:
|
|
# Mixed or unclear
|
|
return entity
|
|
|
|
def consolidate(self) -> Tuple[Dict[str, str], Dict]:
|
|
"""
|
|
Consolidate duplicate schemas.
|
|
|
|
Returns: (rename_map, stats)
|
|
"""
|
|
duplicates = self.find_duplicates()
|
|
rename_map = {}
|
|
used_canonical_names = set()
|
|
stats = {
|
|
'original_count': len(self.schemas),
|
|
'duplicate_groups': len(duplicates),
|
|
'consolidated_names': {},
|
|
}
|
|
|
|
for content, names in duplicates.items():
|
|
# Check DO_NOT_MERGE: split the group if it contains schemas that must stay separate
|
|
names_set = set(names)
|
|
must_split = False
|
|
for exclusion in self.DO_NOT_MERGE:
|
|
if exclusion.issubset(names_set):
|
|
must_split = True
|
|
break
|
|
|
|
if must_split:
|
|
# Exclude the protected schemas from this group
|
|
excluded = set()
|
|
for exclusion in self.DO_NOT_MERGE:
|
|
if exclusion.issubset(names_set):
|
|
excluded.update(exclusion)
|
|
remaining = [n for n in names if n not in excluded]
|
|
# Only consolidate the remaining names (if 2+)
|
|
if len(remaining) < 2:
|
|
continue
|
|
names = remaining
|
|
|
|
canonical = self.generate_canonical_name(names)
|
|
|
|
# Ensure unique canonical names - if already used, make unique
|
|
base_canonical = canonical
|
|
counter = 2
|
|
while canonical in used_canonical_names:
|
|
canonical = f"{base_canonical}{counter}"
|
|
counter += 1
|
|
|
|
used_canonical_names.add(canonical)
|
|
stats['consolidated_names'][canonical] = names
|
|
|
|
for name in names:
|
|
rename_map[name] = canonical
|
|
|
|
# Calculate actual reduction from rename_map (accounts for DO_NOT_MERGE splits)
|
|
renamed_count = len(rename_map)
|
|
canonical_count = len(set(rename_map.values()))
|
|
stats['final_count'] = len(self.schemas) - renamed_count + canonical_count
|
|
stats['reduction'] = stats['original_count'] - stats['final_count']
|
|
|
|
return rename_map, stats
|
|
|
|
def apply_consolidation(self, rename_map: Dict[str, str]) -> dict:
|
|
"""Apply consolidation to the spec"""
|
|
new_spec = copy.deepcopy(self.spec)
|
|
|
|
# Update schemas
|
|
new_schemas = {}
|
|
for name, schema_def in new_spec['components']['schemas'].items():
|
|
canonical = rename_map.get(name, name)
|
|
if canonical not in new_schemas:
|
|
new_schemas[canonical] = schema_def
|
|
|
|
new_spec['components']['schemas'] = new_schemas
|
|
|
|
# Update all $ref references
|
|
def update_refs(obj):
|
|
if isinstance(obj, dict):
|
|
for key, value in list(obj.items()):
|
|
if key == '$ref' and isinstance(value, str):
|
|
if value.startswith('#/components/schemas/'):
|
|
old_name = value.replace('#/components/schemas/', '')
|
|
new_name = rename_map.get(old_name, old_name)
|
|
obj[key] = f'#/components/schemas/{new_name}'
|
|
else:
|
|
update_refs(value)
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
update_refs(item)
|
|
|
|
update_refs(new_spec)
|
|
|
|
return new_spec
|
|
|
|
|
|
class InlineSchemaExtractor:
|
|
"""
|
|
Extracts inline schemas into $ref definitions to enable ogen reuse.
|
|
|
|
For example, if UserResponse.response and UsersResponse.response[].item
|
|
have the same structure, extract it as a shared schema.
|
|
|
|
Also handles deeply nested inline schemas like:
|
|
- GetExternalSquadsResponseDto.response.externalSquads[].subscriptionSettings
|
|
- CreateUserResponseDto.response.activeInternalSquads[]
|
|
"""
|
|
|
|
def __init__(self, spec: dict):
|
|
self.spec = copy.deepcopy(spec)
|
|
self.schemas = self.spec.get('components', {}).get('schemas', {})
|
|
self.structure_analyzer = SchemaStructureAnalyzer()
|
|
|
|
def get_inline_schema_signature(self, schema: dict) -> Optional[str]:
|
|
"""Get structural signature for an inline object schema"""
|
|
if not isinstance(schema, dict):
|
|
return None
|
|
if schema.get('type') != 'object' or 'properties' not in schema:
|
|
return None
|
|
return self.structure_analyzer.get_structural_signature(schema)
|
|
|
|
def find_all_inline_schemas(self) -> Dict[str, List[Tuple[str, str, dict]]]:
|
|
"""
|
|
Find ALL inline object schemas across all schema definitions at any depth.
|
|
|
|
Returns: {signature: [(parent_schema, property_path, inline_schema), ...]}
|
|
"""
|
|
inline_schemas = defaultdict(list)
|
|
|
|
def extract_recursive(parent_name: str, obj: dict, path: str = ""):
|
|
"""Recursively extract all inline object schemas"""
|
|
if not isinstance(obj, dict):
|
|
return
|
|
|
|
# Check properties
|
|
if 'properties' in obj:
|
|
for prop_name, prop_schema in obj['properties'].items():
|
|
prop_path = f"{path}.{prop_name}" if path else prop_name
|
|
|
|
# Direct inline object
|
|
if prop_schema.get('type') == 'object' and 'properties' in prop_schema:
|
|
# Skip empty objects (no meaningful properties to share)
|
|
if not prop_schema['properties']:
|
|
continue
|
|
sig = self.get_inline_schema_signature(prop_schema)
|
|
if sig:
|
|
inline_schemas[sig].append((parent_name, prop_path, prop_schema))
|
|
# Recurse into nested properties
|
|
extract_recursive(parent_name, prop_schema, prop_path)
|
|
|
|
# Array of inline objects
|
|
elif prop_schema.get('type') == 'array':
|
|
items = prop_schema.get('items', {})
|
|
if items.get('type') == 'object' and 'properties' in items:
|
|
# Skip empty objects
|
|
if not items['properties']:
|
|
continue
|
|
sig = self.get_inline_schema_signature(items)
|
|
if sig:
|
|
inline_schemas[sig].append((parent_name, f"{prop_path}[]", items))
|
|
# Recurse into array item properties
|
|
extract_recursive(parent_name, items, f"{prop_path}[]")
|
|
|
|
for schema_name, schema_def in self.schemas.items():
|
|
extract_recursive(schema_name, schema_def)
|
|
|
|
# Return only groups with duplicates (2+ occurrences)
|
|
return {k: v for k, v in inline_schemas.items() if len(v) >= 2}
|
|
|
|
def generate_extracted_name(self, locations: List[Tuple[str, str, dict]]) -> str:
|
|
"""Generate a name for extracted schema based on usage locations"""
|
|
|
|
parent_names = [loc[0] for loc in locations]
|
|
paths = [loc[1] for loc in locations]
|
|
|
|
# If all paths are "response" or "response[]", use parent entity
|
|
if all(p in ('response', 'response[]') for p in paths):
|
|
entities = []
|
|
for name in parent_names:
|
|
clean = name.replace('Response', '').replace('Request', '').replace('Dto', '')
|
|
for prefix in ['Create', 'Update', 'Get', 'Delete', 'Bulk', 'GetAll', 'GetOne']:
|
|
if clean.startswith(prefix):
|
|
clean = clean[len(prefix):]
|
|
break
|
|
if clean:
|
|
entities.append(clean)
|
|
|
|
if entities:
|
|
common = Counter(entities).most_common(1)[0][0]
|
|
is_array = any('[]' in p for p in paths)
|
|
if is_array:
|
|
return f"{common}Item"
|
|
return common
|
|
|
|
# Extract last property name from paths
|
|
last_props = []
|
|
for path in paths:
|
|
parts = path.replace('[]', '').split('.')
|
|
last_prop = parts[-1] if parts else ''
|
|
if last_prop and last_prop != 'response':
|
|
last_props.append(last_prop)
|
|
|
|
if last_props:
|
|
common_prop = Counter(last_props).most_common(1)[0][0]
|
|
# Capitalize first letter
|
|
name = common_prop[0].upper() + common_prop[1:] if common_prop else common_prop
|
|
|
|
# Generic names need context from parent entity
|
|
generic_names = {'Config', 'Items', 'Data', 'Info', 'Details', 'Settings',
|
|
'Options', 'Params', 'Json', 'Object', 'Value', 'Item'}
|
|
if name in generic_names or name.endswith('Item'):
|
|
# Extract entity context from parent schemas
|
|
entities = []
|
|
for parent in parent_names:
|
|
clean = parent.replace('ResponseDto', '').replace('RequestDto', '').replace('Dto', '')
|
|
for prefix in ['Create', 'Update', 'Get', 'Delete', 'Bulk', 'GetAll', 'GetOne',
|
|
'GetRaw', 'Subscription', 'By', 'All']:
|
|
if clean.startswith(prefix):
|
|
clean = clean[len(prefix):]
|
|
# Take first meaningful word
|
|
if clean:
|
|
entities.append(clean[:20]) # Limit length
|
|
|
|
if entities:
|
|
common_entity = Counter(entities).most_common(1)[0][0]
|
|
# Extract just the entity name (e.g., "SubscriptionByShortUuid" -> "Subscription")
|
|
for suffix in ['ByShortUuid', 'ByUuid', 'ByUsername', 'ByEmail', 'ById']:
|
|
common_entity = common_entity.replace(suffix, '')
|
|
name = f"{common_entity}{name}"
|
|
|
|
# Check if array item
|
|
is_array = any('[]' in p for p in paths)
|
|
if is_array and not name.endswith('Item') and not name.endswith('s'):
|
|
name = f"{name}Item"
|
|
|
|
# Remove trailing 's' for arrays (users[] -> User)
|
|
if is_array and name.endswith('s') and len(name) > 3 and not name.endswith('Settings'):
|
|
name = name[:-1]
|
|
|
|
return name
|
|
|
|
# Fallback: use first parent + path
|
|
first_parent = parent_names[0].replace('ResponseDto', '').replace('RequestDto', '').replace('Dto', '')
|
|
first_path = paths[0].replace('[]', '').split('.')[-1]
|
|
return f"{first_parent}{first_path[0].upper() + first_path[1:] if first_path else ''}"
|
|
|
|
def _resolve_conflict_suffix(self, base_name: str, locations: List[Tuple[str, str, dict]],
|
|
all_groups: List[Tuple[str, List]]) -> str:
|
|
"""
|
|
Generate a unique suffix for a conflicting schema name based on context.
|
|
"""
|
|
paths = [loc[1] for loc in locations]
|
|
parents = [loc[0] for loc in locations]
|
|
|
|
# Helper to check patterns
|
|
def has_pattern(pattern, in_paths=True, in_parents=False):
|
|
if in_paths:
|
|
return any(pattern in p for p in paths)
|
|
if in_parents:
|
|
return any(pattern in p for p in parents)
|
|
return False
|
|
|
|
# Context-based suffix rules
|
|
rules = [
|
|
# Request types (CreateNodeRequestDto.configProfile -> ConfigProfileRef)
|
|
(lambda: any('Request' in p for p in parents) and not any('Response' in p for p in parents), 'Ref'),
|
|
|
|
# Delete response (simple object with isDeleted field)
|
|
(lambda: any('Delete' in p for p in parents) and base_name == 'User', 'Deleted'),
|
|
|
|
# Subscription user info (nested .user field)
|
|
(lambda: any('.user' in p for p in paths), 'Info'),
|
|
|
|
# Subscription settings in squads vs global settings
|
|
(lambda: base_name == 'SubscriptionSettings' and has_pattern('Squad', in_parents=True), 'Squad'),
|
|
|
|
# Templates in squads vs standalone templates
|
|
(lambda: base_name == 'Template' and has_pattern('Squad', in_parents=True), 'Ref'),
|
|
|
|
# Billing records vs subscription request history records
|
|
(lambda: base_name == 'Record' and has_pattern('Billing', in_parents=True), 'Billing'),
|
|
|
|
# Snippets response vs snippet items
|
|
(lambda: base_name == 'Snippet' and has_pattern('snippets[]', in_paths=True), 'Item'),
|
|
|
|
# Inbounds in config profiles vs standalone inbounds
|
|
(lambda: base_name == 'Inbound' and has_pattern('GetAllInbounds', in_parents=True), 'Full'),
|
|
(lambda: base_name == 'Inbound' and has_pattern('.inbound', in_paths=True) and not has_pattern('inbounds[]', in_paths=True), 'Embed'),
|
|
|
|
# Nodes in reorder requests
|
|
(lambda: base_name == 'Node' and has_pattern('Reorder', in_parents=True), 'Order'),
|
|
|
|
# Billing nodes in providers vs standalone
|
|
(lambda: base_name == 'BillingNode' and has_pattern('Provider', in_parents=True), 'Ref'),
|
|
|
|
# Provider in different contexts
|
|
(lambda: base_name == 'ProviderItem' and has_pattern('BillingHistory', in_parents=True), 'HistoryRef'),
|
|
(lambda: base_name == 'ProviderItem' and has_pattern('BillingNode', in_parents=True), 'BillingRef'),
|
|
]
|
|
|
|
for condition, suffix in rules:
|
|
try:
|
|
if condition():
|
|
return suffix
|
|
except Exception:
|
|
pass
|
|
|
|
return ""
|
|
|
|
def extract_inline_schemas(self) -> Tuple[dict, Dict]:
|
|
"""
|
|
Extract duplicate inline schemas into shared definitions.
|
|
Processes from deepest to shallowest to avoid conflicts.
|
|
Handles name conflicts by adding context-based suffixes.
|
|
|
|
Returns: (new_spec, stats)
|
|
"""
|
|
duplicates = self.find_all_inline_schemas()
|
|
|
|
if not duplicates:
|
|
return self.spec, {'extracted_count': 0, 'extracted_schemas': {}}
|
|
|
|
# First pass: group by generated name to detect conflicts
|
|
name_to_groups = defaultdict(list)
|
|
for sig, locations in duplicates.items():
|
|
base_name = self.generate_extracted_name(locations)
|
|
name_to_groups[base_name].append((sig, locations))
|
|
|
|
# Sort by path depth (deepest first) to avoid replacing parent before child
|
|
sorted_duplicates = sorted(
|
|
duplicates.items(),
|
|
key=lambda x: -max(loc[1].count('.') + loc[1].count('[]') for loc in x[1])
|
|
)
|
|
|
|
extracted_schemas = {}
|
|
extraction_log = {}
|
|
sig_to_name = {} # Map signature to final name for conflict resolution
|
|
|
|
# Resolve name conflicts
|
|
for base_name, groups in name_to_groups.items():
|
|
if len(groups) == 1:
|
|
# No conflict
|
|
sig, locations = groups[0]
|
|
sig_to_name[sig] = base_name
|
|
else:
|
|
# Conflict: need unique names for each structure
|
|
used_suffixes = set()
|
|
for i, (sig, locations) in enumerate(groups):
|
|
suffix = self._resolve_conflict_suffix(base_name, locations, groups)
|
|
|
|
# If suffix already used or empty, use index
|
|
if not suffix or suffix in used_suffixes:
|
|
suffix = str(i + 1) if i > 0 else ""
|
|
|
|
used_suffixes.add(suffix)
|
|
final_name = f"{base_name}{suffix}" if suffix else base_name
|
|
sig_to_name[sig] = final_name
|
|
|
|
for sig, locations in sorted_duplicates:
|
|
# Get resolved name
|
|
extracted_name = sig_to_name.get(sig, self.generate_extracted_name(locations))
|
|
|
|
# Ensure unique name (safety check for edge cases)
|
|
base_name = extracted_name
|
|
counter = 2
|
|
while extracted_name in self.schemas or extracted_name in extracted_schemas:
|
|
extracted_name = f"{base_name}{counter}"
|
|
counter += 1
|
|
|
|
# Use first schema as the definition
|
|
schema_def = copy.deepcopy(locations[0][2])
|
|
extracted_schemas[extracted_name] = schema_def
|
|
extraction_log[extracted_name] = [f"{loc[0]}.{loc[1]}" for loc in locations]
|
|
|
|
# Replace all occurrences with $ref
|
|
for parent_name, prop_path, _ in locations:
|
|
self._replace_with_ref(parent_name, prop_path, extracted_name)
|
|
|
|
# Add extracted schemas to spec
|
|
self.spec['components']['schemas'].update(extracted_schemas)
|
|
|
|
return self.spec, {
|
|
'extracted_count': len(extracted_schemas),
|
|
'extracted_schemas': extraction_log
|
|
}
|
|
|
|
def _replace_with_ref(self, schema_name: str, prop_path: str, ref_name: str):
|
|
"""Replace inline schema at path with $ref"""
|
|
schema = self.schemas.get(schema_name)
|
|
if not schema:
|
|
return
|
|
|
|
parts = prop_path.replace('[]', '.[]').split('.')
|
|
parts = [p for p in parts if p] # Remove empty strings
|
|
|
|
current = schema
|
|
for i, part in enumerate(parts[:-1]):
|
|
if part == '[]':
|
|
current = current.get('items', {})
|
|
elif 'properties' in current:
|
|
current = current['properties'].get(part, {})
|
|
else:
|
|
return
|
|
|
|
last_part = parts[-1]
|
|
if last_part == '[]':
|
|
# Replace array items
|
|
current['items'] = {'$ref': f'#/components/schemas/{ref_name}'}
|
|
elif 'properties' in current and last_part in current['properties']:
|
|
# Replace property
|
|
current['properties'][last_part] = {'$ref': f'#/components/schemas/{ref_name}'}
|
|
|
|
|
|
def fix_nullable_without_type(spec: dict) -> Tuple[dict, int]:
|
|
"""
|
|
Fix schema properties that have 'nullable: true' but no 'type' field.
|
|
Ogen generates invalid code for such properties.
|
|
|
|
For empty objects with nullable: true, we use x-ogen-name with {} to make ogen
|
|
generate jx.Raw type which properly handles null values.
|
|
|
|
Returns: (spec, count_of_fixed_properties)
|
|
"""
|
|
spec = copy.deepcopy(spec)
|
|
fixed_count = 0
|
|
|
|
def is_empty_object(schema: dict) -> bool:
|
|
"""Check if schema is an empty object (type: object with no/empty properties)"""
|
|
if schema.get('type') != 'object':
|
|
return False
|
|
props = schema.get('properties', {})
|
|
return len(props) == 0
|
|
|
|
def fix_schema(schema: dict, path: str = "") -> int:
|
|
"""Recursively fix nullable properties without type"""
|
|
count = 0
|
|
|
|
if not isinstance(schema, dict):
|
|
return 0
|
|
|
|
# For nullable empty objects (or nullable without type), convert to {} (any JSON)
|
|
# This makes ogen generate jx.Raw which properly handles null
|
|
if schema.get('nullable') is True and 'type' not in schema and '$ref' not in schema:
|
|
# No type → treat as empty object, strip to bare {nullable: true}
|
|
schema.pop('properties', None)
|
|
count += 1
|
|
elif schema.get('nullable') is True and is_empty_object(schema):
|
|
# Has type: object but no properties → strip to bare {nullable: true}
|
|
schema.pop('properties', None)
|
|
schema.pop('type', None)
|
|
count += 1
|
|
|
|
# Recurse into properties
|
|
if 'properties' in schema:
|
|
for prop_name, prop_def in schema['properties'].items():
|
|
if isinstance(prop_def, dict):
|
|
count += fix_schema(prop_def, f"{path}.properties.{prop_name}")
|
|
|
|
# Recurse into items
|
|
if 'items' in schema and isinstance(schema['items'], dict):
|
|
count += fix_schema(schema['items'], f"{path}.items")
|
|
|
|
# Recurse into allOf/oneOf/anyOf
|
|
for key in ['allOf', 'oneOf', 'anyOf']:
|
|
if key in schema:
|
|
for i, sub_schema in enumerate(schema[key]):
|
|
if isinstance(sub_schema, dict):
|
|
count += fix_schema(sub_schema, f"{path}.{key}[{i}]")
|
|
|
|
# Recurse into additionalProperties
|
|
if 'additionalProperties' in schema and isinstance(schema['additionalProperties'], dict):
|
|
count += fix_schema(schema['additionalProperties'], f"{path}.additionalProperties")
|
|
|
|
return count
|
|
|
|
# Fix schemas in components
|
|
for schema_name, schema_def in spec.get('components', {}).get('schemas', {}).items():
|
|
fixed_count += fix_schema(schema_def, f"#/components/schemas/{schema_name}")
|
|
|
|
# Fix inline schemas in paths
|
|
for path, methods in spec.get('paths', {}).items():
|
|
for method, op in methods.items():
|
|
if not isinstance(op, dict):
|
|
continue
|
|
|
|
# Fix request body
|
|
if 'requestBody' in op:
|
|
for content_type, content in op['requestBody'].get('content', {}).items():
|
|
if 'schema' in content:
|
|
fixed_count += fix_schema(content['schema'], f"paths.{path}.{method}.requestBody")
|
|
|
|
# Fix responses
|
|
for status, response in op.get('responses', {}).items():
|
|
if isinstance(response, dict):
|
|
for content_type, content in response.get('content', {}).items():
|
|
if 'schema' in content:
|
|
fixed_count += fix_schema(content['schema'], f"paths.{path}.{method}.responses.{status}")
|
|
|
|
return spec, fixed_count
|
|
|
|
|
|
def unify_error_responses(spec: dict) -> Tuple[dict, Dict]:
|
|
"""
|
|
Unify duplicate error response schemas (400 BadRequest, 401 Unauthorized, etc.)
|
|
by extracting common inline schemas into shared definitions.
|
|
|
|
Returns: (new_spec, stats)
|
|
"""
|
|
spec = copy.deepcopy(spec)
|
|
|
|
# Define common error schemas to extract
|
|
ERROR_SCHEMAS = {
|
|
'BadRequestError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'message': {'type': 'string'},
|
|
'statusCode': {'type': 'number', 'example': 400},
|
|
'errors': {
|
|
'type': 'array',
|
|
'items': {
|
|
'$ref': '#/components/schemas/ValidationError'
|
|
}
|
|
}
|
|
},
|
|
'required': ['message', 'statusCode', 'errors']
|
|
},
|
|
'ValidationError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'validation': {'type': 'string', 'example': 'uuid'},
|
|
'code': {'type': 'string', 'example': 'invalid_string'},
|
|
'message': {'type': 'string', 'example': 'Invalid uuid'},
|
|
'path': {
|
|
'type': 'array',
|
|
'items': {'type': 'string'},
|
|
'example': ['uuid']
|
|
}
|
|
},
|
|
'required': ['validation', 'code', 'message', 'path']
|
|
},
|
|
'UnauthorizedError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'message': {'type': 'string', 'example': 'Unauthorized'},
|
|
'statusCode': {'type': 'number', 'example': 401}
|
|
},
|
|
'required': ['message', 'statusCode']
|
|
},
|
|
'ForbiddenError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'message': {'type': 'string', 'example': 'Forbidden'},
|
|
'statusCode': {'type': 'number', 'example': 403}
|
|
},
|
|
'required': ['message', 'statusCode']
|
|
},
|
|
'NotFoundError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'message': {'type': 'string', 'example': 'Not Found'},
|
|
'statusCode': {'type': 'number', 'example': 404}
|
|
},
|
|
'required': ['message', 'statusCode']
|
|
},
|
|
'InternalServerError': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'timestamp': {'type': 'string'},
|
|
'path': {'type': 'string'},
|
|
'message': {'type': 'string'},
|
|
'errorCode': {'type': 'string'}
|
|
}
|
|
}
|
|
}
|
|
|
|
# Add schemas to spec
|
|
if 'components' not in spec:
|
|
spec['components'] = {}
|
|
if 'schemas' not in spec['components']:
|
|
spec['components']['schemas'] = {}
|
|
|
|
spec['components']['schemas'].update(ERROR_SCHEMAS)
|
|
|
|
# Map status codes to schema names
|
|
STATUS_TO_SCHEMA = {
|
|
'400': 'BadRequestError',
|
|
'401': 'UnauthorizedError',
|
|
'403': 'ForbiddenError',
|
|
'404': 'NotFoundError',
|
|
'500': 'InternalServerError',
|
|
}
|
|
|
|
# Replace inline error schemas with $ref
|
|
replaced_count = {status: 0 for status in STATUS_TO_SCHEMA}
|
|
|
|
for path, methods in spec.get('paths', {}).items():
|
|
for method, op in methods.items():
|
|
if not isinstance(op, dict) or 'responses' not in op:
|
|
continue
|
|
|
|
for status_code, schema_name in STATUS_TO_SCHEMA.items():
|
|
if status_code in op['responses']:
|
|
response = op['responses'][status_code]
|
|
|
|
# For 404 responses without content, add content with NotFoundError schema
|
|
# This prevents ogen from generating empty per-operation NotFound structs
|
|
if status_code == '404' and 'content' not in response:
|
|
response['content'] = {
|
|
'application/json': {
|
|
'schema': {
|
|
'$ref': f'#/components/schemas/{schema_name}'
|
|
}
|
|
}
|
|
}
|
|
replaced_count[status_code] += 1
|
|
elif 'content' in response and 'application/json' in response['content']:
|
|
# Replace inline schema with $ref
|
|
response['content']['application/json']['schema'] = {
|
|
'$ref': f'#/components/schemas/{schema_name}'
|
|
}
|
|
replaced_count[status_code] += 1
|
|
|
|
total_replaced = sum(replaced_count.values())
|
|
|
|
return spec, {
|
|
'schemas_added': len(ERROR_SCHEMAS),
|
|
'responses_unified': replaced_count,
|
|
'total_replaced': total_replaced
|
|
}
|
|
|
|
|
|
def analyze_spec(spec_file: str):
|
|
"""Analyze a spec file and show consolidation suggestions"""
|
|
with open(spec_file, 'r') as f:
|
|
spec = json.load(f)
|
|
|
|
consolidator = SmartConsolidator(spec)
|
|
rename_map, stats = consolidator.consolidate()
|
|
|
|
print(f"=== SMART CONSOLIDATION ANALYSIS ===\n")
|
|
print(f"Original schemas: {stats['original_count']}")
|
|
print(f"Duplicate groups: {stats['duplicate_groups']}")
|
|
print(f"After consolidation: {stats['final_count']}")
|
|
print(f"Reduction: {stats['reduction']} ({stats['reduction']*100//stats['original_count']}%)")
|
|
|
|
print(f"\n=== CONSOLIDATED NAMES ===\n")
|
|
for canonical, names in sorted(stats['consolidated_names'].items()):
|
|
print(f"\n{canonical} ({len(names)} merged):")
|
|
for n in sorted(names):
|
|
print(f" <- {n}")
|
|
|
|
return consolidator, rename_map, stats
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import sys
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python smart_consolidate.py <spec.json>")
|
|
sys.exit(1)
|
|
|
|
analyze_spec(sys.argv[1])
|