import os
import os.path

def check_file_existence_basic(file_path):
    """
    基礎(chǔ)文件存在性檢測
    """
    # 檢查文件是否存在
    if os.path.exists(file_path):
        print(f"文件 {file_path} 存在")
        
        # 進(jìn)一步檢查是否是文件（不是目錄）
        if os.path.isfile(file_path):
            print("這是一個(gè)文件")
            return True
        else:
            print("這不是一個(gè)普通文件（可能是目錄）")
            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):
    """
    詳細(xì)分析路徑類型
    """
    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} 是符號(hào)鏈接")
        # 解析符號(hào)鏈接指向的實(shí)際路徑
        real_path = os.path.realpath(path)
        print(f"指向: {real_path}")
        
    # 檢查文件權(quán)限
    if os.access(path, os.R_OK):
        print("當(dāng)前用戶有讀取權(quán)限")
    if os.access(path, os.W_OK):
        print("當(dāng)前用戶有寫入權(quán)限")
    if os.access(path, os.X_OK):
        print("當(dāng)前用戶有執(zhí)行權(quán)限")
    
    return True

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

import os

def check_file_eafp(file_path):
    """
    EAFP風(fēng)格：直接嘗試操作，捕獲異常
    """
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        print("文件存在且可讀取")
        return True, content
    except FileNotFoundError:
        print("文件不存在")
        return False, None
    except PermissionError:
        print("沒有讀取權(quán)限")
        return False, None
    except IOError as e:
        print(f"讀取文件時(shí)發(fā)生錯(cuò)誤: {e}")
        return False, None

def check_file_lbyl(file_path):
    """
    LBYL風(fēng)格：先檢查再操作
    """
    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("沒有讀取權(quán)限")
        return False, None
        
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        return True, content
    except IOError as e:
        print(f"讀取文件時(shí)發(fā)生錯(cuò)誤: {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("這是一個(gè)文件")
            # 獲取文件信息
            print(f"文件大小: {path.stat().st_size} 字節(jié)")
            print(f"最后修改時(shí)間: {path.stat().st_mtime}")
            return True
        elif path.is_dir():
            print("這是一個(gè)目錄")
            return False
        elif path.is_symlink():
            print("這是一個(gè)符號(hào)鏈接")
            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():
        """稍后創(chuàng)建文件"""
        time.sleep(0.1)  # 短暫延遲
        with open(test_file, 'w') as f:
            f.write("測試內(nèi)容")
    
    # 啟動(dòng)線程創(chuàng)建文件
    thread = Thread(target=create_file_later)
    thread.start()
    
    # 檢查文件是否存在
    if os.path.exists(test_file):
        print("文件存在，嘗試讀取...")
        # 這里可能出現(xiàn)競爭條件：文件可能在被檢查后立即被刪除
        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)

# 運(yùn)行演示
competitive_check_demo()

import os
import tempfile

def safe_file_operation(file_path, operation_func, *args, **kwargs):
    """
    安全的文件操作函數(shù)，處理競爭條件
    """
    max_retries = 3
    retry_delay = 0.1  # 100毫秒
    
    for attempt in range(max_retries):
        try:
            # 嘗試執(zhí)行操作
            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("成功讀取文件內(nèi)容")
except FileNotFoundError:
    print("文件不存在")
except Exception as e:
    print(f"讀取文件時(shí)發(fā)生錯(cuò)誤: {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()  # 處理~擴(kuò)展
            
            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"加載配置文件時(shí)發(fā)生錯(cuò)誤: {e}")
            return None

# 使用示例
config_manager = ConfigManager()
config = config_manager.load_config()
if config:
    print("配置加載成功")
else:
    print("使用默認(rèn)配置")

import os
import time
from datetime import datetime
from pathlib import Path

class LogManager:
    """
    日志文件管理器，支持文件存在性檢查和輪轉(zhuǎn)
    """
    
    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):
        """
        獲取當(dāng)前日志文件路徑，必要時(shí)創(chuàng)建新文件
        """
        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):
        """
        檢查是否需要日志輪轉(zhuǎn)
        """
        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):
        """
        執(zhí)行日志輪轉(zhuǎn)
        """
        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"已輪轉(zhuǎn)日志文件: {archived_file}")
            except OSError as e:
                print(f"日志輪轉(zhuǎn)失敗: {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]:
        """
        通過內(nèi)容哈希檢查文件是否已存在
        """
        content_hash = hashlib.sha256(file_content).hexdigest()
        
        # 檢查是否已有相同內(nèi)容的文件
        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:
        """
        安全地上傳文件，處理各種邊界情況
        """
        # 檢查是否已存在相同內(nèi)容的文件
        existing_file = self.file_exists_by_content(file_content)
        if existing_file:
            print(f"文件已存在，內(nèi)容與 {existing_file} 相同")
            return existing_file
        
        # 生成唯一文件名
        target_path = self.generate_unique_filename(original_filename)
        
        # 確保目標(biāo)文件不存在（盡管概率極低）
        if target_path.exists():
            # 如果發(fā)生沖突，重新生成文件名
            target_path = self.generate_unique_filename(original_filename)
        
        # 寫入文件
        try:
            with open(target_path, 'wb') as f:
                f.write(file_content)
            
            # 再次驗(yàn)證文件寫入成功
            if not target_path.exists():
                raise IOError("文件寫入后驗(yàn)證失敗")
            
            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:
    """
    批量文件檢查器，優(yōu)化大量文件存在性檢查的性能
    """
    
    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):
        """
        性能基準(zhǔn)測試
        """
        # 順序檢查
        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"順序檢查耗時(shí): {seq_duration:.4f}秒")
        print(f"并行檢查耗時(shí): {par_duration:.4f}秒")
        print(f"性能提升: {seq_duration/par_duration:.2f}倍")
        
        # 驗(yàn)證結(jié)果一致性
        assert seq_results == par_results, "檢查結(jié)果不一致"
        
        return seq_results

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

# 創(chuàng)建一些測試文件
for i in range(10):  # 創(chuàng)建前10個(gè)文件
    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:
    """
    帶緩存的文件檢查器，減少重復(fù)檢查的開銷
    """
    
    def __init__(self, cache_ttl=5):  # 默認(rèn)緩存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
        
        # 實(shí)際檢查文件
        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:
    """
    健壯的文件檢查器，包含完整的錯(cuò)誤處理和日志記錄
    """
    
    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]:
        """
        安全地檢查文件是否存在，包含重試機(jī)制
        """
        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]]:
        """
        獲取文件元數(shù)據(jù)，包含完整錯(cuò)誤處理
        """
        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"獲取文件元數(shù)據(jù)失敗: {file_path} - {e}")
            return None

# 使用示例
file_checker = RobustFileChecker()

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

# 獲取文件元數(shù)據(jù)
metadata = file_checker.get_file_metadata("/etc/passwd")
if metadata:
    print("文件元數(shù)據(jù):", metadata)