Skip to content

Batch

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.file.operations.detectors.batch

Batch operation detectors.

Classes

BatchOperationDetector

Detects batch operations and rename sequences.

Functions
detect_backup_create 
detect_backup_create(
    events: list[FileEvent],
) -> FileOperation | None

Detect backup creation pattern.

Source code in provide/foundation/file/operations/detectors/batch.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def detect_backup_create(self, events: list[FileEvent]) -> FileOperation | None:
    """Detect backup creation pattern."""
    if len(events) < 2:
        return None

    # Look for move to backup name followed by create of original
    for i in range(len(events) - 1):
        move_event = events[i]
        create_event = events[i + 1]

        if (
            move_event.event_type == "moved"
            and create_event.event_type == "created"
            and is_backup_file(move_event.dest_path or move_event.path)
            and move_event.path == create_event.path
            and not is_temp_file(create_event.path)
        ):
            # Time window check (backup operations usually happen quickly)
            time_diff = (create_event.timestamp - move_event.timestamp).total_seconds()
            if time_diff <= 2.0:
                return FileOperation(
                    operation_type=OperationType.BACKUP_CREATE,
                    primary_path=create_event.path,
                    events=[move_event, create_event],
                    confidence=0.90,
                    description=f"Backup created for {create_event.path.name}",
                    start_time=move_event.timestamp,
                    end_time=create_event.timestamp,
                    is_atomic=True,
                    is_safe=True,
                    has_backup=True,
                    files_affected=[create_event.path],
                    metadata={
                        "backup_file": str(move_event.dest_path or move_event.path),
                        "pattern": "backup_create",
                    },
                )

    return None
detect_batch_update 
detect_batch_update(
    events: list[FileEvent],
) -> FileOperation | None

Detect batch update pattern (multiple related files updated together).

Source code in provide/foundation/file/operations/detectors/batch.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def detect_batch_update(self, events: list[FileEvent]) -> FileOperation | None:
    """Detect batch update pattern (multiple related files updated together)."""
    if len(events) < 3:
        return None

    # Group events by directory and time proximity
    directory_groups = defaultdict(list)
    for event in events:
        if event.event_type in {"created", "modified", "deleted"}:
            directory_groups[event.path.parent].append(event)

    for directory, dir_events in directory_groups.items():
        if len(dir_events) < 3:
            continue

        dir_events.sort(key=lambda e: e.timestamp)

        # Check if events are clustered in time (within 5 seconds)
        time_span = (dir_events[-1].timestamp - dir_events[0].timestamp).total_seconds()
        if time_span <= 5.0 and self._files_are_related(dir_events):
            return FileOperation(
                operation_type=OperationType.BATCH_UPDATE,
                primary_path=directory,
                events=dir_events,
                confidence=0.85,
                description=f"Batch operation on {len(dir_events)} files",
                start_time=dir_events[0].timestamp,
                end_time=dir_events[-1].timestamp,
                is_atomic=False,
                is_safe=True,
                files_affected=[e.path for e in dir_events],
                metadata={
                    "file_count": len(dir_events),
                    "pattern": "batch_update",
                },
            )

    return None
detect_rename_sequence 
detect_rename_sequence(
    events: list[FileEvent],
) -> FileOperation | None

Detect rename sequence pattern.

Source code in provide/foundation/file/operations/detectors/batch.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def detect_rename_sequence(self, events: list[FileEvent]) -> FileOperation | None:
    """Detect rename sequence pattern."""
    if len(events) < 2:
        return None

    # Look for chain of moves: A -> B -> C
    move_events = [e for e in events if e.event_type == "moved"]
    if len(move_events) < 2:
        return None

    # Build rename chains
    chains = []
    for move_event in move_events:
        # Find chains where this move's source path is another move's destination
        chain = [move_event]

        # Look backwards
        current_src = move_event.path
        for other_move in move_events:
            if other_move != move_event and other_move.dest_path == current_src:
                chain.insert(0, other_move)
                current_src = other_move.path

        # Look forwards
        current_dest = move_event.dest_path
        for other_move in move_events:
            if other_move != move_event and other_move.path == current_dest:
                chain.append(other_move)
                current_dest = other_move.dest_path

        if len(chain) >= 2:
            chains.append(chain)

    # Find the longest chain
    if chains:
        longest_chain = max(chains, key=len)
        longest_chain.sort(key=lambda e: e.timestamp)

        final_path = longest_chain[-1].dest_path or longest_chain[-1].path
        return FileOperation(
            operation_type=OperationType.RENAME_SEQUENCE,
            primary_path=final_path,
            events=longest_chain,
            confidence=0.90,
            description=f"Rename sequence of {len(longest_chain)} moves",
            start_time=longest_chain[0].timestamp,
            end_time=longest_chain[-1].timestamp,
            is_atomic=True,
            is_safe=True,
            files_affected=[final_path],
            metadata={
                "original_path": str(longest_chain[0].path),
                "chain_length": len(longest_chain),
                "pattern": "rename_sequence",
            },
        )

    return None

Functions