410 lines
14 KiB
Python
410 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mass migration: [DEF:id:Type] -> #region id [C:N] [TYPE Type] [SEMANTICS ...]
|
|
|
|
Processes all Python files under backend/src/ that contain [DEF: annotations.
|
|
|
|
STRATEGY: For each [DEF:...[/DEF:...] block, only replace the two boundary lines
|
|
and remove the intermediate metadata tag lines. This preserves any already-converted
|
|
inner #region blocks when processing nested DEF blocks.
|
|
|
|
Consolidation rules:
|
|
- COMPLEXITY + [C:N]: keep only inline [C:N], remove @COMPLEXITY line
|
|
- Only @COMPLEXITY: move to inline [C:N], remove @COMPLEXITY line
|
|
- BRIEF + PURPOSE: keep @BRIEF, remove @PURPOSE line
|
|
- Only @PURPOSE: rename to @BRIEF
|
|
- @SEMANTICS: ... -> [SEMANTICS ...] inline in region header
|
|
- @LAYER: ... -> @LAYER ... (normalized format)
|
|
- @RELATION: ..., @PRE:..., @POST:..., etc -> normalize format
|
|
"""
|
|
|
|
import re
|
|
import os
|
|
import sys
|
|
|
|
SOURCE_DIR = os.path.join(os.path.dirname(__file__), "src")
|
|
|
|
# Matches: # [DEF:ContractId:Type]
|
|
DEF_OPEN_RE = re.compile(r'^(\s*)#\s*\[DEF:([^:]+):([^\]]+)\]')
|
|
|
|
# Matches: # [/DEF:ContractId] or # [/DEF:ContractId:Type]
|
|
DEF_CLOSE_RE = re.compile(r'^(\s*)#\s*\[/DEF:([^:\]]+)')
|
|
|
|
# Inline [C:N] pattern
|
|
C_INLINE_RE = re.compile(r'\[C:\s*(\d+)\]')
|
|
|
|
# Tag lines to match
|
|
TAG_LINE_RE = re.compile(
|
|
r'^(\s*)#\s*@(COMPLEXITY|BRIEF|PURPOSE|SEMANTICS|LAYER|RELATION|'
|
|
r'PRE|POST|SIDE_EFFECT|DATA_CONTRACT|INVARIANT|RETURN|RATIONALE|REJECTED|'
|
|
r'STATUS|DEPRECATED)\s*:\s*(.+?)\s*$',
|
|
re.IGNORECASE
|
|
)
|
|
|
|
|
|
def _ids_match(close_id, open_id):
|
|
"""Check if close_id matches open_id, handling fully qualified paths."""
|
|
if close_id == open_id:
|
|
return True
|
|
# Case-insensitive comparison
|
|
c_lower = close_id.lower().rstrip('.')
|
|
o_lower = open_id.lower().rstrip('.')
|
|
if c_lower == o_lower:
|
|
return True
|
|
# Fully qualified close e.g. backend.src.services.X.Y
|
|
if c_lower.endswith('.' + o_lower):
|
|
return True
|
|
if o_lower.endswith('.' + c_lower):
|
|
return True
|
|
# Last component matches
|
|
if c_lower.split('.')[-1] == o_lower:
|
|
return True
|
|
if o_lower.split('.')[-1] == c_lower:
|
|
return True
|
|
# One contains the other
|
|
if o_lower in c_lower or c_lower in o_lower:
|
|
return True
|
|
return False
|
|
|
|
|
|
def normalize_relation(text):
|
|
text = text.strip()
|
|
# [PRED] -> [Target]
|
|
m = re.match(r'\[(\w+)\]\s*->\s*\[(.+?)\]$', text)
|
|
if m:
|
|
return f"{m.group(1)} -> [{m.group(2)}]"
|
|
# PRED -> [Target]
|
|
m = re.match(r'(\w+)\s*->\s*\[(.+?)\]$', text)
|
|
if m:
|
|
return f"{m.group(1)} -> [{m.group(2)}]"
|
|
# PRED -> Target
|
|
m = re.match(r'(\w+)\s*->\s*(.+)$', text)
|
|
if m:
|
|
target = m.group(2).strip()
|
|
if not target.startswith('['):
|
|
target = f'[{target}]'
|
|
return f"{m.group(1)} -> {target}"
|
|
return text
|
|
|
|
|
|
def process_file(filepath):
|
|
"""Process a single file, converting [DEF:... blocks to #region/#endregion.
|
|
|
|
Only replaces boundary lines and removes metadata tags between them.
|
|
Does NOT replace the body content, preserving nested conversions.
|
|
"""
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()
|
|
|
|
if not lines:
|
|
return False
|
|
|
|
# Quick check
|
|
if not any('[DEF:' in line for line in lines):
|
|
return False
|
|
|
|
###############################################################
|
|
# PASS 1: Find all opens and match them to closes
|
|
###############################################################
|
|
opens = [] # (line_idx, contract_id, type_str, indent)
|
|
for i, line in enumerate(lines):
|
|
m = DEF_OPEN_RE.match(line)
|
|
if m:
|
|
opens.append((i, m.group(2).strip(), m.group(3).strip(), m.group(1)))
|
|
|
|
# Match closes to opens using stack
|
|
close_for_open = {} # open_idx -> close_idx
|
|
open_stack = []
|
|
for i, line in enumerate(lines):
|
|
m = DEF_CLOSE_RE.match(line)
|
|
if m:
|
|
close_id = m.group(2).strip()
|
|
matched = False
|
|
for j in range(len(open_stack) - 1, -1, -1):
|
|
oi, oid, otype, oindent = open_stack[j]
|
|
if _ids_match(close_id, oid):
|
|
close_for_open[oi] = i
|
|
open_stack.pop(j)
|
|
matched = True
|
|
break
|
|
if not matched:
|
|
pass # Silently skip unmatched closes
|
|
else:
|
|
m2 = DEF_OPEN_RE.match(line)
|
|
if m2:
|
|
open_stack.append((i, m2.group(2).strip(), m2.group(3).strip(), m2.group(1)))
|
|
|
|
if not close_for_open:
|
|
return False
|
|
|
|
###############################################################
|
|
# PASS 2: For each matched block, find metadata and plan edits
|
|
###############################################################
|
|
# Edits: list of (line_idx, action, data)
|
|
# action = 'replace_header' | 'replace_footer' | 'remove_line'
|
|
edits = []
|
|
|
|
for open_idx, contract_id, type_str, indent in opens:
|
|
close_idx = close_for_open.get(open_idx)
|
|
if close_idx is None:
|
|
continue
|
|
|
|
# Find nested opens to know where to stop scanning for metadata
|
|
nested_opens = [oi for oi, _, _, _ in opens if open_idx < oi < close_idx]
|
|
|
|
# Scan for metadata lines between open and first code/nested open
|
|
metadata_indices = set()
|
|
metadata = {} # line_idx -> (tagname, content)
|
|
|
|
for i in range(open_idx + 1, close_idx):
|
|
# Stop at nested DEF open
|
|
if i in nested_opens:
|
|
break
|
|
|
|
stripped = lines[i].strip()
|
|
|
|
# Stop at code line
|
|
if stripped and not stripped.startswith('#'):
|
|
break
|
|
|
|
# Skip blank lines, bare comments, and inline comments
|
|
if not stripped or stripped == '#':
|
|
continue
|
|
|
|
# Check for tag
|
|
m = TAG_LINE_RE.match(lines[i])
|
|
if m:
|
|
tagname = m.group(2).upper()
|
|
content = m.group(3).strip()
|
|
metadata_indices.add(i)
|
|
metadata[i] = (tagname, content)
|
|
else:
|
|
# Non-tag comment - stop here
|
|
break
|
|
|
|
# Check for inline [C:N]
|
|
c_val = None
|
|
c_inline = C_INLINE_RE.search(lines[open_idx])
|
|
if c_inline:
|
|
c_val = int(c_inline.group(1))
|
|
|
|
# Process metadata with deduplication
|
|
brief_val = None
|
|
purpose_val = None
|
|
sem_val = None
|
|
layer_val = None
|
|
relations = []
|
|
pres = []
|
|
posts = []
|
|
side_effects = []
|
|
data_contracts = []
|
|
invariants = []
|
|
returns = []
|
|
rationales = []
|
|
rejecteds = []
|
|
statuses = []
|
|
|
|
for line_idx, (tagname, content) in sorted(metadata.items()):
|
|
if tagname == 'COMPLEXITY':
|
|
if c_val is None:
|
|
try:
|
|
c_val = int(content)
|
|
except ValueError:
|
|
pass
|
|
elif tagname == 'BRIEF':
|
|
if brief_val is None:
|
|
brief_val = content
|
|
elif tagname == 'PURPOSE':
|
|
if brief_val is not None:
|
|
pass # Drop purpose when brief exists
|
|
elif purpose_val is None:
|
|
purpose_val = content
|
|
elif tagname == 'SEMANTICS':
|
|
if sem_val is None:
|
|
sem_val = content
|
|
elif tagname == 'LAYER':
|
|
if layer_val is None:
|
|
layer_val = content
|
|
elif tagname == 'RELATION':
|
|
relations.append(normalize_relation(content))
|
|
elif tagname == 'PRE':
|
|
pres.append(content)
|
|
elif tagname == 'POST':
|
|
posts.append(content)
|
|
elif tagname == 'SIDE_EFFECT':
|
|
side_effects.append(content)
|
|
elif tagname == 'DATA_CONTRACT':
|
|
data_contracts.append(content)
|
|
elif tagname == 'INVARIANT':
|
|
invariants.append(content)
|
|
elif tagname == 'RETURN':
|
|
returns.append(content)
|
|
elif tagname == 'RATIONALE':
|
|
rationales.append(content)
|
|
elif tagname == 'REJECTED':
|
|
rejecteds.append(content)
|
|
elif tagname in ('STATUS', 'DEPRECATED'):
|
|
statuses.append(f'{tagname.upper()} {content}')
|
|
|
|
# Consolidate brief/purpose
|
|
if purpose_val and not brief_val:
|
|
brief_val = purpose_val
|
|
# If both existed, purpose was already discarded above
|
|
|
|
# Build region header
|
|
c_part = f' [C:{c_val}]' if c_val is not None else ''
|
|
sem_part = f' [SEMANTICS {sem_val}]' if sem_val else ''
|
|
region_header = f'{indent}# #region {contract_id}{c_part} [TYPE {type_str}]{sem_part}'
|
|
|
|
# Build metadata body
|
|
meta_lines = []
|
|
if brief_val:
|
|
meta_lines.append(f'{indent}# @BRIEF {brief_val}')
|
|
if layer_val:
|
|
meta_lines.append(f'{indent}# @LAYER {layer_val}')
|
|
for pre in pres:
|
|
meta_lines.append(f'{indent}# @PRE {pre}')
|
|
for post in posts:
|
|
meta_lines.append(f'{indent}# @POST {post}')
|
|
for se in side_effects:
|
|
meta_lines.append(f'{indent}# @SIDE_EFFECT {se}')
|
|
for dc in data_contracts:
|
|
meta_lines.append(f'{indent}# @DATA_CONTRACT {dc}')
|
|
for inv in invariants:
|
|
meta_lines.append(f'{indent}# @INVARIANT {inv}')
|
|
for ret in returns:
|
|
meta_lines.append(f'{indent}# @RETURN {ret}')
|
|
for rel in relations:
|
|
meta_lines.append(f'{indent}# @RELATION {rel}')
|
|
for st in statuses:
|
|
meta_lines.append(f'{indent}# @{st}')
|
|
for rat in rationales:
|
|
meta_lines.append(f'{indent}# @RATIONALE {rat}')
|
|
for rej in rejecteds:
|
|
meta_lines.append(f'{indent}# @REJECTED {rej}')
|
|
|
|
region_footer = f'{indent}# #endregion {contract_id}'
|
|
|
|
# Plan edits:
|
|
# 1. Replace [DEF: line with region_header + meta_lines
|
|
edits.append(('replace_multi', open_idx, [region_header] + meta_lines))
|
|
|
|
# 2. Remove each metadata tag line
|
|
for mi in metadata_indices:
|
|
edits.append(('remove', mi, None))
|
|
|
|
# 3. Replace [/DEF: line with #endregion
|
|
edits.append(('replace_single', close_idx, region_footer))
|
|
|
|
if not edits:
|
|
return False
|
|
|
|
# Process from bottom to top (descending line number) to preserve indices
|
|
edits.sort(key=lambda e: e[1], reverse=True)
|
|
|
|
for action, line_idx, data in edits:
|
|
if action == 'replace_multi':
|
|
lines[line_idx] = data[0] + '\n'
|
|
# Insert remaining lines after
|
|
for j, extra_line in enumerate(data[1:], 1):
|
|
lines.insert(line_idx + j, extra_line + '\n')
|
|
elif action == 'replace_single':
|
|
lines[line_idx] = data + '\n'
|
|
elif action == 'remove':
|
|
lines[line_idx] = None
|
|
|
|
# Strip None lines
|
|
lines = [l for l in lines if l is not None]
|
|
|
|
# Remove empty bare comment lines that might have been left
|
|
cleaned = []
|
|
for line in lines:
|
|
if line.strip() == '#':
|
|
continue
|
|
cleaned.append(line)
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.writelines(cleaned)
|
|
|
|
return True
|
|
|
|
|
|
def find_remaining_defs():
|
|
"""Find all files that still have [DEF: patterns."""
|
|
remaining = []
|
|
for root, dirs, files in os.walk(SOURCE_DIR):
|
|
dirs[:] = [d for d in dirs if d not in ('__pycache__', '.venv', 'node_modules')]
|
|
for f in files:
|
|
if f.endswith('.py'):
|
|
fp = os.path.join(root, f)
|
|
try:
|
|
with open(fp, 'r', encoding='utf-8') as fh:
|
|
content = fh.read()
|
|
if '[DEF:' in content:
|
|
remaining.append(fp)
|
|
except Exception:
|
|
pass
|
|
return remaining
|
|
|
|
|
|
def main():
|
|
all_files = []
|
|
for root, dirs, files in os.walk(SOURCE_DIR):
|
|
dirs[:] = [d for d in dirs if d not in ('__pycache__', '.venv', 'node_modules')]
|
|
for f in files:
|
|
if f.endswith('.py'):
|
|
all_files.append(os.path.join(root, f))
|
|
|
|
print(f"Scanning {len(all_files)} Python files...")
|
|
|
|
# Multi-pass until stable
|
|
max_passes = 5
|
|
for pass_num in range(1, max_passes + 1):
|
|
remaining = find_remaining_defs()
|
|
if not remaining:
|
|
print(f"\n✅ All DEF annotations migrated after pass {pass_num - 1}!")
|
|
return
|
|
|
|
print(f"\nPass {pass_num}: Processing {len(remaining)} remaining files...")
|
|
modified = 0
|
|
errors = 0
|
|
|
|
for fp in sorted(remaining):
|
|
rel_path = os.path.relpath(fp, SOURCE_DIR)
|
|
try:
|
|
changed = process_file(fp)
|
|
if changed:
|
|
print(f" ✓ {rel_path}")
|
|
modified += 1
|
|
except Exception as e:
|
|
print(f" ✗ {rel_path}: {e}")
|
|
errors += 1
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
print(f" Modified: {modified}, Errors: {errors}")
|
|
|
|
if modified == 0 and errors == 0:
|
|
# No changes and no errors - files might have DEF patterns
|
|
# that can't be matched (e.g. orphaned opens/closes)
|
|
break
|
|
|
|
# Final report
|
|
remaining = find_remaining_defs()
|
|
if remaining:
|
|
print(f"\n⚠️ {len(remaining)} files still have unconverted [DEF: patterns:")
|
|
for fp in sorted(remaining):
|
|
rel = os.path.relpath(fp, SOURCE_DIR)
|
|
# Count the DEF patterns
|
|
with open(fp, 'r') as f:
|
|
content = f.read()
|
|
opens = content.count('[DEF:')
|
|
closes = content.count('[/DEF:')
|
|
print(f" {rel} ({opens} opens, {closes} closes)")
|
|
else:
|
|
print(f"\n✅ All [DEF:] annotations have been migrated to #region format!")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|