#!/usr/bin/env python3
"""
Loguru庫演示：簡化Python日志記錄
"""

from loguru import logger
import sys
import time
from typing import Any, Dict

class LoguruDemo:
    """
    Loguru功能演示類
    """
    
    def __init__(self):
        """初始化日志配置"""
        # 移除默認配置，添加自定義配置
        logger.remove()
        
        # 添加控制臺輸出，帶有顏色和格式
        logger.add(
            sys.stderr,
            format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
                   "<level>{level: <8}</level> | "
                   "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
                   "<level>{message}</level>",
            level="DEBUG",
            colorize=True
        )
        
        # 添加文件輸出，自動輪轉(zhuǎn)和壓縮
        logger.add(
            "logs/demo_{time}.log",
            rotation="10 MB",  # 文件達到10MB時輪轉(zhuǎn)
            compression="zip",  # 壓縮舊日志
            retention="30 days",  # 保留30天
            level="INFO"
        )
    
    def basic_usage(self) -> None:
        """基礎(chǔ)用法演示"""
        logger.debug("這是一條調(diào)試信息")
        logger.info("程序正常運行")
        logger.warning("這是一個警告")
        logger.error("發(fā)生了一個錯誤")
        logger.critical("嚴重錯誤！")
    
    def advanced_features(self) -> None:
        """高級功能演示"""
        
        # 1. 結(jié)構(gòu)化日志
        user_data = {"user_id": 123, "name": "張三", "age": 25}
        logger.info("用戶信息: {}", user_data)
        
        # 2. 異常捕獲裝飾器
        @logger.catch
        def risky_operation(x: float, y: float) -> float:
            """可能拋出異常的操作"""
            result = x / y
            logger.success("計算成功: {} / {} = {}", x, y, result)
            return result
        
        # 3. 性能監(jiān)控
        with logger.catch(message="處理用戶數(shù)據(jù)時發(fā)生錯誤"):
            for i in range(3):
                start_time = time.time()
                time.sleep(0.1)  # 模擬工作負載
                elapsed = time.time() - start_time
                logger.info("操作 {} 耗時 {:.3f} 秒", i, elapsed)
        
        # 測試異常捕獲
        try:
            risky_operation(10, 0)
        except Exception:
            logger.exception("除法操作失敗")
    
    def context_aware_logging(self) -> None:
        """上下文感知日志"""
        
        # 使用bind添加上下文信息
        contextual_logger = logger.bind(
            request_id="req_12345",
            user_ip="192.168.1.100"
        )
        
        contextual_logger.info("收到用戶請求")
        contextual_logger.bind(action="login").info("用戶登錄")
        contextual_logger.bind(action="logout").info("用戶登出")
    
    def performance_comparison(self) -> Dict[str, Any]:
        """性能對比測試"""
        import logging as std_logging
        
        # 標準logging配置
        std_logger = std_logging.getLogger('std_demo')
        std_logger.setLevel(std_logging.DEBUG)
        handler = std_logging.StreamHandler()
        formatter = std_logging.Formatter(
            '%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s'
        )
        handler.setFormatter(formatter)
        std_logger.addHandler(handler)
        
        # 性能測試
        test_iterations = 1000
        
        # Loguru性能
        loguru_start = time.time()
        for i in range(test_iterations):
            logger.debug("測試消息 {}", i)
        loguru_time = time.time() - loguru_start
        
        # 標準logging性能
        std_start = time.time()
        for i in range(test_iterations):
            std_logger.debug("測試消息 %s", i)
        std_time = time.time() - std_start
        
        return {
            "loguru_time": loguru_time,
            "std_logging_time": std_time,
            "iterations": test_iterations,
            "loguru_faster": std_time / loguru_time if loguru_time > 0 else 0
        }

def main():
    """主函數(shù)"""
    demo = LoguruDemo()
    
    print("=" * 60)
    print("Loguru 演示")
    print("=" * 60)
    
    print("\n1. 基礎(chǔ)用法:")
    demo.basic_usage()
    
    print("\n2. 高級功能:")
    demo.advanced_features()
    
    print("\n3. 上下文感知日志:")
    demo.context_aware_logging()
    
    print("\n4. 性能對比:")
    results = demo.performance_comparison()
    print(f"   測試迭代次數(shù): {results['iterations']}")
    print(f"   Loguru耗時: {results['loguru_time']:.4f}秒")
    print(f"   標準logging耗時: {results['std_logging_time']:.4f}秒")
    print(f"   Loguru快 {results['loguru_faster']:.2f} 倍")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Rich庫演示：創(chuàng)建美觀的終端輸出
"""

from rich.console import Console
from rich.table import Table
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from rich.panel import Panel
from rich.layout import Layout
from rich.markdown import Markdown
from rich.syntax import Syntax
from rich.tree import Tree
from rich import box
import time
from typing import List, Dict, Any

class RichDemo:
    """
    Rich庫功能演示
    """
    
    def __init__(self):
        self.console = Console()
    
    def basic_styling(self) -> None:
        """基礎(chǔ)樣式演示"""
        self.console.print("[bold red]紅色粗體文本[/bold red]")
        self.console.print("[italic blue]藍色斜體文本[/italic blue]")
        self.console.print("[underline green]綠色下劃線文本[/underline green]")
        self.console.print("[reverse yellow]黃色反色文本[/reverse yellow]")
        self.console.print(":smiley: :snake: :rocket: 表情符號支持")
    
    def advanced_tables(self) -> None:
        """高級表格功能"""
        
        # 創(chuàng)建表格
        table = Table(
            title="?? 員工信息表",
            show_header=True,
            header_style="bold magenta",
            box=box.ROUNDED
        )
        
        # 添加列
        table.add_column("ID", style="cyan", width=8)
        table.add_column("姓名", style="green", width=15)
        table.add_column("職位", style="yellow", width=20)
        table.add_column("薪資", justify="right", style="bold red", width=12)
        table.add_column("狀態(tài)", style="blue", width=10)
        
        # 添加數(shù)據(jù)行
        employees = [
            (1, "張三", "軟件工程師", "￥15,000", "在職"),
            (2, "李四", "產(chǎn)品經(jīng)理", "￥18,000", "在職"),
            (3, "王五", "前端開發(fā)", "￥12,000", "離職"),
            (4, "趙六", "數(shù)據(jù)科學(xué)家", "￥20,000", "在職")
        ]
        
        for emp in employees:
            status_style = "green" if emp[4] == "在職" else "red"
            table.add_row(str(emp[0]), emp[1], emp[2], emp[3], f"[{status_style}]{emp[4]}[/{status_style}]")
        
        self.console.print(table)
    
    def progress_bars_demo(self) -> None:
        """進度條演示"""
        
        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
            console=self.console
        ) as progress:
            
            # 模擬多個任務(wù)
            task1 = progress.add_task("[red]下載文件...", total=100)
            task2 = progress.add_task("[green]處理數(shù)據(jù)...", total=150)
            task3 = progress.add_task("[blue]生成報告...", total=80)
            
            while not progress.finished:
                progress.update(task1, advance=0.5)
                progress.update(task2, advance=1.2)
                progress.update(task3, advance=0.8)
                time.sleep(0.01)
    
    def layout_and_panels(self) -> None:
        """布局和面板演示"""
        
        layout = Layout()
        
        # 分割布局
        layout.split_column(
            Layout(name="header", size=3),
            Layout(name="body"),
            Layout(name="footer", size=3)
        )
        
        # 進一步分割body區(qū)域
        layout["body"].split_row(
            Layout(name="left"),
            Layout(name="right")
        )
        
        # 設(shè)置各區(qū)域內(nèi)容
        layout["header"].update(
            Panel("?? Rich庫演示系統(tǒng)", style="bold blue")
        )
        
        # 左側(cè)內(nèi)容：Markdown
        markdown_content = """
        # Rich Markdown支持
        
        ## 特性列表
        - ? **粗體** 和 *斜體* 文本
        - ? `代碼塊` 支持
        - ? 列表和表格
        - ? 標題層級
        
        ```python
        def hello_rich():
            print("Hello, Rich!")
        ```
        """
        layout["left"].update(
            Panel(Markdown(markdown_content), title="Markdown渲染", style="green")
        )
        
        # 右側(cè)內(nèi)容：語法高亮
        python_code = '''
        def fibonacci(n: int) -> int:
            """計算斐波那契數(shù)列"""
            if n <= 1:
                return n
            a, b = 0, 1
            for _ in range(2, n + 1):
                a, b = b, a + b
            return b
        
        # 使用示例
        result = fibonacci(10)
        print(f"斐波那契(10) = {result}")
        '''
        layout["right"].update(
            Panel(
                Syntax(python_code, "python", theme="monokai", line_numbers=True),
                title="Python代碼高亮",
                style="yellow"
            )
        )
        
        layout["footer"].update(
            Panel("?? 演示完成 - 使用Rich創(chuàng)建美觀的終端界面", style="bold cyan")
        )
        
        self.console.print(layout)
    
    def tree_structure(self) -> None:
        """樹形結(jié)構(gòu)演示"""
        
        tree = Tree("?? 公司組織結(jié)構(gòu)")
        
        # 技術(shù)部門
        tech_branch = tree.add("?? 技術(shù)部")
        backend_branch = tech_branch.add("?? 后端開發(fā)")
        backend_branch.add("Python開發(fā)組")
        backend_branch.add("Java開發(fā)組")
        
        frontend_branch = tech_branch.add("?? 前端開發(fā)")
        frontend_branch.add("React小組")
        frontend_branch.add("Vue小組")
        
        # 市場部門
        market_branch = tree.add("?? 市場部")
        market_branch.add("數(shù)字營銷組")
        market_branch.add("品牌推廣組")
        
        self.console.print(tree)
    
    def data_visualization(self) -> None:
        """簡單的數(shù)據(jù)可視化"""
        
        # 模擬銷售數(shù)據(jù)
        sales_data = {
            "一月": 12000,
            "二月": 15000,
            "三月": 18000,
            "四月": 22000,
            "五月": 19000,
            "六月": 25000
        }
        
        table = Table(title="?? 2024年上半年銷售數(shù)據(jù)", show_header=True)
        table.add_column("月份", style="cyan")
        table.add_column("銷售額", style="green", justify="right")
        table.add_column("進度條", width=20)
        
        max_sales = max(sales_data.values())
        
        for month, sales in sales_data.items():
            # 創(chuàng)建文本進度條
            percentage = sales / max_sales
            bar_length = int(percentage * 15)
            bar = "█" * bar_length + "?" * (15 - bar_length)
            table.add_row(month, f"￥{sales:,}", f"[green]{bar}[/green] {percentage:.1%}")
        
        self.console.print(table)

def main():
    """主函數(shù)"""
    demo = RichDemo()
    
    print("=" * 60)
    print("Rich庫功能演示")
    print("=" * 60)
    
    demo.console.print("\n1. 基礎(chǔ)樣式演示", style="bold underline")
    demo.basic_styling()
    
    demo.console.print("\n2. 高級表格", style="bold underline")
    demo.advanced_tables()
    
    demo.console.print("\n3. 進度條演示", style="bold underline")
    demo.progress_bars_demo()
    
    demo.console.print("\n4. 布局和面板", style="bold underline")
    demo.layout_and_panels()
    
    demo.console.print("\n5. 樹形結(jié)構(gòu)", style="bold underline")
    demo.tree_structure()
    
    demo.console.print("\n6. 數(shù)據(jù)可視化", style="bold underline")
    demo.data_visualization()

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Typer庫演示：構(gòu)建強大的命令行應(yīng)用
"""

import typer
from typing import Optional, List
from pathlib import Path
import json
import csv
from datetime import datetime
from enum import Enum
import requests

# 創(chuàng)建Typer應(yīng)用
app = typer.Typer(
    name="文件管理工具",
    help="一個強大的文件操作和系統(tǒng)管理命令行工具",
    rich_markup_mode="rich"
)

# 子命令組
file_app = typer.Typer(help="文件操作命令")
data_app = typer.Typer(help="數(shù)據(jù)處理命令")
app.add_typer(file_app, name="file")
app.add_typer(data_app, name="data")

class FormatOptions(str, Enum):
    """輸出格式選項"""
    JSON = "json"
    CSV = "csv"
    TABLE = "table"

@app.command()
def hello(
    name: str = typer.Argument(..., help="你的名字"),
    count: int = typer.Option(1, "--count", "-c", help="問候次數(shù)"),
    formal: bool = typer.Option(False, "--formal", help="使用正式問候")
) -> None:
    """
    簡單的問候命令
    
    Args:
        name: 你的名字
        count: 問候次數(shù)
        formal: 是否使用正式問候
    """
    greeting = "您好" if formal else "你好"
    for i in range(count):
        typer.echo(f"{greeting}, [bold blue]{name}[/bold blue]! (#{i+1})")

@file_app.command("info")
def file_info(
    path: Path = typer.Argument(..., help="文件路徑"),
    detailed: bool = typer.Option(False, "--detailed", "-d", help="顯示詳細信息")
) -> None:
    """
    顯示文件信息
    
    Args:
        path: 文件路徑
        detailed: 是否顯示詳細信息
    """
    if not path.exists():
        typer.echo(f"[red]錯誤: 文件 {path} 不存在[/red]")
        raise typer.Exit(code=1)
    
    # 基礎(chǔ)信息
    typer.echo(f"[bold]文件信息: {path}[/bold]")
    typer.echo(f"  大小: {path.stat().st_size:,} 字節(jié)")
    typer.echo(f"  修改時間: {datetime.fromtimestamp(path.stat().st_mtime)}")
    typer.echo(f"  是否文件: {path.is_file()}")
    
    if detailed:
        # 詳細信息
        typer.echo(f"  創(chuàng)建時間: {datetime.fromtimestamp(path.stat().st_ctime)}")
        typer.echo(f"  權(quán)限: {oct(path.stat().st_mode)[-3:]}")
        typer.echo(f"  絕對路徑: {path.absolute()}")

@file_app.command("search")
def search_files(
    directory: Path = typer.Argument(..., help="搜索目錄"),
    pattern: str = typer.Option("*", "--pattern", "-p", help="文件模式"),
    recursive: bool = typer.Option(False, "--recursive", "-r", help="遞歸搜索"),
    max_results: int = typer.Option(50, "--max-results", "-m", help="最大結(jié)果數(shù)")
) -> None:
    """
    搜索文件
    
    Args:
        directory: 搜索目錄
        pattern: 文件匹配模式
        recursive: 是否遞歸搜索
        max_results: 最大結(jié)果數(shù)量
    """
    if not directory.exists():
        typer.echo(f"[red]錯誤: 目錄 {directory} 不存在[/red]")
        raise typer.Exit(code=1)
    
    typer.echo(f"在 [bold blue]{directory}[/bold blue] 中搜索模式 '[green]{pattern}[/green]'")
    
    search_method = directory.rglob if recursive else directory.glob
    files_found = 0
    
    with typer.progressbar(
        search_method(pattern),
        label="搜索文件中...",
        length=min(max_results, 1000)
    ) as progress:
        for file_path in progress:
            if files_found >= max_results:
                break
            if file_path.is_file():
                typer.echo(f"  ?? {file_path.relative_to(directory)} "
                          f"({file_path.stat().st_size:,} bytes)")
                files_found += 1
    
    typer.echo(f"\n[green]找到 {files_found} 個文件[/green]")

@data_app.command("convert")
def convert_data(
    input_file: Path = typer.Argument(..., help="輸入文件"),
    output_format: FormatOptions = typer.Argument(..., help="輸出格式"),
    output_file: Optional[Path] = typer.Option(None, "--output", "-o", help="輸出文件")
) -> None:
    """
    轉(zhuǎn)換數(shù)據(jù)格式
    
    Args:
        input_file: 輸入文件路徑
        output_format: 輸出格式
        output_file: 輸出文件路徑（可選）
    """
    if not input_file.exists():
        typer.echo(f"[red]錯誤: 輸入文件 {input_file} 不存在[/red]")
        raise typer.Exit(code=1)
    
    # 讀取輸入數(shù)據(jù)（這里使用模擬數(shù)據(jù)）
    sample_data = [
        {"name": "張三", "age": 25, "city": "北京"},
        {"name": "李四", "age": 30, "city": "上海"},
        {"name": "王五", "age": 28, "city": "深圳"}
    ]
    
    if output_file is None:
        output_file = input_file.with_suffix(f".{output_format.value}")
    
    # 根據(jù)格式轉(zhuǎn)換數(shù)據(jù)
    if output_format == FormatOptions.JSON:
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(sample_data, f, ensure_ascii=False, indent=2)
    elif output_format == FormatOptions.CSV:
        with open(output_file, 'w', encoding='utf-8', newline='') as f:
            if sample_data:
                writer = csv.DictWriter(f, fieldnames=sample_data[0].keys())
                writer.writeheader()
                writer.writerows(sample_data)
    
    typer.echo(f"[green]成功轉(zhuǎn)換數(shù)據(jù)到 {output_file}[/green]")

@app.command()
def weather(
    city: str = typer.Argument(..., help="城市名稱"),
    units: str = typer.Option("metric", "--units", "-u", 
                             help="溫度單位: metric(攝氏度) 或 imperial(華氏度)")
) -> None:
    """
    獲取天氣信息（模擬）
    
    Args:
        city: 城市名稱
        units: 溫度單位
    """
    # 模擬天氣數(shù)據(jù)
    weather_data = {
        "beijing": {"temp": 25, "humidity": 60, "condition": "晴朗"},
        "shanghai": {"temp": 28, "humidity": 75, "condition": "多云"},
        "shenzhen": {"temp": 30, "humidity": 80, "condition": "陣雨"}
    }
    
    city_key = city.lower()
    if city_key not in weather_data:
        typer.echo(f"[red]錯誤: 未找到城市 {city} 的天氣數(shù)據(jù)[/red]")
        raise typer.Exit(code=1)
    
    data = weather_data[city_key]
    temp_unit = "°C" if units == "metric" else "°F"
    
    # 顯示天氣信息
    typer.echo(f"[bold]???  {city} 天氣[/bold]")
    typer.echo(f"  溫度: [yellow]{data['temp']}{temp_unit}[/yellow]")
    typer.echo(f"  濕度: [blue]{data['humidity']}%[/blue]")
    typer.echo(f"  天氣狀況: [green]{data['condition']}[/green]")

# 回調(diào)函數(shù)示例
@app.callback()
def main_callback(
    verbose: bool = typer.Option(False, "--verbose", "-v", help="詳細輸出"),
    version: bool = typer.Option(False, "--version", help="顯示版本信息")
) -> None:
    """
    文件管理工具 - 強大的命令行文件操作工具
    """
    if version:
        typer.echo("文件管理工具 v1.0.0")
        raise typer.Exit()
    
    if verbose:
        typer.echo("?? 詳細模式已啟用")

# 錯誤處理
@app.command()
def divide(
    numerator: float = typer.Argument(..., help="被除數(shù)"),
    denominator: float = typer.Argument(..., help="除數(shù)")
) -> None:
    """
    除法運算（演示錯誤處理）
    
    Args:
        numerator: 被除數(shù)
        denominator: 除數(shù)
    """
    try:
        result = numerator / denominator
        typer.echo(f"[green]結(jié)果: {numerator} / {denominator} = {result:.2f}[/green]")
    except ZeroDivisionError:
        typer.echo("[red]錯誤: 除數(shù)不能為零[/red]")
        raise typer.Exit(code=1)

def run_demo():
    """運行演示"""
    typer.echo("=" * 60)
    typer.echo("Typer命令行工具演示")
    typer.echo("=" * 60)
    typer.echo("\n嘗試運行以下命令體驗功能:")
    typer.echo("  python typer_demo.py hello 張三 --count 3")
    typer.echo("  python typer_demo.py file info . --detailed")
    typer.echo("  python typer_demo.py data convert data.txt json")
    typer.echo("  python typer_demo.py weather 北京")
    typer.echo("\n使用 --help 查看完整命令列表")

if __name__ == "__main__":
    if len(typer.get_current_context().args) == 0:
        run_demo()
    else:
        app()

#!/usr/bin/env python3
"""
Textual庫演示：構(gòu)建終端用戶界面
注意：運行此代碼需要安裝textual庫: pip install textual
"""

from textual.app import App, ComposeResult
from textual.widgets import (
    Header, Footer, Button, Static, Input, 
    Select, Label, DataTable, RichLog, ListView, ListItem
)
from textual.containers import Container, Horizontal, VerticalScroll
from textual.reactive import reactive
from textual.screen import Screen
from textual import events
from typing import List, Dict, Any
import asyncio
from datetime import datetime
import json

class WelcomeScreen(Screen):
    """歡迎屏幕"""
    
    def compose(self) -> ComposeResult:
        yield Container(
            Static("?? Textual 演示應(yīng)用", classes="welcome-title"),
            Static("一個強大的終端用戶界面庫", classes="welcome-subtitle"),
            Button("開始探索", variant="primary", id="start-btn"),
            classes="welcome-container"
        )
    
    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "start-btn":
            self.app.push_screen(MainScreen())

class MainScreen(Screen):
    """主屏幕"""
    
    current_time = reactive(datetime.now())
    
    def compose(self) -> ComposeResult:
        yield Header()
        yield Container(
            Horizontal(
                VerticalScroll(
                    Static("?? 控制面板", classes="panel-title"),
                    Input(placeholder="搜索...", id="search-input"),
                    Select(
                        [(f"選項 {i}", str(i)) for i in range(1, 6)],
                        prompt="選擇選項",
                        id="demo-select"
                    ),
                    Button("刷新數(shù)據(jù)", variant="success", id="refresh-btn"),
                    Button("顯示日志", variant="warning", id="log-btn"),
                    Button("返回", variant="error", id="back-btn"),
                    classes="control-panel"
                ),
                VerticalScroll(
                    Static("?? 數(shù)據(jù)表格", classes="panel-title"),
                    DataTable(id="data-table"),
                    Static("\n?? 系統(tǒng)信息", classes="panel-title"),
                    Label("", id="time-label"),
                    Label("", id="status-label"),
                    classes="data-panel"
                ),
            ),
            RichLog(highlight=True, max_lines=10, id="rich-log"),
            classes="main-container"
        )
        yield Footer()
    
    def on_mount(self) -> None:
        # 初始化數(shù)據(jù)表格
        table = self.query_one("#data-table", DataTable)
        table.add_columns("ID", "名稱", "狀態(tài)", "數(shù)值")
        
        # 添加示例數(shù)據(jù)
        sample_data = [
            ("1", "任務(wù)A", "進行中", "75%"),
            ("2", "任務(wù)B", "已完成", "100%"),
            ("3", "任務(wù)C", "待開始", "0%"),
            ("4", "任務(wù)D", "進行中", "50%"),
            ("5", "任務(wù)E", "已取消", "25%"),
        ]
        
        for row in sample_data:
            table.add_row(*row)
        
        # 啟動定時器更新時鐘
        self.set_interval(1, self.update_time)
    
    def update_time(self) -> None:
        """更新時間顯示"""
        self.current_time = datetime.now()
        time_label = self.query_one("#time-label", Label)
        time_label.update(f"當前時間: {self.current_time.strftime('%Y-%m-%d %H:%M:%S')}")
    
    def on_button_pressed(self, event: Button.Pressed) -> None:
        log = self.query_one("#rich-log", RichLog)
        
        if event.button.id == "refresh-btn":
            log.write("?? 刷新數(shù)據(jù)...")
            self.refresh_data()
        elif event.button.id == "log-btn":
            log.write("?? 顯示日志信息")
            self.demo_logging()
        elif event.button.id == "back-btn":
            self.app.pop_screen()
    
    def refresh_data(self) -> None:
        """刷新數(shù)據(jù)演示"""
        table = self.query_one("#data-table", DataTable)
        status_label = self.query_one("#status-label", Label)
        
        # 模擬數(shù)據(jù)更新
        import random
        new_value = random.randint(1, 100)
        status_label.update(f"最后更新: {datetime.now().strftime('%H:%M:%S')} | 隨機值: {new_value}")
        
        log = self.query_one("#rich-log", RichLog)
        log.write(f"? 數(shù)據(jù)已刷新，新值: {new_value}")
    
    def demo_logging(self) -> None:
        """日志記錄演示"""
        log = self.query_one("#rich-log", RichLog)
        
        log.write("?? 開始記錄日志...")
        log.write("   這是一條信息日志")
        log.write("   [yellow]這是一條警告日志[/yellow]")
        log.write("   [red]這是一條錯誤日志[/red]")
        log.write("? 日志記錄完成")

class TextualDemoApp(App):
    """Textual演示應(yīng)用"""
    
    CSS = """
    Screen {
        background: $surface;
    }
    
    .welcome-container {
        width: 100%;
        height: 100%;
        align: center middle;
    }
    
    .welcome-title {
        text-align: center;
        color: $accent;
        text-style: bold;
        margin-bottom: 1;
    }
    
    .welcome-subtitle {
        text-align: center;
        color: $text-muted;
        margin-bottom: 2;
    }
    
    .main-container {
        padding: 1;
    }
    
    .control-panel {
        width: 30%;
        padding: 1;
        border: solid $accent;
        margin-right: 1;
    }
    
    .data-panel {
        width: 70%;
        padding: 1;
        border: solid $boost;
    }
    
    .panel-title {
        color: $accent;
        text-style: bold;
        margin-bottom: 1;
    }
    
    Button {
        margin: 1 0;
    }
    
    DataTable {
        height: 12;
    }
    
    RichLog {
        height: 10;
        border: solid $secondary;
        margin-top: 1;
        padding: 1;
    }
    """
    
    BINDINGS = [
        ("d", "toggle_dark", "切換暗色模式"),
        ("q", "quit", "退出應(yīng)用"),
        ("r", "refresh_data", "刷新數(shù)據(jù)"),
    ]
    
    def on_mount(self) -> None:
        self.push_screen(WelcomeScreen())
    
    def action_toggle_dark(self) -> None:
        self.dark = not self.dark
    
    def action_refresh_data(self) -> None:
        current_screen = self.screen
        if isinstance(current_screen, MainScreen):
            current_screen.refresh_data()

def run_textual_demo():
    """運行Textual演示"""
    print("啟動Textual演示應(yīng)用...")
    print("如果出現(xiàn)導(dǎo)入錯誤，請安裝: pip install textual")
    
    try:
        app = TextualDemoApp()
        app.run()
    except ImportError as e:
        print(f"錯誤: 需要安裝Textual庫 - {e}")
        print("請運行: pip install textual")

if __name__ == "__main__":
    run_textual_demo()

#!/usr/bin/env python3
"""
Polars庫演示：高性能DataFrame操作
"""

import polars as pl
import numpy as np
import time
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple
import sys

class PolarsDemo:
    """
    Polars庫功能演示
    """
    
    def __init__(self):
        self.df = None
    
    def create_sample_data(self, num_rows: int = 100000) -> pl.DataFrame:
        """創(chuàng)建示例數(shù)據(jù)集"""
        np.random.seed(42)
        
        # 生成日期范圍
        start_date = datetime(2020, 1, 1)
        dates = [start_date + timedelta(days=x) for x in range(num_rows)]
        
        # 創(chuàng)建DataFrame
        self.df = pl.DataFrame({
            "date": dates,
            "category": np.random.choice(["A", "B", "C", "D"], num_rows),
            "value1": np.random.normal(100, 15, num_rows),
            "value2": np.random.exponential(2, num_rows),
            "value3": np.random.randint(1, 1000, num_rows),
            "group": np.random.choice(["X", "Y", "Z"], num_rows)
        })
        
        print(f"創(chuàng)建數(shù)據(jù)集: {num_rows:,} 行")
        print(f"內(nèi)存使用: {self.df.estimated_size() / 1024 / 1024:.2f} MB")
        return self.df
    
    def basic_operations(self) -> None:
        """基礎(chǔ)操作演示"""
        print("\n" + "="*50)
        print("基礎(chǔ)操作演示")
        print("="*50)
        
        # 顯示基本信息
        print("DataFrame基本信息:")
        print(f"  形狀: {self.df.shape}")
        print(f"  列名: {self.df.columns}")
        print(f"  數(shù)據(jù)類型: {self.df.dtypes}")
        
        # 數(shù)據(jù)預(yù)覽
        print("\n數(shù)據(jù)預(yù)覽:")
        print(self.df.head(8))
        
        # 描述性統(tǒng)計
        print("\n描述性統(tǒng)計:")
        print(self.df.describe())
    
    def lazy_evaluation_demo(self) -> None:
        """惰性評估演示"""
        print("\n" + "="*50)
        print("惰性評估演示")
        print("="*50)
        
        # 急切執(zhí)行（立即計算）
        start_time = time.time()
        eager_result = (
            self.df
            .filter(pl.col("value1") > 100)
            .group_by("category")
            .agg([
                pl.col("value2").mean().alias("avg_value2"),
                pl.col("value3").sum().alias("total_value3")
            ])
            .sort("avg_value2", descending=True)
        )
        eager_time = time.time() - start_time
        
        # 惰性執(zhí)行（優(yōu)化后計算）
        start_time = time.time()
        lazy_result = (
            self.df
            .lazy()
            .filter(pl.col("value1") > 100)
            .group_by("category")
            .agg([
                pl.col("value2").mean().alias("avg_value2"),
                pl.col("value3").sum().alias("total_value3")
            ])
            .sort("avg_value2", descending=True)
            .collect()  # 實際執(zhí)行
        )
        lazy_time = time.time() - start_time
        
        print("急切執(zhí)行結(jié)果:")
        print(eager_result)
        print(f"\n急切執(zhí)行時間: {eager_time:.4f}秒")
        
        print("\n惰性執(zhí)行結(jié)果:")
        print(lazy_result)
        print(f"惰性執(zhí)行時間: {lazy_time:.4f}秒")
        
        print(f"\n性能提升: {eager_time/lazy_time:.2f}x")
    
    def advanced_operations(self) -> None:
        """高級操作演示"""
        print("\n" + "="*50)
        print("高級操作演示")
        print("="*50)
        
        # 復(fù)雜數(shù)據(jù)轉(zhuǎn)換
        result = (
            self.df
            .lazy()
            .with_columns([
                # 創(chuàng)建新列
                (pl.col("value1") * pl.col("value2")).alias("product"),
                pl.col("date").dt.year().alias("year"),
                pl.col("date").dt.month().alias("month"),
                
                # 條件列
                pl.when(pl.col("value1") > 100)
                 .then(pl.lit("high"))
                 .otherwise(pl.lit("low"))
                 .alias("value1_category"),
                
                # 窗口函數(shù)
                pl.col("value3").sort_by("date").over("category").alias("sorted_value3")
            ])
            .filter(
                (pl.col("year") == 2020) &
                (pl.col("value1_category") == "high")
            )
            .group_by(["category", "year", "month"])
            .agg([
                pl.col("value1").mean().alias("avg_value1"),
                pl.col("value2").std().alias("std_value2"),
                pl.col("product").sum().alias("total_product"),
                pl.col("value3").count().alias("record_count")
            ])
            .sort(["category", "year", "month"])
            .collect()
        )
        
        print("復(fù)雜數(shù)據(jù)轉(zhuǎn)換結(jié)果:")
        print(result)
    
    def performance_comparison(self) -> Dict[str, float]:
        """性能對比測試"""
        print("\n" + "="*50)
        print("性能對比測試 (Polars vs Pandas)")
        print("="*50)
        
        try:
            import pandas as pd
            
            # 創(chuàng)建相同的數(shù)據(jù)
            polars_df = self.df
            pandas_df = polars_df.to_pandas()
            
            # 測試操作：分組聚合
            operations = {
                "簡單分組聚合": lambda df: df.groupby("category")["value1"].mean(),
                "多列聚合": lambda df: df.groupby("category").agg({
                    "value1": ["mean", "std"],
                    "value2": ["sum", "count"]
                }),
                "條件過濾+聚合": lambda df: df[df["value1"] > 100].groupby("category")["value2"].mean()
            }
            
            results = {}
            
            for op_name, operation in operations.items():
                # Polars性能
                start_time = time.time()
                if "polars" in op_name.lower():
                    # 對于polars使用惰性執(zhí)行
                    result_pl = (
                        polars_df.lazy()
                        .filter(pl.col("value1") > 100)
                        .group_by("category")
                        .agg(pl.col("value2").mean())
                        .collect()
                    )
                else:
                    # 實際執(zhí)行操作
                    if op_name == "簡單分組聚合":
                        result_pl = polars_df.group_by("category").agg(pl.col("value1").mean())
                    elif op_name == "多列聚合":
                        result_pl = polars_df.group_by("category").agg([
                            pl.col("value1").mean().alias("value1_mean"),
                            pl.col("value1").std().alias("value1_std"),
                            pl.col("value2").sum().alias("value2_sum"),
                            pl.col("value2").count().alias("value2_count")
                        ])
                polars_time = time.time() - start_time
                
                # Pandas性能
                start_time = time.time()
                result_pd = operation(pandas_df)
                pandas_time = time.time() - start_time
                
                results[op_name] = {
                    "polars_time": polars_time,
                    "pandas_time": pandas_time,
                    "speedup": pandas_time / polars_time if polars_time > 0 else 0
                }
                
                print(f"\n{op_name}:")
                print(f"  Polars: {polars_time:.4f}秒")
                print(f"  Pandas: {pandas_time:.4f}秒")
                print(f"  速度提升: {results[op_name]['speedup']:.2f}x")
            
            return results
            
        except ImportError:
            print("Pandas未安裝，跳過性能對比")
            return {}
    
    def data_processing_pipeline(self) -> None:
        """完整的數(shù)據(jù)處理管道"""
        print("\n" + "="*50)
        print("完整數(shù)據(jù)處理管道")
        print("="*50)
        
        # 模擬真實的數(shù)據(jù)處理流程
        pipeline_result = (
            self.df
            .lazy()
            .pipe(self._clean_data)
            .pipe(self._enrich_data)
            .pipe(self._analyze_trends)
            .collect()
        )
        
        print("管道處理結(jié)果:")
        print(pipeline_result)
    
    def _clean_data(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """數(shù)據(jù)清洗步驟"""
        return (
            df
            .filter(
                pl.col("value1").is_between(0, 200) &  # 移除異常值
                pl.col("value2") > 0 &                 # 確保正值
                pl.col("value3").is_not_null()         # 移除空值
            )
            .unique(subset=["date", "category"])       # 去重
        )
    
    def _enrich_data(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """數(shù)據(jù)增強步驟"""
        return (
            df
            .with_columns([
                pl.col("date").dt.strftime("%Y-%m").alias("year_month"),
                (pl.col("value1") / pl.col("value2")).alias("ratio"),
                pl.col("value3").log().alias("log_value3"),
                
                # 創(chuàng)建標志列
                pl.when(pl.col("value1") > pl.col("value1").mean())
                 .then(pl.lit(1))
                 .otherwise(pl.lit(0))
                 .alias("above_avg")
            ])
        )
    
    def _analyze_trends(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """趨勢分析步驟"""
        return (
            df
            .group_by(["category", "year_month"])
            .agg([
                pl.col("value1").mean().alias("avg_value1"),
                pl.col("value2").sum().alias("total_value2"),
                pl.col("ratio").median().alias("median_ratio"),
                pl.col("above_avg").mean().alias("pct_above_avg"),
                pl.col("value3").count().alias("record_count")
            ])
            .sort(["category", "year_month"])
            .with_columns([
                # 計算月度變化
                pl.col("avg_value1").pct_change().over("category").alias("value1_monthly_change"),
                pl.col("total_value2").diff().over("category").alias("value2_monthly_diff")
            ])
        )

def main():
    """主函數(shù)"""
    demo = PolarsDemo()
    
    print("Polars高性能DataFrame庫演示")
    print("=" * 60)
    
    # 創(chuàng)建示例數(shù)據(jù)
    demo.create_sample_data(50000)
    
    # 運行各種演示
    demo.basic_operations()
    demo.lazy_evaluation_demo()
    demo.advanced_operations()
    demo.performance_comparison()
    demo.data_processing_pipeline()
    
    print("\n" + "="*60)
    print("演示完成！")
    print("Polars提供了:")
    print("  ? 驚人的性能表現(xiàn)")
    print("  ? 惰性評估優(yōu)化")
    print("  ? 簡潔的API設(shè)計") 
    print("  ? 強大的并行處理")
    print("  ? 出色的內(nèi)存效率")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
庫集成示例：結(jié)合使用多個有用庫
"""

from loguru import logger
from rich.console import Console
from rich.table import Table
from rich.progress import track
import typer
import polars as pl
from typing import Optional
import time

# 初始化
console = Console()
app = typer.Typer(help="集成演示應(yīng)用")

@app.command()
def analyze_data(
    file_path: str = typer.Argument(..., help="數(shù)據(jù)文件路徑"),
    output_format: str = typer.Option("table", "--format", "-f", help="輸出格式")
) -> None:
    """
    數(shù)據(jù)分析命令，集成多個庫的功能
    """
    logger.info("開始數(shù)據(jù)分析")
    
    try:
        # 使用Polars讀取數(shù)據(jù)
        with console.status("[bold green]加載數(shù)據(jù)..."):
            df = pl.read_csv(file_path)
            logger.success(f"成功加載數(shù)據(jù): {df.shape}")
        
        # 數(shù)據(jù)分析
        with console.status("[bold blue]分析數(shù)據(jù)..."):
            analysis_result = perform_analysis(df)
            logger.info("數(shù)據(jù)分析完成")
        
        # 使用Rich顯示結(jié)果
        if output_format == "table":
            display_rich_table(analysis_result)
        else:
            console.print(analysis_result)
            
        logger.success("分析流程完成")
        
    except Exception as e:
        logger.error(f"分析失敗: {e}")
        raise typer.Exit(code=1)

def perform_analysis(df: pl.DataFrame) -> pl.DataFrame:
    """執(zhí)行數(shù)據(jù)分析"""
    return (
        df.lazy()
        .group_by("category")
        .agg([
            pl.col("value").mean().alias("平均值"),
            pl.col("value").std().alias("標準差"),
            pl.col("value").count().alias("計數(shù)")
        ])
        .sort("平均值", descending=True)
        .collect()
    )

def display_rich_table(df: pl.DataFrame) -> None:
    """使用Rich顯示表格"""
    table = Table(title="?? 數(shù)據(jù)分析結(jié)果", show_header=True)
    
    # 添加列
    for col in df.columns:
        table.add_column(col, style="cyan")
    
    # 添加行
    for row in df.rows():
        table.add_row(*[str(x) for x in row])
    
    console.print(table)

if __name__ == "__main__":
    console.print("?? 集成庫演示應(yīng)用", style="bold blue")
    console.print("=" * 50)
    
    # 模擬數(shù)據(jù)創(chuàng)建和分析
    sample_data = pl.DataFrame({
        "category": ["A", "B", "A", "C", "B", "C", "A"],
        "value": [100, 150, 120, 80, 200, 90, 110]
    })
    
    console.print("\n示例數(shù)據(jù)分析:")
    display_rich_table(perform_analysis(sample_data))
    
    console.print("\n?? 使用 --help 查看完整命令選項")

pip install loguru rich typer textual polars