import os
import shutil
from pathlib import Path
from datetime import datetime

class SmartOrganizer:
    """智能文件分類器，支持自定義規(guī)則與日志記錄"""
    
    # 文件類型映射規(guī)則（可擴(kuò)展）
    FILE_TYPES = {
        '圖片': ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg', '.webp'],
        '文檔': ['.pdf', '.doc', '.docx', '.txt', '.md', '.xls', '.xlsx', '.ppt', '.pptx'],
        '視頻': ['.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv'],
        '音頻': ['.mp3', '.wav', '.flac', '.aac', '.ogg'],
        '壓縮包': ['.zip', '.rar', '.7z', '.tar', '.gz'],
        '代碼': ['.py', '.js', '.html', '.css', '.java', '.cpp', '.c', '.go', '.rs'],
        '軟件包': ['.exe', '.msi', '.dmg', '.pkg', '.deb']
    }
    
    def __init__(self, source_dir: str, target_dir: str = None):
        self.source = Path(source_dir)
        self.target = Path(target_dir) if target_dir else self.source / '已整理'
        self.stats = {'moved': 0, 'skipped': 0, 'errors': []}
        
    def get_category(self, filename: str) -> str:
        """根據(jù)擴(kuò)展名判斷文件類別"""
        ext = Path(filename).suffix.lower()
        for category, extensions in self.FILE_TYPES.items():
            if ext in extensions:
                return category
        return '其他'
    
    def organize(self, group_by_date: bool = True):
        """
        執(zhí)行整理
        :param group_by_date: 是否按日期子文件夾分組
        """
        if not self.source.exists():
            raise FileNotFoundError(f"源目錄不存在: {self.source}")
        
        self.target.mkdir(parents=True, exist_ok=True)
        
        for file_path in self.source.iterdir():
            if file_path.is_file() and file_path != Path(__file__):
                try:
                    self._process_file(file_path, group_by_date)
                except Exception as e:
                    self.stats['errors'].append(f"{file_path.name}: {str(e)}")
        
        self._generate_report()
    
    def _process_file(self, file_path: Path, group_by_date: bool):
        """處理單個文件"""
        category = self.get_category(file_path.name)
        
        # 構(gòu)建目標(biāo)路徑
        dest_dir = self.target / category
        if group_by_date:
            # 按修改日期分組：2026-04/10
            mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            date_folder = f"{mtime.year}-{mtime.month:02d}"
            dest_dir = dest_dir / date_folder
        
        dest_dir.mkdir(parents=True, exist_ok=True)
        dest_path = dest_dir / file_path.name
        
        # 處理重名文件
        counter = 1
        original_dest = dest_path
        while dest_path.exists():
            stem = original_dest.stem
            suffix = original_dest.suffix
            dest_path = original_dest.with_name(f"{stem}_{counter}{suffix}")
            counter += 1
        
        shutil.move(str(file_path), str(dest_path))
        self.stats['moved'] += 1
        print(f"? {file_path.name} -> {category}/")
    
    def _generate_report(self):
        """生成整理報告"""
        report = f"""
        整理完成報告
        =================
        已移動文件: {self.stats['moved']}
        跳過文件:   {self.stats['skipped']}
        錯誤數(shù)量:   {len(self.stats['errors'])}
        輸出目錄:   {self.target.absolute()}
        """
        print(report)
        
        if self.stats['errors']:
            print("\n??  錯誤詳情:")
            for error in self.stats['errors']:
                print(f"  - {error}")

# 使用示例
if __name__ == "__main__":
    # 整理下載文件夾
    organizer = SmartOrganizer(
        source_dir=os.path.expanduser("~/Downloads"),
        target_dir=os.path.expanduser("~/Downloads/已整理")
    )
    organizer.organize(group_by_date=True)

import pandas as pd
import numpy as np
from typing import List, Dict, Callable
import re

class ExcelCleaner:
    """Excel數(shù)據(jù)清洗流水線"""
    
    def __init__(self, file_path: str):
        self.df = pd.read_excel(file_path)
        self.original_shape = self.df.shape
        self.operations_log = []
        
    def clean_pipeline(self, config: Dict) -> pd.DataFrame:
        """
        執(zhí)行清洗流水線
        :param config: 配置字典，控制各步驟是否執(zhí)行
        """
        steps = [
            ('去除完全空行', self.remove_empty_rows, config.get('remove_empty_rows', True)),
            ('去除完全空列', self.remove_empty_cols, config.get('remove_empty_cols', True)),
            ('去除重復(fù)行', self.remove_duplicates, config.get('remove_duplicates', True)),
            ('標(biāo)準(zhǔn)化列名', self.standardize_columns, config.get('standardize_columns', True)),
            ('處理缺失值', self.handle_missing, config.get('handle_missing', 'fill')),
            ('數(shù)據(jù)類型轉(zhuǎn)換', self.convert_types, config.get('convert_types', {})),
            ('字符串清理', self.clean_strings, config.get('clean_strings', True)),
        ]
        
        for desc, func, enabled in steps:
            if enabled:
                try:
                    func() if not isinstance(enabled, dict) else func(**enabled)
                    self.operations_log.append(f"? {desc}")
                except Exception as e:
                    self.operations_log.append(f"? {desc}: {str(e)}")
        
        return self.df
    
    def remove_empty_rows(self):
        """去除完全為空的行"""
        self.df.dropna(how='all', inplace=True)
    
    def remove_empty_cols(self):
        """去除完全為空的列"""
        self.df.dropna(axis=1, how='all', inplace=True)
    
    def remove_duplicates(self, subset: List[str] = None, keep: str = 'first'):
        """
        去除重復(fù)行
        :param subset: 基于哪些列判斷重復(fù)，None表示所有列
        :param keep: 'first'(保留首次)/'last'(保留末次)/False(全部刪除)
        """
        before = len(self.df)
        self.df.drop_duplicates(subset=subset, keep=keep, inplace=True)
        removed = before - len(self.df)
        if removed > 0:
            self.operations_log.append(f"  移除 {removed} 行重復(fù)數(shù)據(jù)")
    
    def standardize_columns(self):
        """標(biāo)準(zhǔn)化列名：去空格、轉(zhuǎn)小寫、替換特殊字符"""
        def clean_col(name):
            name = str(name).strip().lower()
            name = re.sub(r'\s+', '_', name)  # 空格轉(zhuǎn)下劃線
            name = re.sub(r'[^\w]', '', name)  # 去除特殊字符
            return name
        
        self.df.columns = [clean_col(col) for col in self.df.columns]
    
    def handle_missing(self, strategy: str = 'fill', fill_value = 'N/A'):
        """
        處理缺失值
        :param strategy: 'drop'(刪除)/'fill'(填充)/'interpolate'(插值)
        """
        if strategy == 'drop':
            self.df.dropna(inplace=True)
        elif strategy == 'fill':
            # 針對不同類型填充不同值
            for col in self.df.columns:
                if self.df[col].dtype == 'object':
                    self.df[col].fillna(fill_value, inplace=True)
                else:
                    self.df[col].fillna(0, inplace=True)
        elif strategy == 'interpolate':
            self.df.interpolate(method='linear', inplace=True)
    
    def convert_types(self, type_map: Dict[str, str]):
        """
        轉(zhuǎn)換數(shù)據(jù)類型
        :param type_map: {'列名': '目標(biāo)類型', ...}
        """
        for col, dtype in type_map.items():
            if col in self.df.columns:
                try:
                    if dtype == 'datetime':
                        self.df[col] = pd.to_datetime(self.df[col], errors='coerce')
                    elif dtype == 'numeric':
                        self.df[col] = pd.to_numeric(self.df[col], errors='coerce')
                    elif dtype == 'category':
                        self.df[col] = self.df[col].astype('category')
                    else:
                        self.df[col] = self.df[col].astype(dtype)
                except Exception as e:
                    print(f"類型轉(zhuǎn)換失敗 {col}: {e}")
    
    def clean_strings(self):
        """清理字符串：去除首尾空格、統(tǒng)一換行符"""
        for col in self.df.select_dtypes(include=['object']).columns:
            self.df[col] = self.df[col].astype(str).str.strip()
            self.df[col] = self.df[col].str.replace(r'\r\n|\r|\n', ' ', regex=True)
    
    def add_calculated_column(self, name: str, formula: Callable):
        """添加計算列"""
        self.df[name] = self.df.apply(formula, axis=1)
        self.operations_log.append(f"? 添加計算列: {name}")
    
    def export(self, output_path: str, sheet_name: str = '清洗后數(shù)據(jù)'):
        """導(dǎo)出結(jié)果"""
        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            self.df.to_excel(writer, sheet_name=sheet_name, index=False)
            
            # 添加清洗報告sheet
            report_df = pd.DataFrame({
                '項目': ['原始行數(shù)', '原始列數(shù)', '處理后行數(shù)', '處理后列數(shù)', '操作記錄'],
                '值': [
                    self.original_shape[0],
                    self.original_shape[1],
                    self.df.shape[0],
                    self.df.shape[1],
                    '\n'.join(self.operations_log)
                ]
            })
            report_df.to_excel(writer, sheet_name='清洗報告', index=False)
        
        print(f"? 已導(dǎo)出至: {output_path}")
        print(f"清洗報告:\n" + "\n".join(self.operations_log))

# 使用示例
if __name__ == "__main__":
    # 假設(shè)有一個銷售數(shù)據(jù)Excel，包含臟數(shù)據(jù)
    cleaner = ExcelCleaner("raw_sales_data.xlsx")
    
    config = {
        'remove_empty_rows': True,
        'remove_duplicates': True,
        'standardize_columns': True,
        'handle_missing': 'fill',  # 填充缺失值
        'convert_types': {
            'order_date': 'datetime',
            'amount': 'numeric',
            'customer_id': 'category'
        },
        'clean_strings': True
    }
    
    # 執(zhí)行清洗
    clean_df = cleaner.clean_pipeline(config)
    
    # 添加計算列示例：計算折扣后金額
    if 'amount' in clean_df.columns and 'discount' in clean_df.columns:
        cleaner.add_calculated_column(
            'final_amount',
            lambda row: row['amount'] * (1 - row['discount']) if pd.notna(row['discount']) else row['amount']
        )
    
    # 導(dǎo)出
    cleaner.export("cleaned_sales_data.xlsx")

import os
import re
from pathlib import Path

def batch_rename(directory, old_pattern, new_pattern, dry_run=True):
    """
    批量重命名文件
    :param directory: 目標(biāo)目錄
    :param old_pattern: 舊文件名匹配模式（正則）
    :param new_pattern: 新文件名替換模板，可用 \\1, \\2 引用捕獲組
    :param dry_run: 是否為試運(yùn)行模式（默認(rèn)True，只打印不執(zhí)行）
    """
    path = Path(directory)
    if not path.exists():
        print(f"? 目錄不存在: {directory}")
        return
    
    renamed_count = 0
    
    for file in path.iterdir():
        if file.is_file():
            new_name = re.sub(old_pattern, new_pattern, file.name)
            if new_name != file.name:
                new_path = file.parent / new_name
                if dry_run:
                    print(f"[試運(yùn)行] {file.name} -> {new_name}")
                else:
                    file.rename(new_path)
                    print(f"? 已重命名: {file.name} -> {new_name}")
                renamed_count += 1
    
    print(f"\n總計: {renamed_count} 個文件")
    if dry_run:
        print("? 這是試運(yùn)行模式，添加 dry_run=False 參數(shù)以實際執(zhí)行")

# 使用示例：將 "IMG_2026410_123456.jpg" 重命名為 "2026-04-10_123456.jpg"
if __name__ == "__main__":
    batch_rename(
        directory="./photos",
        old_pattern=r"IMG_(\d{4})(\d{2})(\d{2})_(\d+)\.jpg",
        new_pattern=r"\1-\2-\3_\4.jpg",
        dry_run=True  # 先試運(yùn)行，確認(rèn)無誤后改為 False
    )