rename_duplicate_dals Namespace Reference

Classes
class	ConfigCommitError
class	DalCollector
class	DalGroup
class	ExtendedDal
class	RelationshipCache
class	RenameDalCli

Functions
list	_natural_sort_key (str s)
List[DalGroup]	sort_groups_by_depth (List[DalGroup] groups, Dict[Configuration, RelationshipCache] trees)
	rename_duplicate_dals (str input_folder)

Variables
	console = Console()

Detailed Description

HW: Finds non-unique DAL objects and provides a simple CLI to rename them iteratively

Function Documentation

_natural_sort_key()

list rename_duplicate_dals._natural_sort_key ( str s )

protected

Definition at line 160 of file rename_duplicate_dals.py.

def _natural_sort_key(s: str) -> list:
    return [int(c) if c.isdigit() else c.lower() for c in re.split(r"(\d+)", s)]
 
 

rename_duplicate_dals()

rename_duplicate_dals.rename_duplicate_dals

(

str

input_folder

)

Definition at line 379 of file rename_duplicate_dals.py.

def rename_duplicate_dals(input_folder: str):
    folder = Path(input_folder)
    if not folder.exists() or not folder.is_dir():
        raise FileNotFoundError(f"Cannot find {folder}")
 
    configs = [Configuration(f"oksconflibs:{f}") for f in folder.glob("*.data.xml")]
    if not configs:
        raise FileNotFoundError(
            "No configs found — ensure the folder contains .data.xml files."
        )
 
    collector = DalCollector(configs)
    RenameDalCli(collector).run()
 
 

sort_groups_by_depth()

List[DalGroup] rename_duplicate_dals.sort_groups_by_depth	(	List[DalGroup]	groups,
		Dict[Configuration, RelationshipCache]	trees )

Sort DalGroups by their depth in the configuration tree (BFS from Session root).

Definition at line 164 of file rename_duplicate_dals.py.

) -> List[DalGroup]:
    """Sort DalGroups by their depth in the configuration tree (BFS from Session root)."""
    all_ids = {g.dals[0].id for g in groups}
    tree_list = list(trees.values())
    covered: Set[str] = set()
    sorted_groups: List[DalGroup] = []
    iteration = 0
 
    while covered != all_ids:
        iteration += 1
 
        if iteration == 1:
            sessions = tree_list[0].db.get_dals("Session")
            if not sessions:
                raise ValueError("Configuration must contain a Session object!")
            root = sessions[0]
        else:
            remaining = all_ids - covered
            root = next((g.dals[0] for g in groups if g.dals[0].id in remaining), None)
            if root is None:
                break
 
        # BFS to build depth map
        depth_map: Dict[str, int] = {}
        queue = [(root, 0)]
        visited: Set[int] = set()
 
        while queue:
            obj, depth = queue.pop(0)
            if id(obj) in visited:
                continue
            visited.add(id(obj))
            depth_map[obj.id] = depth
            covered.add(obj.id)
 
            seen_children: Set[int] = set()
            for tree in tree_list:
                for child in tree.graph.get(obj, []):
                    if id(child) not in visited and id(child) not in seen_children:
                        queue.append((child, depth + 1))
                        seen_children.add(id(child))
 
        batch = [g for g in groups if g.dals[0].id in depth_map]
        batch.sort(key=lambda g: (depth_map[g.dals[0].id], _natural_sort_key(g.dals[0].id)))
        sorted_groups.extend(batch)
 
    return sorted_groups
 
 
# ----- DAL Collector -----
 

Variable Documentation

console

rename_duplicate_dals.console = Console()

Definition at line 19 of file rename_duplicate_dals.py.