اجتياز المجلدات واكتشاف الملفات

لماذا يحتاج الوكلاء إلى اجتياز المجلدات

import os
from pathlib import Path

# Count all files in a directory tree
root = Path('/tmp/agent_workspace')
file_count = sum(1 for f in root.rglob('*') if f.is_file())
print(f'Found {file_count} files under {root}')

os.walk() — الاجتياز التكراري للمجلدات

import os
import tempfile

# --- demo setup: build a small project tree so the walk has something to show ---
root = tempfile.mkdtemp(prefix='project_')
os.makedirs(os.path.join(root, 'src'), exist_ok=True)
os.makedirs(os.path.join(root, '.git'), exist_ok=True)
with open(os.path.join(root, 'README.md'), 'w') as f:
    f.write('demo')
with open(os.path.join(root, 'src', 'main.py'), 'w') as f:
    f.write('print(1)')

for dirpath, dirnames, filenames in os.walk(root):
    # Skip hidden directories (like .git)
    dirnames[:] = [d for d in dirnames if not d.startswith('.')]

    print(f'In directory: {dirpath}')
    print(f'  Subdirs: {dirnames}')
    print(f'  Files: {filenames}')

    for filename in filenames:
        full_path = os.path.join(dirpath, filename)
        print(f'  File: {full_path}')

Path.iterdir() — سرد مستوى واحد

from pathlib import Path

data_dir = Path('data')
data_dir.mkdir(exist_ok=True)
(data_dir / 'a.csv').write_text('x')
(data_dir / 'sub').mkdir(exist_ok=True)

for item in data_dir.iterdir():
    if item.is_file():
        print(f'FILE: {item.name} ({item.stat().st_size} bytes)')
    elif item.is_dir():
        print(f'DIR:  {item.name}/')

csv_files = [f for f in data_dir.iterdir()
             if f.is_file() and f.suffix == '.csv']
print(f'Found {len(csv_files)} CSV files')

Path.rglob() — البحث التكراري بالنمط

from pathlib import Path

project = Path('/tmp/my_project')

# Find all Python files recursively
python_files = list(project.rglob('*.py'))
print(f'Found {len(python_files)} Python files:')
for f in python_files:
    print(f'  {f.relative_to(project)}')

# Find all JSON files
json_files = list(project.rglob('*.json'))

# Find files matching a name pattern
test_files = list(project.rglob('test_*.py'))
print(f'Found {len(test_files)} test files')

glob.glob() — المطابقة بالنمط

import glob

# Find all CSV files recursively
csv_files = glob.glob('data/**/*.csv', recursive=True)
print(f'Found {len(csv_files)} CSV files')
for f in csv_files:
    print(f'  {f}')

# Find all log files in a specific directory (non-recursive)
logs = glob.glob('/var/log/*.log')

# Find files matching multiple criteria
import fnmatch
all_files = glob.glob('/tmp/**/*', recursive=True)
config_files = [
    f for f in all_files
    if fnmatch.fnmatch(f, '*.yaml') or fnmatch.fnmatch(f, '*.yml')
]
print(f'Found {len(config_files)} YAML config files')

التصفية حسب امتداد الملف

from pathlib import Path

directory = Path('/tmp/mixed_files')

ALLOWED_EXTENSIONS = {'.py', '.js', '.ts', '.json', '.yaml'}

# Filter by allowed extensions
code_files = [
    f for f in directory.rglob('*')
    if f.is_file() and f.suffix.lower() in ALLOWED_EXTENSIONS
]

print(f'Found {len(code_files)} code/config files')

# Group files by extension
from collections import defaultdict
by_ext = defaultdict(list)
for f in directory.rglob('*'):
    if f.is_file():
        by_ext[f.suffix].append(f)

for ext, files in sorted(by_ext.items()):
    print(f'{ext}: {len(files)} files')

التصفية حسب حجم الملف

from pathlib import Path

data_dir = Path('/tmp/data')

MIN_SIZE = 1024        # 1 KB — skip empty/tiny files
MAX_SIZE = 50_000_000  # 50 MB — skip very large files

processable = []
skipped_small = []
skipped_large = []

for f in data_dir.rglob('*.json'):
    if not f.is_file():
        continue
    size = f.stat().st_size
    if size < MIN_SIZE:
        skipped_small.append(f.name)
    elif size > MAX_SIZE:
        skipped_large.append(f.name)
    else:
        processable.append(f)

print(f'Processable: {len(processable)}')
print(f'Too small: {len(skipped_small)}')
print(f'Too large: {len(skipped_large)}')

التصفية حسب وقت التعديل

from pathlib import Path
import time
import datetime

data_dir = Path('/tmp/data')

# Files modified in the last 24 hours
cutoff = time.time() - (24 * 60 * 60)

recent_files = [
    f for f in data_dir.rglob('*.log')
    if f.is_file() and f.stat().st_mtime > cutoff
]

print(f'Modified in last 24h: {len(recent_files)} files')

# Sort by modification time (newest first)
recent_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)

for f in recent_files[:5]:  # top 5 most recent
    mtime = datetime.datetime.fromtimestamp(f.stat().st_mtime)
    print(f'  {f.name}: {mtime.strftime("%Y-%m-%d %H:%M:%S")}')

تخطي الملفات المخفية ومجلدات النظام

from pathlib import Path

SKIP_DIRS = {'.git', '__pycache__', 'node_modules', '.venv',
             '.env', 'dist', 'build', '.mypy_cache'}

def iter_source_files(root, extensions):
    root = Path(root)
    for path in root.rglob('*'):
        # Skip if any parent is in SKIP_DIRS
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        # Skip hidden files
        if path.name.startswith('.'):
            continue
        if path.is_file() and path.suffix in extensions:
            yield path

python_files = list(iter_source_files('/tmp/project', {'.py'}))
print(f'Found {len(python_files)} Python source files')

إنشاء جرد للملفات

from pathlib import Path
import datetime
import json

def build_inventory(root_dir, pattern='**/*'):
    root = Path(root_dir)
    inventory = []

    for f in root.glob(pattern):
        if not f.is_file():
            continue
        stat = f.stat()
        inventory.append({
            'name': f.name,
            'path': str(f.relative_to(root)),
            'extension': f.suffix,
            'size_bytes': stat.st_size,
            'modified': datetime.datetime.fromtimestamp(
                stat.st_mtime
            ).isoformat()
        })

    return sorted(inventory, key=lambda x: x['path'])

inventory = build_inventory('/tmp/data')
with open('file_inventory.json', 'w') as out:
    json.dump(inventory, out, indent=2)
print(f'Inventoried {len(inventory)} files')

مراقبة مجلد بحثًا عن التغييرات

from pathlib import Path
import time

def watch_for_new_files(watch_dir, extension='.json', interval=5):
    watch_dir = Path(watch_dir)
    known_files = set(watch_dir.glob(f'*{extension}'))
    print(f'Watching {watch_dir} for new {extension} files...')

    while True:
        time.sleep(interval)
        current_files = set(watch_dir.glob(f'*{extension}'))
        new_files = current_files - known_files

        for f in new_files:
            print(f'New file detected: {f.name}')
            process_new_file(f)  # handle the new file

        known_files = current_files

def process_new_file(file_path):
    print(f'Processing: {file_path}')

# --- demo (one polling cycle, without the infinite while True loop) ---
import tempfile
demo_dir = Path(tempfile.mkdtemp())
known_files = set(demo_dir.glob('*.json'))
print(f'Watching {demo_dir} for new .json files...')

(demo_dir / 'result_a.json').write_text('{}', encoding='utf-8')
(demo_dir / 'result_b.json').write_text('{}', encoding='utf-8')

current_files = set(demo_dir.glob('*.json'))
new_files = current_files - known_files
for f in sorted(new_files, key=lambda p: p.name):
    print(f'New file detected: {f.name}')
    process_new_file(f)

اختبار سريع: rglob مقابل iterdir

مراجعة اجتياز المجلدات