Python实现检测文件是否存在的方法完整指南

import os
import os.path

def check_file_existence_basic(file_path):
    """
    基础文件存在性检测
    """
    # 检查文件是否存在
    if os.path.exists(file_path):
        print(f"文件 {file_path} 存在")
        
        # 进一步检查是否是文件（不是目录）
        if os.path.isfile(file_path):
            print("这是一个文件")
            return true
        else:
            print("这不是一个普通文件（可能是目录）")
            return false
    else:
        print(f"文件 {file_path} 不存在")
        return false

# 使用示例
file_path = "example.txt"
result = check_file_existence_basic(file_path)

import os
import stat

def analyze_path_type(path):
    """
    详细分析路径类型
    """
    if not os.path.exists(path):
        print(f"路径 {path} 不存在")
        return false
    
    # 检查文件类型
    if os.path.isfile(path):
        print(f"{path} 是普通文件")
        
    if os.path.isdir(path):
        print(f"{path} 是目录")
        
    if os.path.islink(path):
        print(f"{path} 是符号链接")
        # 解析符号链接指向的实际路径
        real_path = os.path.realpath(path)
        print(f"指向: {real_path}")
        
    # 检查文件权限
    if os.access(path, os.r_ok):
        print("当前用户有读取权限")
    if os.access(path, os.w_ok):
        print("当前用户有写入权限")
    if os.access(path, os.x_ok):
        print("当前用户有执行权限")
    
    return true

# 使用示例
analyze_path_type("/etc/passwd")

import os

def check_file_eafp(file_path):
    """
    eafp风格：直接尝试操作，捕获异常
    """
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        print("文件存在且可读取")
        return true, content
    except filenotfounderror:
        print("文件不存在")
        return false, none
    except permissionerror:
        print("没有读取权限")
        return false, none
    except ioerror as e:
        print(f"读取文件时发生错误: {e}")
        return false, none

def check_file_lbyl(file_path):
    """
    lbyl风格：先检查再操作
    """
    if not os.path.exists(file_path):
        print("文件不存在")
        return false, none
        
    if not os.path.isfile(file_path):
        print("路径不是文件")
        return false, none
        
    if not os.access(file_path, os.r_ok):
        print("没有读取权限")
        return false, none
        
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        return true, content
    except ioerror as e:
        print(f"读取文件时发生错误: {e}")
        return false, none

# 使用示例
file_path = "config.ini"

# eafp方式
exists_eafp, content_eafp = check_file_eafp(file_path)

# lbyl方式
exists_lbyl, content_lbyl = check_file_lbyl(file_path)

from pathlib import path

def check_file_pathlib(file_path):
    """
    使用pathlib检测文件存在性
    """
    path = path(file_path)
    
    if path.exists():
        print(f"路径 {path} 存在")
        
        if path.is_file():
            print("这是一个文件")
            # 获取文件信息
            print(f"文件大小: {path.stat().st_size} 字节")
            print(f"最后修改时间: {path.stat().st_mtime}")
            return true
        elif path.is_dir():
            print("这是一个目录")
            return false
        elif path.is_symlink():
            print("这是一个符号链接")
            return false
    else:
        print(f"路径 {path} 不存在")
        return false

# 使用示例
check_file_pathlib("example.txt")

import os
from threading import thread
import time

def competitive_check_demo():
    """
    演示竞争条件问题
    """
    test_file = "temp_file.txt"
    
    # 删除可能存在的旧文件
    if os.path.exists(test_file):
        os.remove(test_file)
    
    def create_file_later():
        """稍后创建文件"""
        time.sleep(0.1)  # 短暂延迟
        with open(test_file, 'w') as f:
            f.write("测试内容")
    
    # 启动线程创建文件
    thread = thread(target=create_file_later)
    thread.start()
    
    # 检查文件是否存在
    if os.path.exists(test_file):
        print("文件存在，尝试读取...")
        # 这里可能出现竞争条件：文件可能在被检查后立即被删除
        try:
            with open(test_file, 'r') as f:
                content = f.read()
            print("成功读取文件")
        except filenotfounderror:
            print("文件在检查后消失了！")
    else:
        print("文件不存在")
    
    thread.join()
    # 清理
    if os.path.exists(test_file):
        os.remove(test_file)

# 运行演示
competitive_check_demo()

import os
import tempfile

def safe_file_operation(file_path, operation_func, *args, **kwargs):
    """
    安全的文件操作函数，处理竞争条件
    """
    max_retries = 3
    retry_delay = 0.1  # 100毫秒
    
    for attempt in range(max_retries):
        try:
            # 尝试执行操作
            return operation_func(file_path, *args, **kwargs)
        except filenotfounderror:
            if attempt == max_retries - 1:
                raise
            time.sleep(retry_delay)
        except permissionerror:
            if attempt == max_retries - 1:
                raise
            time.sleep(retry_delay)
        except ioerror as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(retry_delay)

def read_file_safely(file_path):
    """安全读取文件"""
    with open(file_path, 'r') as f:
        return f.read()

# 使用示例
try:
    content = safe_file_operation("important_file.txt", read_file_safely)
    print("成功读取文件内容")
except filenotfounderror:
    print("文件不存在")
except exception as e:
    print(f"读取文件时发生错误: {e}")

from pathlib import path
import json
import yaml

class configmanager:
    """
    配置文件管理器，处理多种配置格式
    """
    
    def __init__(self, config_paths=none):
        self.config_paths = config_paths or [
            "./config.json",
            "./config.yaml",
            "./config.yml",
            "~/app_config.json",
            "/etc/app/config.json"
        ]
    
    def find_config_file(self):
        """
        查找存在的配置文件
        """
        for config_path in self.config_paths:
            path = path(config_path).expanduser()  # 处理~扩展
            
            if path.exists() and path.is_file() and os.access(path, os.r_ok):
                print(f"找到配置文件: {path}")
                return path
        
        print("未找到任何配置文件")
        return none
    
    def load_config(self):
        """
        加载配置文件
        """
        config_path = self.find_config_file()
        if not config_path:
            return none
        
        try:
            if config_path.suffix == '.json':
                with open(config_path, 'r') as f:
                    return json.load(f)
            elif config_path.suffix in ['.yaml', '.yml']:
                with open(config_path, 'r') as f:
                    return yaml.safe_load(f)
            else:
                print(f"不支持的配置文件格式: {config_path.suffix}")
                return none
        except exception as e:
            print(f"加载配置文件时发生错误: {e}")
            return none

# 使用示例
config_manager = configmanager()
config = config_manager.load_config()
if config:
    print("配置加载成功")
else:
    print("使用默认配置")

import os
import time
from datetime import datetime
from pathlib import path

class logmanager:
    """
    日志文件管理器，支持文件存在性检查和轮转
    """
    
    def __init__(self, log_dir="./logs", max_file_size=10 * 1024 * 1024):  # 10mb
        self.log_dir = path(log_dir)
        self.max_file_size = max_file_size
        self.current_log_file = none
        
        # 确保日志目录存在
        self.log_dir.mkdir(exist_ok=true)
    
    def get_current_log_file(self):
        """
        获取当前日志文件路径，必要时创建新文件
        """
        if self.current_log_file and self._should_rotate_log():
            self._rotate_log_file()
        
        if not self.current_log_file:
            self.current_log_file = self.log_dir / f"app_{datetime.now().strftime('%y%m%d_%h%m%s')}.log"
            
        return self.current_log_file
    
    def _should_rotate_log(self):
        """
        检查是否需要日志轮转
        """
        if not self.current_log_file.exists():
            return false
        
        try:
            return self.current_log_file.stat().st_size >= self.max_file_size
        except oserror:
            return true
    
    def _rotate_log_file(self):
        """
        执行日志轮转
        """
        if self.current_log_file and self.current_log_file.exists():
            # 重命名旧日志文件
            timestamp = datetime.now().strftime('%y%m%d_%h%m%s')
            archived_file = self.log_dir / f"app_{timestamp}.log"
            
            try:
                self.current_log_file.rename(archived_file)
                print(f"已轮转日志文件: {archived_file}")
            except oserror as e:
                print(f"日志轮转失败: {e}")
        
        self.current_log_file = none
    
    def write_log(self, message):
        """
        写入日志消息
        """
        log_file = self.get_current_log_file()
        
        try:
            with open(log_file, 'a') as f:
                timestamp = datetime.now().strftime('%y-%m-%d %h:%m:%s')
                f.write(f"[{timestamp}] {message}\n")
        except oserror as e:
            print(f"写入日志失败: {e}")

# 使用示例
log_manager = logmanager()
for i in range(1000):
    log_manager.write_log(f"测试日志消息 {i}")

from pathlib import path
import hashlib
import uuid
from typing import optional

class fileuploadhandler:
    """
    文件上传处理器，包含存在性检查和冲突解决
    """
    
    def __init__(self, upload_dir: str = "./uploads"):
        self.upload_dir = path(upload_dir)
        self.upload_dir.mkdir(exist_ok=true)
    
    def generate_unique_filename(self, original_filename: str) -> path:
        """
        生成唯一的文件名，避免冲突
        """
        extension = path(original_filename).suffix
        unique_id = uuid.uuid4().hex
        return self.upload_dir / f"{unique_id}{extension}"
    
    def file_exists_by_content(self, file_content: bytes) -> optional[path]:
        """
        通过内容哈希检查文件是否已存在
        """
        content_hash = hashlib.sha256(file_content).hexdigest()
        
        # 检查是否已有相同内容的文件
        for existing_file in self.upload_dir.iterdir():
            if existing_file.is_file():
                try:
                    with open(existing_file, 'rb') as f:
                        existing_content = f.read()
                    existing_hash = hashlib.sha256(existing_content).hexdigest()
                    
                    if existing_hash == content_hash:
                        return existing_file
                except ioerror:
                    continue
        
        return none
    
    def safe_upload_file(self, file_content: bytes, original_filename: str) -> path:
        """
        安全地上传文件，处理各种边界情况
        """
        # 检查是否已存在相同内容的文件
        existing_file = self.file_exists_by_content(file_content)
        if existing_file:
            print(f"文件已存在，内容与 {existing_file} 相同")
            return existing_file
        
        # 生成唯一文件名
        target_path = self.generate_unique_filename(original_filename)
        
        # 确保目标文件不存在（尽管概率极低）
        if target_path.exists():
            # 如果发生冲突，重新生成文件名
            target_path = self.generate_unique_filename(original_filename)
        
        # 写入文件
        try:
            with open(target_path, 'wb') as f:
                f.write(file_content)
            
            # 再次验证文件写入成功
            if not target_path.exists():
                raise ioerror("文件写入后验证失败")
            
            print(f"文件上传成功: {target_path}")
            return target_path
            
        except ioerror as e:
            print(f"文件上传失败: {e}")
            # 清理可能部分写入的文件
            if target_path.exists():
                try:
                    target_path.unlink()
                except ioerror:
                    pass
            raise

# 使用示例
upload_handler = fileuploadhandler()

# 模拟文件上传
test_content = b"this is test file content"
try:
    saved_path = upload_handler.safe_upload_file(test_content, "test.txt")
    print(f"文件保存路径: {saved_path}")
except exception as e:
    print(f"上传失败: {e}")

import os
from pathlib import path
from concurrent.futures import threadpoolexecutor
import time

class batchfilechecker:
    """
    批量文件检查器，优化大量文件存在性检查的性能
    """
    
    def __init__(self, max_workers=none):
        self.max_workers = max_workers or os.cpu_count() or 4
    
    def check_files_sequential(self, file_paths):
        """
        顺序检查文件列表
        """
        results = {}
        for file_path in file_paths:
            path = path(file_path)
            results[file_path] = path.exists()
        return results
    
    def check_files_parallel(self, file_paths):
        """
        并行检查文件列表
        """
        results = {}
        
        def check_file(path):
            return path, path(path).exists()
        
        with threadpoolexecutor(max_workers=self.max_workers) as executor:
            future_results = executor.map(check_file, file_paths)
            for path, exists in future_results:
                results[path] = exists
        
        return results
    
    def benchmark_checkers(self, file_paths):
        """
        性能基准测试
        """
        # 顺序检查
        start_time = time.time()
        seq_results = self.check_files_sequential(file_paths)
        seq_duration = time.time() - start_time
        
        # 并行检查
        start_time = time.time()
        par_results = self.check_files_parallel(file_paths)
        par_duration = time.time() - start_time
        
        print(f"顺序检查耗时: {seq_duration:.4f}秒")
        print(f"并行检查耗时: {par_duration:.4f}秒")
        print(f"性能提升: {seq_duration/par_duration:.2f}倍")
        
        # 验证结果一致性
        assert seq_results == par_results, "检查结果不一致"
        
        return seq_results

# 使用示例
# 生成测试文件列表
test_files = [f"test_file_{i}.txt" for i in range(100)]

# 创建一些测试文件
for i in range(10):  # 创建前10个文件
    with open(f"test_file_{i}.txt", 'w') as f:
        f.write("test")

checker = batchfilechecker()
results = checker.benchmark_checkers(test_files)

# 清理测试文件
for i in range(10):
    try:
        os.remove(f"test_file_{i}.txt")
    except oserror:
        pass

from pathlib import path
import time
from functools import lru_cache

class cachedfilechecker:
    """
    带缓存的文件检查器，减少重复检查的开销
    """
    
    def __init__(self, cache_ttl=5):  # 默认缓存5秒
        self.cache_ttl = cache_ttl
        self.cache = {}
    
    @lru_cache(maxsize=1024)
    def check_file_cached(self, file_path):
        """
        带缓存的文件检查
        """
        current_time = time.time()
        
        # 检查缓存是否有效
        if file_path in self.cache:
            cached_time, exists = self.cache[file_path]
            if current_time - cached_time < self.cache_ttl:
                return exists
        
        # 实际检查文件
        exists = path(file_path).exists()
        self.cache[file_path] = (current_time, exists)
        
        return exists
    
    def clear_cache(self):
        """清空缓存"""
        self.cache.clear()
        self.check_file_cached.cache_clear()

# 使用示例
cached_checker = cachedfilechecker(cache_ttl=2)  # 2秒缓存

# 第一次检查
print("第一次检查:", cached_checker.check_file_cached("example.txt"))

# 立即再次检查（使用缓存）
print("第二次检查:", cached_checker.check_file_cached("example.txt"))

# 等待缓存过期
time.sleep(3)
print("缓存过期后检查:", cached_checker.check_file_cached("example.txt"))

import os
import logging
from pathlib import path
from typing import optional, dict, any

# 配置日志
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getlogger(__name__)

class robustfilechecker:
    """
    健壮的文件检查器，包含完整的错误处理和日志记录
    """
    
    def __init__(self, retry_attempts: int = 3, retry_delay: float = 0.1):
        self.retry_attempts = retry_attempts
        self.retry_delay = retry_delay
    
    def safe_check_exists(self, file_path: str) -> optional[bool]:
        """
        安全地检查文件是否存在，包含重试机制
        """
        for attempt in range(self.retry_attempts):
            try:
                exists = path(file_path).exists()
                logger.debug(f"文件检查成功: {file_path} -> {exists}")
                return exists
                
            except oserror as e:
                logger.warning(
                    f"文件检查失败 (尝试 {attempt + 1}/{self.retry_attempts}): "
                    f"{file_path} - {e}"
                )
                
                if attempt == self.retry_attempts - 1:
                    logger.error(f"文件检查最终失败: {file_path}")
                    return none
                
                time.sleep(self.retry_delay)
    
    def get_file_metadata(self, file_path: str) -> optional[dict[str, any]]:
        """
        获取文件元数据，包含完整错误处理
        """
        if not self.safe_check_exists(file_path):
            return none
        
        try:
            path = path(file_path)
            stat_info = path.stat()
            
            return {
                'size': stat_info.st_size,
                'modified_time': stat_info.st_mtime,
                'access_time': stat_info.st_atime,
                'creation_time': getattr(stat_info, 'st_birthtime', none),
                'is_file': path.is_file(),
                'is_dir': path.is_dir(),
                'is_symlink': path.is_symlink(),
            }
            
        except oserror as e:
            logger.error(f"获取文件元数据失败: {file_path} - {e}")
            return none

# 使用示例
file_checker = robustfilechecker()

# 检查文件是否存在
exists = file_checker.safe_check_exists("/etc/passwd")
print(f"/etc/passwd 存在: {exists}")

# 获取文件元数据
metadata = file_checker.get_file_metadata("/etc/passwd")
if metadata:
    print("文件元数据:", metadata)