saberbud/PocketFlow-Tutorial-Codebase-Knowledge
1
0
Fork 0
mirror of https://github.com/The-Pocket/PocketFlow-Tutorial-Codebase-Knowledge.git synced 2026-08-29 08:34:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

#feat: Report progress of the crawling #64

Browse Source
This commit is contained in:
Hari Haran
2025-04-30 04:27:22 +00:00
parent 7478f1d990
commit cd947bbdc6
2 changed files with 71 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/crawl_local_files.py
+59 -41
View File
@@ -1,7 +1,7 @@
import os
import fnmatch
def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, max_file_size=None, use_relative_paths=True):
def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, max_file_size=None, use_relative_paths=True, progress_callback=None):
"""
Crawl files in a local directory with similar interface as crawl_github_files.
@@ -11,6 +11,7 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
exclude_patterns (set): File patterns to exclude (e.g. {"tests/*"})
max_file_size (int): Maximum file size in bytes
use_relative_paths (bool): Whether to use paths relative to directory
progress_callback (callable): Function to report progress, takes (processed, total) as arguments
Returns:
dict: {"files": {filepath: content}}
@@ -19,49 +20,66 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
raise ValueError(f"Directory does not exist: {directory}")
files_dict = {}
all_files = []
# Collect all files first to calculate total
for root, _, files in os.walk(directory):
for filename in files:
filepath = os.path.join(root, filename)
# Get path relative to directory if requested
if use_relative_paths:
relpath = os.path.relpath(filepath, directory)
else:
relpath = filepath
# Check if file matches any include pattern
included = False
if include_patterns:
for pattern in include_patterns:
if fnmatch.fnmatch(relpath, pattern):
included = True
break
else:
included = True
# Check if file matches any exclude pattern
excluded = False
if exclude_patterns:
for pattern in exclude_patterns:
if fnmatch.fnmatch(relpath, pattern):
excluded = True
break
if not included or excluded:
continue
# Check file size
if max_file_size and os.path.getsize(filepath) > max_file_size:
continue
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
files_dict[relpath] = content
except Exception as e:
print(f"Warning: Could not read file {filepath}: {e}")
all_files.append(filepath)
total_files = len(all_files)
processed_files = 0
for filepath in all_files:
# Get path relative to directory if requested
if use_relative_paths:
relpath = os.path.relpath(filepath, directory)
else:
relpath = filepath
# Check if file matches any include pattern
included = False
if include_patterns:
for pattern in include_patterns:
if fnmatch.fnmatch(relpath, pattern):
included = True
break
else:
included = True
# Check if file matches any exclude pattern
excluded = False
if exclude_patterns:
for pattern in exclude_patterns:
if fnmatch.fnmatch(relpath, pattern):
excluded = True
break
if not included or excluded:
processed_files += 1
if progress_callback:
progress_callback(processed_files, total_files)
continue
# Check file size
if max_file_size and os.path.getsize(filepath) > max_file_size:
processed_files += 1
if progress_callback:
progress_callback(processed_files, total_files)
continue
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
files_dict[relpath] = content
except Exception as e:
print(f"Warning: Could not read file {filepath}: {e}")
processed_files += 1
if progress_callback:
progress_callback(processed_files, total_files)
return {"files": files_dict}
if __name__ == "__main__":