Python实现目录遍历和内容获取的完整指南

import os

def list_directory_basic(directory_path):
    """
    基础目录列表函数
    """
    try:
        # 获取目录内容列表
        contents = os.listdir(directory_path)
        print(f"目录 '{directory_path}' 中的内容:")
        
        for item in contents:
            # 获取完整路径
            full_path = os.path.join(directory_path, item)
            
            # 判断是文件还是目录
            if os.path.isfile(full_path):
                print(f"  文件: {item}")
            elif os.path.isdir(full_path):
                print(f"  目录: {item}/")
            else:
                print(f"  其他: {item}")
                
        return contents
        
    except FileNotFoundError:
        print(f"错误: 目录 '{directory_path}' 不存在")
        return []
    except PermissionError:
        print(f"错误: 没有权限访问目录 '{directory_path}'")
        return []
    except NotADirectoryError:
        print(f"错误: '{directory_path}' 不是目录")
        return []

# 使用示例
directory_path = "/path/to/directory"
contents = list_directory_basic(directory_path)

import os

def categorize_directory_contents(directory_path):
    """
    分类返回目录内容
    """
    try:
        items = os.listdir(directory_path)
        
        # 使用列表推导式进行分类
        files = [item for item in items 
                if os.path.isfile(os.path.join(directory_path, item))]
        
        directories = [item for item in items 
                      if os.path.isdir(os.path.join(directory_path, item))]
        
        other_items = [item for item in items 
                      if not os.path.isfile(os.path.join(directory_path, item)) and 
                         not os.path.isdir(os.path.join(directory_path, item))]
        
        return {
            'files': files,
            'directories': directories,
            'other': other_items
        }
        
    except OSError as e:
        print(f"访问目录时出错: {e}")
        return {'files': [], 'directories': [], 'other': []}

# 使用示例
contents = categorize_directory_contents("/tmp")
print("文件:", contents['files'])
print("目录:", contents['directories'])
print("其他:", contents['other'])

import os

def recursive_directory_walk(root_dir):
    """
    递归遍历目录树
    """
    file_count = 0
    dir_count = 0
    
    print(f"开始遍历: {root_dir}")
    print("-" * 50)
    
    for root, dirs, files in os.walk(root_dir):
        # 当前目录级别
        level = root.replace(root_dir, '').count(os.sep)
        indent = '  ' * level
        
        print(f"{indent}{os.path.basename(root)}/")
        
        # 处理子目录
        sub_indent = '  ' * (level + 1)
        for directory in dirs:
            print(f"{sub_indent}{directory}/")
            dir_count += 1
        
        # 处理文件
        for file in files:
            file_path = os.path.join(root, file)
            file_size = os.path.getsize(file_path)
            print(f"{sub_indent}{file} ({file_size} bytes)")
            file_count += 1
    
    print("-" * 50)
    print(f"遍历完成: 找到 {dir_count} 个目录, {file_count} 个文件")

# 使用示例
recursive_directory_walk("/path/to/directory")

import os
from datetime import datetime

def scan_directory_efficient(directory_path):
    """
    使用os.scandir()高效扫描目录
    """
    try:
        with os.scandir(directory_path) as entries:
            print(f"目录 '{directory_path}' 内容:")
            print("-" * 60)
            
            for entry in entries:
                # 获取文件信息
                if entry.is_file():
                    file_type = "文件"
                    size = entry.stat().st_size
                    size_str = f"{size} bytes"
                elif entry.is_dir():
                    file_type = "目录"
                    size_str = "<DIR>"
                else:
                    file_type = "其他"
                    size_str = "N/A"
                
                # 获取修改时间
                mod_time = datetime.fromtimestamp(entry.stat().st_mtime)
                mod_time_str = mod_time.strftime("%Y-%m-%d %H:%M:%S")
                
                print(f"{file_type:8} {entry.name:30} {size_str:12} {mod_time_str}")
                
    except FileNotFoundError:
        print(f"错误: 目录不存在")
    except PermissionError:
        print(f"错误: 没有访问权限")

# 使用示例
scan_directory_efficient("/path/to/directory")

from pathlib import Path

def list_directory_pathlib(directory_path):
    """
    使用pathlib列出目录内容
    """
    path = Path(directory_path)
    
    if not path.exists():
        print(f"错误: 路径 '{directory_path}' 不存在")
        return
    
    if not path.is_dir():
        print(f"错误: '{directory_path}' 不是目录")
        return
    
    print(f"目录 '{directory_path}' 内容:")
    print("-" * 50)
    
    # 获取所有条目
    for item in path.iterdir():
        if item.is_file():
            file_type = "文件"
            size = item.stat().st_size
            info = f"{size} bytes"
        elif item.is_dir():
            file_type = "目录"
            info = "包含项目"
        else:
            file_type = "其他"
            info = "特殊文件"
        
        print(f"{file_type:6} {item.name:25} {info}")

# 使用示例
list_directory_pathlib("/path/to/directory")

import os
from pathlib import Path
from datetime import datetime, timedelta

class AdvancedDirectoryFilter:
    """
    高级目录内容过滤器
    """
    
    @staticmethod
    def filter_files(directory, **filters):
        """
        根据多种条件过滤文件
        """
        path = Path(directory)
        if not path.exists() or not path.is_dir():
            return []
        
        results = []
        for item in path.iterdir():
            if not item.is_file():
                continue
            
            stat = item.stat()
            matches = True
            
            # 应用过滤器
            if 'min_size' in filters and stat.st_size < filters['min_size']:
                matches = False
            if 'max_size' in filters and stat.st_size > filters['max_size']:
                matches = False
            if 'extension' in filters and not item.suffix.lower() == filters['extension'].lower():
                matches = False
            if 'min_age_days' in filters:
                file_time = datetime.fromtimestamp(stat.st_mtime)
                if file_time > datetime.now() - timedelta(days=filters['min_age_days']):
                    matches = False
            
            if matches:
                results.append(item)
        
        return results
    
    @staticmethod
    def find_files_by_pattern(directory, pattern, recursive=False):
        """
        使用通配符模式查找文件
        """
        path = Path(directory)
        
        if recursive:
            return list(path.rglob(pattern))
        else:
            return list(path.glob(pattern))

# 使用示例
filterer = AdvancedDirectoryFilter()

# 查找所有.log文件
log_files = filterer.find_files_by_pattern("/var/log", "*.log", recursive=True)

# 查找大于1MB的文件
large_files = filterer.filter_files("/tmp", min_size=1024 * 1024)

# 查找7天前的.txt文件
old_text_files = filterer.filter_files("/backup", extension=".txt", min_age_days=7)

import os
import fnmatch
from pathlib import Path

def find_files_with_pattern(directory, patterns, recursive=False):
    """
    使用fnmatch模式匹配查找文件
    """
    path = Path(directory)
    results = []
    
    if recursive:
        search_method = path.rglob
    else:
        search_method = path.glob
    
    for pattern in patterns:
        # 先使用简单的通配符匹配
        for item in search_method('*'):
            if fnmatch.fnmatch(item.name, pattern):
                results.append(item)
    
    return results

# 使用示例
patterns = ['*.log', '*.txt', 'test_*.py']
matching_files = find_files_with_pattern("/var/log", patterns, recursive=True)

print("找到的匹配文件:")
for file in matching_files:
    print(f"  {file}")

from pathlib import Path
import os

class DirectorySizeCalculator:
    """
    目录大小计算器，支持多种统计方式
    """
    
    @staticmethod
    def get_size(path):
        """
        获取文件或目录的大小（字节）
        """
        path = Path(path)
        if path.is_file():
            return path.stat().st_size
        elif path.is_dir():
            total_size = 0
            for item in path.rglob('*'):
                if item.is_file():
                    total_size += item.stat().st_size
            return total_size
        return 0
    
    @staticmethod
    def get_size_readable(path):
        """
        获取可读格式的大小
        """
        size_bytes = DirectorySizeCalculator.get_size(path)
        
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if size_bytes < 1024.0:
                return f"{size_bytes:.2f} {unit}"
            size_bytes /= 1024.0
        
        return f"{size_bytes:.2f} PB"
    
    @staticmethod
    def analyze_directory_structure(directory):
        """
        分析目录结构，返回大小统计
        """
        path = Path(directory)
        if not path.exists() or not path.is_dir():
            return {}
        
        size_info = {
            'total_size': 0,
            'file_count': 0,
            'dir_count': 0,
            'by_extension': {},
            'largest_files': []
        }
        
        # 临时存储文件信息用于排序
        file_sizes = []
        
        for item in path.rglob('*'):
            if item.is_file():
                file_size = item.stat().st_size
                size_info['total_size'] += file_size
                size_info['file_count'] += 1
                
                # 按扩展名统计
                ext = item.suffix.lower()
                size_info['by_extension'][ext] = size_info['by_extension'].get(ext, 0) + file_size
                
                # 记录文件信息用于查找最大文件
                file_sizes.append((item, file_size))
                
            elif item.is_dir():
                size_info['dir_count'] += 1
        
        # 找出最大的10个文件
        file_sizes.sort(key=lambda x: x[1], reverse=True)
        size_info['largest_files'] = file_sizes[:10]
        
        return size_info

# 使用示例
calculator = DirectorySizeCalculator()
size_info = calculator.analyze_directory_structure("/var/log")

print(f"总大小: {calculator.get_size_readable('/var/log')}")
print(f"文件数: {size_info['file_count']}")
print(f"目录数: {size_info['dir_count']}")

print("\n按扩展名统计:")
for ext, size in size_info['by_extension'].items():
    readable_size = calculator.get_size_readable(Path(f"{ext}_{size}"))
    print(f"  {ext or '无扩展名'}: {readable_size}")

print("\n最大的文件:")
for file, size in size_info['largest_files']:
    print(f"  {file}: {calculator.get_size_readable(Path(f'temp_{size}'))}")

from pathlib import Path
import hashlib
import os

class FileSyncUtility:
    """
    文件同步工具基础类
    """
    
    @staticmethod
    def get_file_hash(file_path, block_size=65536):
        """
        计算文件的MD5哈希值
        """
        path = Path(file_path)
        if not path.is_file():
            return None
        
        hash_md5 = hashlib.md5()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(block_size), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    
    @staticmethod
    def compare_directories(source_dir, target_dir):
        """
        比较两个目录的内容
        """
        source = Path(source_dir)
        target = Path(target_dir)
        
        if not source.exists() or not source.is_dir():
            raise ValueError("源目录不存在或不是目录")
        
        if not target.exists():
            target.mkdir(parents=True)
        
        comparison = {
            'source_only': [],
            'target_only': [],
            'different': [],
            'same': []
        }
        
        # 构建源文件索引
        source_files = {}
        for item in source.rglob('*'):
            if item.is_file():
                rel_path = item.relative_to(source)
                source_files[rel_path] = item
        
        # 构建目标文件索引
        target_files = {}
        for item in target.rglob('*'):
            if item.is_file():
                rel_path = item.relative_to(target)
                target_files[rel_path] = item
        
        # 比较文件
        all_files = set(source_files.keys()) | set(target_files.keys())
        
        for rel_path in all_files:
            source_file = source_files.get(rel_path)
            target_file = target_files.get(rel_path)
            
            if source_file and not target_file:
                comparison['source_only'].append(rel_path)
            elif target_file and not source_file:
                comparison['target_only'].append(rel_path)
            else:
                # 两个目录都有该文件，比较哈希值
                source_hash = FileSyncUtility.get_file_hash(source_file)
                target_hash = FileSyncUtility.get_file_hash(target_file)
                
                if source_hash == target_hash:
                    comparison['same'].append(rel_path)
                else:
                    comparison['different'].append(rel_path)
        
        return comparison

# 使用示例
sync_util = FileSyncUtility()
comparison = sync_util.compare_directories("/path/source", "/path/target")

print("只在源目录存在的文件:", comparison['source_only'])
print("只在目标目录存在的文件:", comparison['target_only'])
print("内容不同的文件:", comparison['different'])
print("内容相同的文件:", comparison['same'])

from pathlib import Path
from datetime import datetime
import re

class LogFileAnalyzer:
    """
    日志文件分析器
    """
    
    def __init__(self, log_directory):
        self.log_dir = Path(log_directory)
        self.log_patterns = {
            'error': re.compile(r'ERROR|CRITICAL|FAILED', re.IGNORECASE),
            'warning': re.compile(r'WARNING|ALERT', re.IGNORECASE),
            'info': re.compile(r'INFO|DEBUG', re.IGNORECASE)
        }
    
    def find_latest_log_file(self):
        """
        查找最新的日志文件
        """
        log_files = list(self.log_dir.glob('*.log'))
        if not log_files:
            return None
        
        # 按修改时间排序
        log_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
        return log_files[0]
    
    def analyze_log_file(self, log_file_path, max_entries=100):
        """
        分析日志文件内容
        """
        path = Path(log_file_path)
        if not path.exists() or not path.is_file():
            return None
        
        analysis = {
            'error_count': 0,
            'warning_count': 0,
            'info_count': 0,
            'errors': [],
            'warnings': [],
            'file_size': path.stat().st_size,
            'last_modified': datetime.fromtimestamp(path.stat().st_mtime)
        }
        
        try:
            with open(log_file_path, 'r', encoding='utf-8', errors='ignore') as f:
                for line in f:
                    if self.log_patterns['error'].search(line):
                        analysis['error_count'] += 1
                        if len(analysis['errors']) < max_entries:
                            analysis['errors'].append(line.strip())
                    elif self.log_patterns['warning'].search(line):
                        analysis['warning_count'] += 1
                        if len(analysis['warnings']) < max_entries:
                            analysis['warnings'].append(line.strip())
                    elif self.log_patterns['info'].search(line):
                        analysis['info_count'] += 1
        
        except IOError as e:
            print(f"读取日志文件时出错: {e}")
            return None
        
        return analysis
    
    def generate_log_report(self):
        """
        生成日志分析报告
        """
        latest_log = self.find_latest_log_file()
        if not latest_log:
            return "没有找到日志文件"
        
        analysis = self.analyze_log_file(latest_log)
        if not analysis:
            return "分析日志文件时出错"
        
        report = [
            f"日志文件分析报告",
            f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            f"分析文件: {latest_log.name}",
            f"文件大小: {analysis['file_size']} bytes",
            f"最后修改: {analysis['last_modified'].strftime('%Y-%m-%d %H:%M:%S')}",
            "",
            "统计信息:",
            f"错误数量: {analysis['error_count']}",
            f"警告数量: {analysis['warning_count']}",
            f"信息数量: {analysis['info_count']}",
            ""
        ]
        
        if analysis['errors']:
            report.extend(["最近错误:", ""] + analysis['errors'][:10])
        
        if analysis['warnings']:
            report.extend(["", "最近警告:", ""] + analysis['warnings'][:10])
        
        return "\n".join(report)

# 使用示例
analyzer = LogFileAnalyzer("/var/log")
report = analyzer.generate_log_report()
print(report)

from pathlib import Path
import os
import time

class EfficientDirectoryLister:
    """
    高效目录列表工具
    """
    
    @staticmethod
    def batch_process_directory(directory_path, batch_size=1000, processor_func=None):
        """
        批量处理大型目录
        """
        path = Path(directory_path)
        if not path.exists() or not path.is_dir():
            return
        
        all_items = []
        batch_count = 0
        
        with os.scandir(directory_path) as entries:
            for entry in entries:
                all_items.append(entry)
                
                # 达到批次大小时处理
                if len(all_items) >= batch_size:
                    if processor_func:
                        processor_func(all_items, batch_count)
                    all_items = []
                    batch_count += 1
        
        # 处理剩余项目
        if all_items and processor_func:
            processor_func(all_items, batch_count)
    
    @staticmethod
    def parallel_directory_scan(directory_path, max_workers=None):
        """
        并行目录扫描（适用于超大目录）
        """
        import concurrent.futures
        
        path = Path(directory_path)
        if not path.exists() or not path.is_dir():
            return []
        
        def process_subdirectory(subdir_path):
            """处理单个子目录"""
            subdir_items = []
            try:
                with os.scandir(subdir_path) as entries:
                    for entry in entries:
                        subdir_items.append(entry)
            except OSError:
                pass
            return subdir_items
        
        # 获取所有子目录
        subdirectories = [str(item.path) for item in os.scandir(directory_path) 
                         if item.is_dir()]
        
        # 并行处理
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            results = list(executor.map(process_subdirectory, subdirectories))
        
        # 合并结果
        all_items = []
        for result in results:
            all_items.extend(result)
        
        return all_items

# 使用示例
def example_processor(items, batch_number):
    """示例处理函数"""
    print(f"处理批次 {batch_number}: {len(items)} 个项目")
    for item in items:
        if item.is_file():
            print(f"  文件: {item.name}")

efficient_lister = EfficientDirectoryLister()
efficient_lister.batch_process_directory("/large/directory", processor_func=example_processor)

from pathlib import Path
import os

class MemoryEfficientDirectoryWalker:
    """
    内存友好的大型目录遍历器
    """
    
    def __init__(self):
        self.file_count = 0
        self.dir_count = 0
        self.total_size = 0
    
    def walk_with_callback(self, root_dir, file_callback=None, dir_callback=None):
        """
        使用回调函数遍历目录，减少内存使用
        """
        path = Path(root_dir)
        
        for root, dirs, files in os.walk(root_dir):
            self.dir_count += 1
            
            if dir_callback:
                dir_callback(root, dirs)
            
            for file in files:
                self.file_count += 1
                file_path = os.path.join(root, file)
                
                try:
                    file_size = os.path.getsize(file_path)
                    self.total_size += file_size
                    
                    if file_callback:
                        file_callback(file_path, file_size)
                        
                except OSError:
                    continue
    
    def get_stats(self):
        """获取统计信息"""
        return {
            'directories': self.dir_count,
            'files': self.file_count,
            'total_size': self.total_size
        }

# 使用示例
def file_callback(file_path, file_size):
    """文件处理回调"""
    if file_size > 1024 * 1024:  # 大于1MB的文件
        print(f"大文件: {file_path} ({file_size} bytes)")

def dir_callback(dir_path, subdirs):
    """目录处理回调"""
    if len(subdirs) > 10:  # 子目录很多的目录
        print(f"多子目录: {dir_path} ({len(subdirs)} 个子目录)")

walker = MemoryEfficientDirectoryWalker()
walker.walk_with_callback("/large/directory", file_callback, dir_callback)

stats = walker.get_stats()
print(f"遍历完成: {stats['directories']} 目录, {stats['files']} 文件, 总大小: {stats['total_size']} bytes")

from pathlib import Path
import os
import logging
import stat

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RobustDirectoryLister:
    """
    健壮的目录遍历框架，包含完整的错误处理
    """
    
    def __init__(self, max_retries=3, timeout=30):
        self.max_retries = max_retries
        self.timeout = timeout
    
    def safe_list_directory(self, directory_path):
        """
        安全地列出目录内容
        """
        path = Path(directory_path)
        
        # 前置检查
        if not self._pre_check(path):
            return []
        
        attempts = 0
        while attempts < self.max_retries:
            try:
                contents = []
                with os.scandir(directory_path) as entries:
                    for entry in entries:
                        contents.append({
                            'name': entry.name,
                            'is_file': entry.is_file(),
                            'is_dir': entry.is_dir(),
                            'path': entry.path
                        })
                return contents
                
            except PermissionError as e:
                logger.warning(f"权限错误 (尝试 {attempts + 1}): {e}")
                attempts += 1
                if attempts >= self.max_retries:
                    logger.error("达到最大重试次数")
                    return []
                
            except OSError as e:
                logger.error(f"系统错误: {e}")
                return []
        
        return []
    
    def _pre_check(self, path):
        """
        前置安全检查
        """
        try:
            # 检查路径是否存在
            if not path.exists():
                logger.error("路径不存在")
                return False
            
            # 检查是否是目录
            if not path.is_dir():
                logger.error("路径不是目录")
                return False
            
            # 检查读取权限
            if not os.access(str(path), os.R_OK):
                logger.error("没有读取权限")
                return False
            
            # 检查是否符号链接（可能需要特殊处理）
            if path.is_symlink():
                logger.warning("路径是符号链接，正在解析真实路径")
                real_path = path.resolve()
                return self._pre_check(real_path)
            
            return True
            
        except Exception as e:
            logger.error(f"前置检查失败: {e}")
            return False

# 使用示例
robust_lister = RobustDirectoryLister()
contents = robust_lister.safe_list_directory("/secure/directory")

for item in contents:
    print(f"{'文件' if item['is_file'] else '目录'}: {item['name']}")