# 典型的CSV格式
name,age,city
張三,30,北京
李四,25,上海

# 用分號(hào)分隔（歐洲常見(jiàn)）
name;age;city
張三;30;北京
李四;25;上海

# 用制表符分隔（TSV文件）
name	age	city
張三	30	北京
李四	25	上海

# 帶引號(hào)的CSV（包含逗號(hào)時(shí)）
product,price,description
"蘋(píng)果,紅富士",5.8,"新鮮,甜"
香蕉,3.2,"進(jìn)口,黃色"

import csv

# 1. 讀取CSV文件
with open('data.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)  # 每行是一個(gè)列表

# 2. 寫(xiě)入CSV文件
data = [
    ['name', 'age', 'city'],
    ['張三', '30', '北京'],
    ['李四', '25', '上海']
]

with open('output.csv', 'w', encoding='utf-8', newline='') as file:
    writer = csv.writer(file)
    writer.writerows(data)

# 3. 用字典方式讀?。ㄍ扑]！）
with open('data.csv', 'r', encoding='utf-8') as file:
    reader = csv.DictReader(file)
    for row in reader:
        print(row['name'], row['age'])  # 通過(guò)列名訪問(wèn)

# 4. 用字典方式寫(xiě)入
data = [
    {'name': '張三', 'age': '30', 'city': '北京'},
    {'name': '李四', 'age': '25', 'city': '上海'}
]

with open('output_dict.csv', 'w', encoding='utf-8', newline='') as file:
    fieldnames = ['name', 'age', 'city']
    writer = csv.DictWriter(file, fieldnames=fieldnames)
    writer.writeheader()  # 寫(xiě)入表頭
    writer.writerows(data)

date,product,quantity,price,region
2023-01-15,Widget-A,10,29.99,North
2023-01-15,Gadget-X,5,99.99,South
2023-01-16,Widget-A,8,29.99,East
2023-01-16,Thingy-B,3,149.99,West

# analyze_sales.py
import csv
from collections import defaultdict
from datetime import datetime

def analyze_sales_data(filepath):
    """
    分析銷售數(shù)據(jù)
    """
    # 存儲(chǔ)統(tǒng)計(jì)結(jié)果
    stats = {
        'total_sales': 0,
        'total_quantity': 0,
        'by_product': defaultdict(float),
        'by_region': defaultdict(float),
        'by_date': defaultdict(float)
    }
    
    with open(filepath, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        
        for row in reader:
            try:
                # 轉(zhuǎn)換數(shù)據(jù)類型
                quantity = int(row['quantity'])
                price = float(row['price'])
                date_str = row['date']
                
                # 計(jì)算銷售額
                sales = quantity * price
                
                # 更新統(tǒng)計(jì)
                stats['total_sales'] += sales
                stats['total_quantity'] += quantity
                stats['by_product'][row['product']] += sales
                stats['by_region'][row['region']] += sales
                stats['by_date'][date_str] += sales
                
            except (ValueError, KeyError) as e:
                print(f"數(shù)據(jù)格式錯(cuò)誤: {row}, 錯(cuò)誤: {e}")
                continue
    
    return stats

def print_statistics(stats):
    """
    打印統(tǒng)計(jì)結(jié)果
    """
    print("=" * 50)
    print("銷售數(shù)據(jù)分析報(bào)告")
    print("=" * 50)
    
    print(f"\n總銷售額: ${stats['total_sales']:.2f}")
    print(f"總銷售數(shù)量: {stats['total_quantity']}")
    
    print("\n按產(chǎn)品統(tǒng)計(jì):")
    for product, sales in sorted(stats['by_product'].items(), 
                                 key=lambda x: x[1], reverse=True):
        print(f"  {product}: ${sales:.2f}")
    
    print("\n按地區(qū)統(tǒng)計(jì):")
    for region, sales in sorted(stats['by_region'].items(),
                                key=lambda x: x[1], reverse=True):
        print(f"  {region}: ${sales:.2f}")
    
    print("\n按日期統(tǒng)計(jì):")
    for date, sales in sorted(stats['by_date'].items()):
        print(f"  {date}: ${sales:.2f}")

# 生成測(cè)試數(shù)據(jù)
def create_sample_data():
    """創(chuàng)建示例銷售數(shù)據(jù)"""
    data = [
        ['date', 'product', 'quantity', 'price', 'region'],
        ['2023-01-15', 'Widget-A', 10, 29.99, 'North'],
        ['2023-01-15', 'Gadget-X', 5, 99.99, 'South'],
        ['2023-01-16', 'Widget-A', 8, 29.99, 'East'],
        ['2023-01-16', 'Thingy-B', 3, 149.99, 'West'],
        ['2023-01-16', 'Gadget-X', 12, 99.99, 'North'],
        ['2023-01-17', 'Widget-A', 15, 27.50, 'South'],
    ]
    
    with open('sales.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        writer.writerows(data)
    
    print("示例數(shù)據(jù)已創(chuàng)建: sales.csv")

if __name__ == "__main__":
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    create_sample_data()
    
    # 分析數(shù)據(jù)
    stats = analyze_sales_data('sales.csv')
    
    # 打印結(jié)果
    print_statistics(stats)

# clean_data.py
import csv
import re

def clean_csv_file(input_file, output_file):
    """
    清洗CSV數(shù)據(jù)
    """
    cleaned_rows = []
    
    with open(input_file, 'r', encoding='utf-8') as infile:
        reader = csv.DictReader(infile)
        fieldnames = reader.fieldnames
        
        for i, row in enumerate(reader, 1):
            cleaned_row = {}
            
            for field in fieldnames:
                original_value = row.get(field, '')
                cleaned_value = clean_field(field, original_value)
                cleaned_row[field] = cleaned_value
            
            # 檢查是否有必要的數(shù)據(jù)
            if is_valid_row(cleaned_row):
                cleaned_rows.append(cleaned_row)
            else:
                print(f"跳過(guò)第{i}行無(wú)效數(shù)據(jù): {row}")
    
    # 寫(xiě)入清洗后的數(shù)據(jù)
    with open(output_file, 'w', encoding='utf-8', newline='') as outfile:
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(cleaned_rows)
    
    print(f"數(shù)據(jù)清洗完成: {len(cleaned_rows)} 行有效數(shù)據(jù)")

def clean_field(field_name, value):
    """
    根據(jù)字段類型清洗數(shù)據(jù)
    """
    if not value:
        return value
    
    # 去除首尾空白
    value = value.strip()
    
    # 根據(jù)不同字段類型進(jìn)行清洗
    if 'email' in field_name.lower():
        return clean_email(value)
    elif 'phone' in field_name.lower():
        return clean_phone(value)
    elif 'date' in field_name.lower():
        return clean_date(value)
    elif 'price' in field_name.lower() or 'amount' in field_name.lower():
        return clean_number(value)
    else:
        return value

def clean_email(email):
    """清洗郵箱地址"""
    email = email.lower().strip()
    # 簡(jiǎn)單的郵箱格式驗(yàn)證
    if '@' in email and '.' in email.split('@')[1]:
        return email
    return ''

def clean_phone(phone):
    """清洗電話號(hào)碼"""
    # 移除非數(shù)字字符
    digits = re.sub(r'\D', '', phone)
    
    if len(digits) == 11:  # 中國(guó)手機(jī)號(hào)
        return digits
    elif len(digits) == 10:  # 固定電話
        return digits
    else:
        return ''

def clean_date(date_str):
    """清洗日期"""
    # 嘗試不同的日期格式
    formats = ['%Y-%m-%d', '%Y/%m/%d', '%Y年%m月%d日', '%d/%m/%Y']
    
    for fmt in formats:
        try:
            from datetime import datetime
            dt = datetime.strptime(date_str, fmt)
            return dt.strftime('%Y-%m-%d')  # 統(tǒng)一格式
        except ValueError:
            continue
    
    return date_str  # 無(wú)法解析，返回原值

def clean_number(number_str):
    """清洗數(shù)字"""
    # 移除貨幣符號(hào)、千分位分隔符等
    cleaned = re.sub(r'[^\d\.-]', '', number_str)
    
    try:
        return str(float(cleaned))
    except ValueError:
        return '0'

def is_valid_row(row):
    """檢查行是否有效"""
    # 必須有姓名
    if not row.get('name', '').strip():
        return False
    
    # 郵箱必須有效（如果有的話）
    email = row.get('email', '')
    if email and '@' not in email:
        return False
    
    return True

def create_dirty_data():
    """創(chuàng)建包含臟數(shù)據(jù)的測(cè)試文件"""
    dirty_data = [
        ['name', 'email', 'phone', 'join_date', 'amount'],
        [' 張三 ', 'ZHANGSAN@EXAMPLE.COM', '138-0013-8000', '2023-01-15', '￥1,000.50'],
        ['李四', 'lisi.example.com', '010-62345678', '2023/02/20', '2,500.00'],
        ['', 'wangwu@example.com', '13912345678', '2023年03月15日', '1500'],
        ['趙六', 'zhaoliu@example', '123-4567', '無(wú)效日期', 'ABC'],
        ['錢(qián)七', 'qianqi@example.com', '15012345678', '20/04/2023', '3,000.00'],
    ]
    
    with open('dirty_data.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        writer.writerows(dirty_data)
    
    print("臟數(shù)據(jù)測(cè)試文件已創(chuàng)建: dirty_data.csv")

if __name__ == "__main__":
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    create_dirty_data()
    
    # 清洗數(shù)據(jù)
    clean_csv_file('dirty_data.csv', 'cleaned_data.csv')
    
    # 顯示清洗結(jié)果
    print("\n清洗前后對(duì)比:")
    print("-" * 50)
    
    with open('dirty_data.csv', 'r', encoding='utf-8') as f:
        print("原始數(shù)據(jù):")
        for line in f:
            print("  ", line.strip())
    
    print("\n清洗后數(shù)據(jù):")
    with open('cleaned_data.csv', 'r', encoding='utf-8') as f:
        for line in f:
            print("  ", line.strip())

# merge_split_csv.py
import csv
import os
from glob import glob

def merge_csv_files(pattern, output_file):
    """
    合并多個(gè)CSV文件
    
    參數(shù):
        pattern: 文件匹配模式，如 'data/*.csv'
        output_file: 合并后的輸出文件
    """
    all_data = []
    headers = None
    
    for filename in glob(pattern):
        print(f"處理文件: {filename}")
        
        with open(filename, 'r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            
            # 檢查表頭是否一致
            if headers is None:
                headers = reader.fieldnames
            elif headers != reader.fieldnames:
                print(f"警告: {filename} 的表頭不一致，跳過(guò)")
                continue
            
            # 讀取數(shù)據(jù)
            for row in reader:
                all_data.append(row)
    
    if not all_data:
        print("沒(méi)有找到可合并的數(shù)據(jù)")
        return
    
    # 寫(xiě)入合并后的文件
    with open(output_file, 'w', encoding='utf-8', newline='') as file:
        writer = csv.DictWriter(file, fieldnames=headers)
        writer.writeheader()
        writer.writerows(all_data)
    
    print(f"合并完成: 共 {len(all_data)} 行數(shù)據(jù)，保存到 {output_file}")

def split_csv_by_column(input_file, column_name, output_dir):
    """
    按某一列的值拆分CSV文件
    
    參數(shù):
        input_file: 輸入文件
        column_name: 按此列拆分
        output_dir: 輸出目錄
    """
    # 確保輸出目錄存在
    os.makedirs(output_dir, exist_ok=True)
    
    # 按列值分組數(shù)據(jù)
    groups = {}
    
    with open(input_file, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        headers = reader.fieldnames
        
        for row in reader:
            key = row.get(column_name, 'unknown')
            
            if key not in groups:
                groups[key] = []
            
            groups[key].append(row)
    
    # 為每個(gè)組創(chuàng)建文件
    for key, rows in groups.items():
        # 創(chuàng)建安全的文件名
        safe_key = "".join(c for c in str(key) if c.isalnum() or c in (' ', '-', '_')).rstrip()
        output_file = os.path.join(output_dir, f"{safe_key}.csv")
        
        with open(output_file, 'w', encoding='utf-8', newline='') as file:
            writer = csv.DictWriter(file, fieldnames=headers)
            writer.writeheader()
            writer.writerows(rows)
        
        print(f"創(chuàng)建文件: {output_file} ({len(rows)} 行)")
    
    print(f"拆分完成: 共創(chuàng)建 {len(groups)} 個(gè)文件")

def create_monthly_data():
    """創(chuàng)建月度測(cè)試數(shù)據(jù)"""
    months = ['2023-01', '2023-02', '2023-03']
    
    for month in months:
        filename = f"monthly_data_{month}.csv"
        data = [
            ['date', 'product', 'sales', 'region'],
            [f'{month}-10', 'Product-A', '1000', 'North'],
            [f'{month}-15', 'Product-B', '1500', 'South'],
            [f'{month}-20', 'Product-A', '800', 'East'],
        ]
        
        with open(filename, 'w', encoding='utf-8', newline='') as file:
            writer = csv.writer(file)
            writer.writerows(data)
        
        print(f"創(chuàng)建: {filename}")

if __name__ == "__main__":
    print("=== CSV文件合并與拆分演示 ===\n")
    
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    print("1. 創(chuàng)建月度測(cè)試數(shù)據(jù)...")
    create_monthly_data()
    
    # 合并文件
    print("\n2. 合并月度數(shù)據(jù)...")
    merge_csv_files('monthly_data_*.csv', 'merged_quarterly.csv')
    
    # 拆分文件
    print("\n3. 按地區(qū)拆分?jǐn)?shù)據(jù)...")
    split_csv_by_column('merged_quarterly.csv', 'region', 'split_by_region')
    
    # 顯示結(jié)果
    print("\n" + "=" * 50)
    print("生成的文件:")
    for file in ['merged_quarterly.csv', 'split_by_region']:
        if os.path.exists(file):
            if os.path.isdir(file):
                print(f"目錄: {file}/")
                for f in os.listdir(file):
                    print(f"  - {f}")
            else:
                print(f"文件: {file}")

# special_csv_formats.py
import csv

def read_excel_csv():
    """
    處理從Excel導(dǎo)出的CSV
    Excel導(dǎo)出的CSV常有BOM和特殊編碼
    """
    files_to_try = [
        ('utf-8-sig', 'excel_utf8_bom.csv'),  # UTF-8 with BOM
        ('gbk', 'excel_gbk.csv'),             # GBK編碼（中文Windows）
        ('utf-8', 'excel_utf8.csv'),          # 普通UTF-8
    ]
    
    for encoding, filename in files_to_try:
        try:
            with open(filename, 'r', encoding=encoding) as file:
                reader = csv.reader(file)
                data = list(reader)
            
            print(f"成功讀取 {filename} (編碼: {encoding})")
            print(f"  表頭: {data[0]}")
            print(f"  行數(shù): {len(data)}")
            break
        except (UnicodeDecodeError, FileNotFoundError):
            continue

def handle_custom_delimiters():
    """處理自定義分隔符"""
    # 用分號(hào)分隔的文件
    with open('european_format.csv', 'w', encoding='utf-8') as f:
        f.write("name;age;city\n")
        f.write("張三;30;北京\n")
        f.write("李四;25;上海\n")
    
    # 讀取時(shí)指定分隔符
    with open('european_format.csv', 'r', encoding='utf-8') as file:
        reader = csv.reader(file, delimiter=';')
        for row in reader:
            print(f"分隔符';': {row}")
    
    # TSV文件（制表符分隔）
    with open('data.tsv', 'w', encoding='utf-8') as f:
        f.write("name\tage\tcity\n")
        f.write("張三\t30\t北京\n")
        f.write("李四\t25\t上海\n")
    
    with open('data.tsv', 'r', encoding='utf-8') as file:
        reader = csv.reader(file, delimiter='\t')
        for row in reader:
            print(f"分隔符'\\t': {row}")

def handle_quoted_data():
    """處理帶引號(hào)的數(shù)據(jù)"""
    # 創(chuàng)建包含逗號(hào)和引號(hào)的數(shù)據(jù)
    tricky_data = [
        ['name', 'description', 'price'],
        ['Apple', 'Red, delicious apple', '5.8'],
        ['Banana', 'Yellow "sweet" banana', '3.2'],
        ['Orange Juice', 'Fresh, 100% pure orange "juice"', '12.5'],
    ]
    
    with open('quoted_data.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file, quoting=csv.QUOTE_ALL)  # 所有字段都加引號(hào)
        writer.writerows(tricky_data)
    
    print("\n帶引號(hào)的數(shù)據(jù)文件已創(chuàng)建")
    
    # 讀取時(shí)處理引號(hào)
    with open('quoted_data.csv', 'r', encoding='utf-8') as file:
        reader = csv.reader(file)
        for row in reader:
            print(f"原始行: {row}")

def handle_large_files():
    """處理大文件的內(nèi)存友好方式"""
    print("\n大文件處理策略:")
    print("1. 流式處理，一次一行，不加載到內(nèi)存")
    print("2. 分批處理，特別是需要排序或聚合時(shí)")
    print("3. 使用生成器避免內(nèi)存累積")
    
    def process_large_file(filename, chunk_size=1000):
        """分批處理大文件"""
        with open(filename, 'r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            
            chunk = []
            for i, row in enumerate(reader, 1):
                chunk.append(row)
                
                if len(chunk) >= chunk_size:
                    yield chunk
                    chunk = []
            
            if chunk:  # 最后一批
                yield chunk
    
    # 模擬處理
    print("\n模擬處理大文件（分塊）:")
    for chunk_num, chunk in enumerate(process_large_file('large_data.csv', chunk_size=2), 1):
        print(f"  處理第{chunk_num}塊: {len(chunk)} 行")
        # 這里可以處理每個(gè)數(shù)據(jù)塊
        # 比如寫(xiě)入數(shù)據(jù)庫(kù)、進(jìn)行聚合計(jì)算等

if __name__ == "__main__":
    print("=== 特殊CSV格式處理 ===\n")
    
    # 處理自定義分隔符
    print("1. 自定義分隔符處理:")
    handle_custom_delimiters()
    
    # 處理帶引號(hào)的數(shù)據(jù)
    print("\n2. 帶引號(hào)數(shù)據(jù)處理:")
    handle_quoted_data()
    
    # 大文件處理
    print("\n3. 大文件處理策略:")
    handle_large_files()

# csv_performance.py
import csv
import time
import random
from memory_profiler import profile

def create_large_csv(filename, num_rows=100000):
    """創(chuàng)建大型測(cè)試CSV文件"""
    print(f"創(chuàng)建大型測(cè)試文件: {filename} ({num_rows} 行)")
    
    with open(filename, 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        
        # 寫(xiě)入表頭
        writer.writerow(['id', 'name', 'value', 'category', 'timestamp'])
        
        # 寫(xiě)入數(shù)據(jù)
        for i in range(num_rows):
            writer.writerow([
                i + 1,
                f'Item-{random.randint(1, 1000)}',
                random.uniform(1, 1000),
                random.choice(['A', 'B', 'C', 'D']),
                f'2023-{random.randint(1,12):02d}-{random.randint(1,28):02d}'
            ])
    
    print("文件創(chuàng)建完成")

def naive_processing(filename):
    """原始方法：一次性加載所有數(shù)據(jù)"""
    print("\n方法1: 一次性加載所有數(shù)據(jù)")
    start = time.time()
    
    with open(filename, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        all_data = list(reader)  # 這里會(huì)加載所有數(shù)據(jù)到內(nèi)存
    
    # 處理數(shù)據(jù)
    total = 0
    for row in all_data:
        total += float(row['value'])
    
    elapsed = time.time() - start
    print(f"  處理了 {len(all_data)} 行數(shù)據(jù)")
    print(f"  總和: {total:.2f}")
    print(f"  耗時(shí): {elapsed:.2f} 秒")
    
    return elapsed

def stream_processing(filename):
    """流式處理：一次處理一行"""
    print("\n方法2: 流式處理（一次一行）")
    start = time.time()
    
    total = 0
    count = 0
    
    with open(filename, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        
        for row in reader:
            total += float(row['value'])
            count += 1
    
    elapsed = time.time() - start
    print(f"  處理了 {count} 行數(shù)據(jù)")
    print(f"  總和: {total:.2f}")
    print(f"  耗時(shí): {elapsed:.2f} 秒")
    
    return elapsed

def batch_processing(filename, batch_size=10000):
    """批量處理：分批處理數(shù)據(jù)"""
    print(f"\n方法3: 批量處理（每批 {batch_size} 行）")
    start = time.time()
    
    total = 0
    count = 0
    batch = []
    
    with open(filename, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        
        for row in reader:
            batch.append(float(row['value']))
            count += 1
            
            if len(batch) >= batch_size:
                # 處理一個(gè)批次
                total += sum(batch)
                batch = []  # 清空批次
        
        # 處理最后一批
        if batch:
            total += sum(batch)
    
    elapsed = time.time() - start
    print(f"  處理了 {count} 行數(shù)據(jù)")
    print(f"  總和: {total:.2f}")
    print(f"  耗時(shí): {elapsed:.2f} 秒")
    
    return elapsed

def optimized_writing(filename, num_rows=10000):
    """優(yōu)化寫(xiě)入性能"""
    print(f"\n寫(xiě)入性能優(yōu)化 ({num_rows} 行):")
    
    # 方法1: 逐行寫(xiě)入
    start = time.time()
    with open('slow_write.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['id', 'value'])
        
        for i in range(num_rows):
            writer.writerow([i, i * 1.5])  # 每次調(diào)用writerow
    
    elapsed1 = time.time() - start
    print(f"  逐行寫(xiě)入: {elapsed1:.2f} 秒")
    
    # 方法2: 批量寫(xiě)入
    start = time.time()
    with open('fast_write.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['id', 'value'])
        
        # 準(zhǔn)備所有數(shù)據(jù)
        all_data = [[i, i * 1.5] for i in range(num_rows)]
        writer.writerows(all_data)  # 一次寫(xiě)入所有行
    
    elapsed2 = time.time() - start
    print(f"  批量寫(xiě)入: {elapsed2:.2f} 秒")
    print(f"  速度提升: {elapsed1/elapsed2:.1f} 倍")

if __name__ == "__main__":
    print("=== CSV性能優(yōu)化技巧 ===\n")
    
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    test_file = 'large_test_data.csv'
    create_large_csv(test_file, num_rows=50000)
    
    # 測(cè)試不同處理方法的性能
    times = []
    times.append(('一次性加載', naive_processing(test_file)))
    times.append(('流式處理', stream_processing(test_file)))
    times.append(('批量處理', batch_processing(test_file)))
    
    # 寫(xiě)入性能測(cè)試
    optimized_writing('test_write.csv', num_rows=10000)
    
    # 總結(jié)
    print("\n" + "=" * 50)
    print("性能總結(jié):")
    for method, t in times:
        print(f"  {method}: {t:.2f} 秒")

# csv_toolkit.py
"""
CSV處理工具箱

一個(gè)實(shí)用的命令行工具，集成各種CSV處理功能
"""
import csv
import argparse
import sys
import os
from pathlib import Path

class CSVToolkit:
    """CSV處理工具箱"""
    
    def __init__(self):
        self.commands = {
            'view': self.view_csv,
            'head': self.head_csv,
            'tail': self.tail_csv,
            'stats': self.csv_stats,
            'filter': self.filter_csv,
            'select': self.select_columns,
            'sample': self.sample_csv,
        }
    
    def run(self):
        """運(yùn)行工具箱"""
        parser = argparse.ArgumentParser(description='CSV處理工具箱')
        parser.add_argument('command', choices=self.commands.keys(), 
                          help='要執(zhí)行的操作')
        parser.add_argument('file', help='CSV文件路徑')
        parser.add_argument('-o', '--output', help='輸出文件路徑')
        parser.add_argument('-n', '--lines', type=int, default=10,
                          help='顯示的行數(shù)（用于head/tail）')
        parser.add_argument('-c', '--columns', help='選擇的列，用逗號(hào)分隔')
        parser.add_argument('-f', '--filter', help='過(guò)濾條件，格式: 列名=值')
        
        args = parser.parse_args()
        
        # 檢查文件是否存在
        if not os.path.exists(args.file):
            print(f"錯(cuò)誤: 文件 '{args.file}' 不存在")
            sys.exit(1)
        
        # 執(zhí)行命令
        self.commands[args.command](args)
    
    def view_csv(self, args):
        """查看CSV文件內(nèi)容"""
        with open(args.file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            
            for i, row in enumerate(reader):
                # 美化輸出
                formatted = ' | '.join(f'{cell:<20}' for cell in row)
                print(formatted)
                
                # 顯示表頭分隔線
                if i == 0:
                    print('-' * len(formatted))
    
    def head_csv(self, args):
        """顯示文件前n行"""
        with open(args.file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            
            for i, row in enumerate(reader):
                if i >= args.lines + 1:  # +1 為了包含表頭
                    break
                print(', '.join(row))
    
    def tail_csv(self, args):
        """顯示文件后n行"""
        with open(args.file, 'r', encoding='utf-8') as file:
            lines = file.readlines()
            
        # 顯示最后n行
        for line in lines[-args.lines:]:
            print(line.rstrip())
    
    def csv_stats(self, args):
        """顯示CSV文件統(tǒng)計(jì)信息"""
        with open(args.file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            data = list(reader)
        
        if not data:
            print("文件為空")
            return
        
        headers = data[0]
        rows = data[1:]
        
        print(f"文件: {args.file}")
        print(f"行數(shù): {len(rows)} 行（不含表頭）")
        print(f"列數(shù): {len(headers)} 列")
        print(f"列名: {', '.join(headers)}")
        
        # 每列的數(shù)據(jù)類型推測(cè)
        if rows:
            print("\n列信息:")
            for i, header in enumerate(headers):
                sample = rows[0][i] if len(rows[0]) > i else ''
                print(f"  {header}: 示例='{sample}'")
    
    def filter_csv(self, args):
        """過(guò)濾CSV文件"""
        if not args.filter:
            print("錯(cuò)誤: 需要指定過(guò)濾條件（-f 列名=值）")
            return
        
        # 解析過(guò)濾條件
        col_name, value = args.filter.split('=', 1)
        
        with open(args.file, 'r', encoding='utf-8') as infile:
            reader = csv.DictReader(infile)
            headers = reader.fieldnames
            
            # 收集匹配的行
            matching_rows = []
            for row in reader:
                if row.get(col_name) == value:
                    matching_rows.append(row)
        
        # 輸出結(jié)果
        if args.output:
            with open(args.output, 'w', encoding='utf-8', newline='') as outfile:
                writer = csv.DictWriter(outfile, fieldnames=headers)
                writer.writeheader()
                writer.writerows(matching_rows)
            print(f"過(guò)濾結(jié)果已保存到: {args.output} ({len(matching_rows)} 行)")
        else:
            # 打印到控制臺(tái)
            writer = csv.DictWriter(sys.stdout, fieldnames=headers)
            writer.writeheader()
            writer.writerows(matching_rows)
    
    def select_columns(self, args):
        """選擇特定列"""
        if not args.columns:
            print("錯(cuò)誤: 需要指定要選擇的列（-c 列1,列2,...）")
            return
        
        selected = [col.strip() for col in args.columns.split(',')]
        
        with open(args.file, 'r', encoding='utf-8') as infile:
            reader = csv.DictReader(infile)
            
            # 檢查列是否存在
            for col in selected:
                if col not in reader.fieldnames:
                    print(f"警告: 列 '{col}' 不存在")
            
            # 只保留選中的列
            filtered_rows = []
            for row in reader:
                filtered_row = {col: row[col] for col in selected if col in row}
                filtered_rows.append(filtered_row)
        
        # 輸出結(jié)果
        if args.output:
            with open(args.output, 'w', encoding='utf-8', newline='') as outfile:
                writer = csv.DictWriter(outfile, fieldnames=selected)
                writer.writeheader()
                writer.writerows(filtered_rows)
            print(f"結(jié)果已保存到: {args.output}")
        else:
            writer = csv.DictWriter(sys.stdout, fieldnames=selected)
            writer.writeheader()
            writer.writerows(filtered_rows)
    
    def sample_csv(self, args):
        """隨機(jī)抽樣"""
        import random
        
        with open(args.file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            data = list(reader)
        
        if len(data) <= 1:
            print("文件數(shù)據(jù)不足")
            return
        
        headers = data[0]
        rows = data[1:]
        
        # 隨機(jī)抽樣
        sample_size = min(args.lines, len(rows))
        sampled_rows = random.sample(rows, sample_size)
        
        # 輸出結(jié)果
        output_data = [headers] + sampled_rows
        
        if args.output:
            with open(args.output, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                writer.writerows(output_data)
            print(f"抽樣結(jié)果已保存到: {args.output} ({sample_size} 行)")
        else:
            for row in output_data:
                print(', '.join(row))

def create_example_csv():
    """創(chuàng)建示例CSV文件"""
    data = [
        ['id', 'name', 'age', 'city', 'salary'],
        ['1', '張三', '30', '北京', '50000'],
        ['2', '李四', '25', '上海', '45000'],
        ['3', '王五', '35', '廣州', '60000'],
        ['4', '趙六', '28', '深圳', '55000'],
        ['5', '錢(qián)七', '32', '杭州', '52000'],
        ['6', '孫八', '29', '南京', '48000'],
        ['7', '周九', '31', '成都', '51000'],
        ['8', '吳十', '27', '武漢', '47000'],
    ]
    
    with open('employees.csv', 'w', encoding='utf-8', newline='') as file:
        writer = csv.writer(file)
        writer.writerows(data)
    
    print("示例文件已創(chuàng)建: employees.csv")

def demo_toolkit():
    """演示工具箱功能"""
    print("=== CSV工具箱功能演示 ===\n")
    
    # 創(chuàng)建示例文件
    create_example_csv()
    
    print("\n1. 查看文件內(nèi)容:")
    print("-" * 50)
    toolkit = CSVToolkit()
    
    # 模擬命令行參數(shù)
    class Args:
        pass
    
    args = Args()
    args.file = 'employees.csv'
    args.lines = 5
    args.columns = 'name,city,salary'
    args.filter = 'city=北京'
    args.output = None
    
    print("\n2. 顯示前3行:")
    print("-" * 50)
    args.lines = 3
    toolkit.head_csv(args)
    
    print("\n3. 文件統(tǒng)計(jì):")
    print("-" * 50)
    toolkit.csv_stats(args)
    
    print("\n4. 選擇特定列:")
    print("-" * 50)
    toolkit.select_columns(args)
    
    print("\n5. 過(guò)濾數(shù)據(jù):")
    print("-" * 50)
    toolkit.filter_csv(args)
    
    print("\n6. 隨機(jī)抽樣:")
    print("-" * 50)
    args.lines = 2
    toolkit.sample_csv(args)

if __name__ == "__main__":
    if len(sys.argv) > 1:
        # 命令行模式
        toolkit = CSVToolkit()
        toolkit.run()
    else:
        # 演示模式
        demo_toolkit()
        
        print("\n" + "=" * 50)
        print("使用說(shuō)明:")
        print("  命令行使用: python csv_toolkit.py <命令> <文件> [選項(xiàng)]")
        print("\n可用命令:")
        for cmd in CSVToolkit().commands:
            print(f"  {cmd}")

encodings = ['utf-8-sig', 'gbk', 'utf-8', 'latin-1']
for enc in encodings:
    try:
        with open('file.csv', 'r', encoding=enc) as f:
            content = f.read()
        print(f"使用編碼: {enc}")
        break
    except UnicodeDecodeError:
        continue

with open('large.csv', 'r') as file:
    reader = csv.DictReader(file)
    for row in reader:  # 一次只加載一行到內(nèi)存
        process(row)