0Pricing
AI Agents · 课时

目录遍历与文件发现

使用 os.walk()、glob 模式，并按类型或日期筛选文件。

目录遍历与文件发现 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

智能体为何需要遍历目录

智能体经常需要查找符合特定条件的文件，例如代码库中的所有 Python 文件、数据目录中的所有 CSV 文件，或所有最近的日志。Python 提供了三种主要工具：os.walk()、Path.iterdir() 和 Path.rglob()。根据所需的搜索深度不同，每种工具都有各自的优势。

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() — 递归遍历目录

os.walk(root) 会为目录树中的每个目录生成一个由 (dirpath, dirnames, filenames) 组成的元组。它是 Python 中经典的递归遍历方式，并且可以让您细致地控制要进入哪些子目录。

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() — 列出单层目录内容

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() — 递归通配匹配

Path.rglob(pattern) 会递归查找所有符合 glob 模式的文件。在整个目录树中的任意位置查找某种类型的所有文件时，这是最简洁的方式。glob 中的 ** 表示“任意数量的目录”。

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() — 模式匹配

glob.glob(pattern, recursive=True) 是传统的 glob 接口。使用 recursive=True 时，** 通配符可以匹配任意子目录路径。它返回字符串，而 pathlib.rglob 返回路径对象。

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')

按文件扩展名筛选

遍历目录时，可以使用 Path.suffix 按文件扩展名筛选文件（该属性包含点号，例如 '.py'），也可以将扩展名与允许的扩展名集合进行比对。对于多种扩展名的筛选，这种方式比 glob 模式更快、更精确。

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')

按文件大小筛选

智能体有时需要查找大于或小于某个大小阈值的文件，例如跳过很小的空文件，或避免处理会耗尽内存的大型文件。使用 Path.stat().st_size 可以获取文件的字节大小。

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)}')

按修改时间筛选

增量流水线中的智能体只需要处理自上次运行以来发生变化的文件。使用 stat().st_mtime（以 Unix 时间戳表示的修改时间）可以查找最近修改的文件。

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")}')

跳过隐藏文件和系统目录

遍历代码仓库或用户目录时，请跳过隐藏文件（以 . 开头）以及 .git、__pycache__、node_modules 和 .venv 等系统目录。这些位置包含数千个与大多数智能体无关的文件。

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')

构建文件清单

文件清单是对目录树的结构化概述，对于需要在处理前规划工作的智能体很有用。请将名称、路径、大小、扩展名和修改时间收集到字典列表中，以便序列化为 JSON。

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')

监视目录变化

有些智能体需要对目录中出现的新文件做出响应。虽然在生产环境中，像 watchdog 这样的完整文件监视库更为合适，但一种简单的方式是定期扫描目录，并与已知文件集合进行比较。

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

检验您对目录遍历方法的理解。

目录遍历回顾

现在，您已经能够查找智能体所需的任何文件：

  • os.walk() — 完全控制递归遍历；可以原地修改 dirnames，以剪除子目录
  • Path.iterdir() — 列出直接子项的单层目录内容
  • Path.rglob('*.ext') — 按模式进行递归搜索的简洁方式
  • glob.glob('**/*.ext', recursive=True) — 返回字符串的传统替代方式
  • 使用 .suffix 按扩展名筛选，使用 .stat().st_size 按大小筛选，使用 .stat().st_mtime 按修改时间筛选
  • 始终跳过隐藏文件和 .git、node_modules 等系统目录

常见问题解答

「目录遍历与文件发现」课时是免费的吗？

是的 — 「目录遍历与文件发现」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「目录遍历与文件发现」这节课中我会学到什么？

使用 os.walk()、glob 模式，并按类型或日期筛选文件。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「目录遍历与文件发现」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 在代理环境中读写文件
  2. 目录遍历与文件发现
  3. 文件格式处理：CSV、JSON 与 TXT
  4. 带错误处理的安全文件操作
← 返回 AI Agents