#!/usr/bin/env python3
"""
Loguru庫(kù)演示：簡(jiǎn)化Python日志記錄
"""

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

class LoguruDemo:
    """
    Loguru功能演示類
    """
    
    def __init__(self):
        """初始化日志配置"""
        # 移除默認(rèn)配置，添加自定義配置
        logger.remove()
        
        # 添加控制臺(tái)輸出，帶有顏色和格式
        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
        )
        
        # 添加文件輸出，自動(dòng)輪轉(zhuǎn)和壓縮
        logger.add(
            "logs/demo_{time}.log",
            rotation="10 MB",  # 文件達(dá)到10MB時(shí)輪轉(zhuǎn)
            compression="zip",  # 壓縮舊日志
            retention="30 days",  # 保留30天
            level="INFO"
        )
    
    def basic_usage(self) -> None:
        """基礎(chǔ)用法演示"""
        logger.debug("這是一條調(diào)試信息")
        logger.info("程序正常運(yùn)行")
        logger.warning("這是一個(gè)警告")
        logger.error("發(fā)生了一個(gè)錯(cuò)誤")
        logger.critical("嚴(yán)重錯(cuò)誤！")
    
    def advanced_features(self) -> None:
        """高級(jí)功能演示"""
        
        # 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("計(jì)算成功: {} / {} = {}", x, y, result)
            return result
        
        # 3. 性能監(jiān)控
        with logger.catch(message="處理用戶數(shù)據(jù)時(shí)發(fā)生錯(cuò)誤"):
            for i in range(3):
                start_time = time.time()
                time.sleep(0.1)  # 模擬工作負(fù)載
                elapsed = time.time() - start_time
                logger.info("操作 {} 耗時(shí) {:.3f} 秒", i, elapsed)
        
        # 測(cè)試異常捕獲
        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("收到用戶請(qǐng)求")
        contextual_logger.bind(action="login").info("用戶登錄")
        contextual_logger.bind(action="logout").info("用戶登出")
    
    def performance_comparison(self) -> Dict[str, Any]:
        """性能對(duì)比測(cè)試"""
        import logging as std_logging
        
        # 標(biāo)準(zhǔn)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)
        
        # 性能測(cè)試
        test_iterations = 1000
        
        # Loguru性能
        loguru_start = time.time()
        for i in range(test_iterations):
            logger.debug("測(cè)試消息 {}", i)
        loguru_time = time.time() - loguru_start
        
        # 標(biāo)準(zhǔn)logging性能
        std_start = time.time()
        for i in range(test_iterations):
            std_logger.debug("測(cè)試消息 %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. 高級(jí)功能:")
    demo.advanced_features()
    
    print("\n3. 上下文感知日志:")
    demo.context_aware_logging()
    
    print("\n4. 性能對(duì)比:")
    results = demo.performance_comparison()
    print(f"   測(cè)試迭代次數(shù): {results['iterations']}")
    print(f"   Loguru耗時(shí): {results['loguru_time']:.4f}秒")
    print(f"   標(biāo)準(zhǔn)logging耗時(shí): {results['std_logging_time']:.4f}秒")
    print(f"   Loguru快 {results['loguru_faster']:.2f} 倍")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Rich庫(kù)演示：創(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庫(kù)功能演示
    """
    
    def __init__(self):
        self.console = Console()
    
    def basic_styling(self) -> None:
        """基礎(chǔ)樣式演示"""
        self.console.print("[bold red]紅色粗體文本[/bold red]")
        self.console.print("[italic blue]藍(lán)色斜體文本[/italic blue]")
        self.console.print("[underline green]綠色下劃線文本[/underline green]")
        self.console.print("[reverse yellow]黃色反色文本[/reverse yellow]")
        self.console.print(":smiley: :snake: :rocket: 表情符號(hào)支持")
    
    def advanced_tables(self) -> None:
        """高級(jí)表格功能"""
        
        # 創(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:
        """進(jìn)度條演示"""
        
        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
            console=self.console
        ) as progress:
            
            # 模擬多個(gè)任務(wù)
            task1 = progress.add_task("[red]下載文件...", total=100)
            task2 = progress.add_task("[green]處理數(shù)據(jù)...", total=150)
            task3 = progress.add_task("[blue]生成報(bào)告...", 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)
        )
        
        # 進(jìn)一步分割body區(qū)域
        layout["body"].split_row(
            Layout(name="left"),
            Layout(name="right")
        )
        
        # 設(shè)置各區(qū)域內(nèi)容
        layout["header"].update(
            Panel("?? Rich庫(kù)演示系統(tǒng)", style="bold blue")
        )
        
        # 左側(cè)內(nèi)容：Markdown
        markdown_content = """
        # Rich Markdown支持
        
        ## 特性列表
        - ? **粗體** 和 *斜體* 文本
        - ? `代碼塊` 支持
        - ? 列表和表格
        - ? 標(biāo)題層級(jí)
        
        ```python
        def hello_rich():
            print("Hello, Rich!")
        ```
        """
        layout["left"].update(
            Panel(Markdown(markdown_content), title="Markdown渲染", style="green")
        )
        
        # 右側(cè)內(nèi)容：語(yǔ)法高亮
        python_code = '''
        def fibonacci(n: int) -> int:
            """計(jì)算斐波那契數(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小組")
        
        # 市場(chǎng)部門
        market_branch = tree.add("?? 市場(chǎng)部")
        market_branch.add("數(shù)字營(yíng)銷組")
        market_branch.add("品牌推廣組")
        
        self.console.print(tree)
    
    def data_visualization(self) -> None:
        """簡(jiǎn)單的數(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("進(jìn)度條", width=20)
        
        max_sales = max(sales_data.values())
        
        for month, sales in sales_data.items():
            # 創(chuàng)建文本進(jìn)度條
            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庫(kù)功能演示")
    print("=" * 60)
    
    demo.console.print("\n1. 基礎(chǔ)樣式演示", style="bold underline")
    demo.basic_styling()
    
    demo.console.print("\n2. 高級(jí)表格", style="bold underline")
    demo.advanced_tables()
    
    demo.console.print("\n3. 進(jìn)度條演示", 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庫(kù)演示：構(gòu)建強(qiáng)大的命令行應(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="一個(gè)強(qiáng)大的文件操作和系統(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):
    """輸出格式選項(xiàng)"""
    JSON = "json"
    CSV = "csv"
    TABLE = "table"

@app.command()
def hello(
    name: str = typer.Argument(..., help="你的名字"),
    count: int = typer.Option(1, "--count", "-c", help="問(wèn)候次數(shù)"),
    formal: bool = typer.Option(False, "--formal", help="使用正式問(wèn)候")
) -> None:
    """
    簡(jiǎn)單的問(wèn)候命令
    
    Args:
        name: 你的名字
        count: 問(wèn)候次數(shù)
        formal: 是否使用正式問(wèn)候
    """
    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="顯示詳細(xì)信息")
) -> None:
    """
    顯示文件信息
    
    Args:
        path: 文件路徑
        detailed: 是否顯示詳細(xì)信息
    """
    if not path.exists():
        typer.echo(f"[red]錯(cuò)誤: 文件 {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"  修改時(shí)間: {datetime.fromtimestamp(path.stat().st_mtime)}")
    typer.echo(f"  是否文件: {path.is_file()}")
    
    if detailed:
        # 詳細(xì)信息
        typer.echo(f"  創(chuàng)建時(shí)間: {datetime.fromtimestamp(path.stat().st_ctime)}")
        typer.echo(f"  權(quán)限: {oct(path.stat().st_mode)[-3:]}")
        typer.echo(f"  絕對(duì)路徑: {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]錯(cuò)誤: 目錄 {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} 個(gè)文件[/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]錯(cuò)誤: 輸入文件 {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]錯(cuò)誤: 未找到城市 {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="詳細(xì)輸出"),
    version: bool = typer.Option(False, "--version", help="顯示版本信息")
) -> None:
    """
    文件管理工具 - 強(qiáng)大的命令行文件操作工具
    """
    if version:
        typer.echo("文件管理工具 v1.0.0")
        raise typer.Exit()
    
    if verbose:
        typer.echo("?? 詳細(xì)模式已啟用")

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

def run_demo():
    """運(yùn)行演示"""
    typer.echo("=" * 60)
    typer.echo("Typer命令行工具演示")
    typer.echo("=" * 60)
    typer.echo("\n嘗試運(yùn)行以下命令體驗(yà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庫(kù)演示：構(gòu)建終端用戶界面
注意：運(yùn)行此代碼需要安裝textual庫(kù): 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("一個(gè)強(qiáng)大的終端用戶界面庫(kù)", 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"選項(xiàng) {i}", str(i)) for i in range(1, 6)],
                        prompt="選擇選項(xiàng)",
                        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", "進(jìn)行中", "75%"),
            ("2", "任務(wù)B", "已完成", "100%"),
            ("3", "任務(wù)C", "待開始", "0%"),
            ("4", "任務(wù)D", "進(jìn)行中", "50%"),
            ("5", "任務(wù)E", "已取消", "25%"),
        ]
        
        for row in sample_data:
            table.add_row(*row)
        
        # 啟動(dòng)定時(shí)器更新時(shí)鐘
        self.set_interval(1, self.update_time)
    
    def update_time(self) -> None:
        """更新時(shí)間顯示"""
        self.current_time = datetime.now()
        time_label = self.query_one("#time-label", Label)
        time_label.update(f"當(dāng)前時(shí)間: {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')} | 隨機(jī)值: {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]這是一條錯(cuò)誤日志[/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():
    """運(yùn)行Textual演示"""
    print("啟動(dòng)Textual演示應(yīng)用...")
    print("如果出現(xiàn)導(dǎo)入錯(cuò)誤，請(qǐng)安裝: pip install textual")
    
    try:
        app = TextualDemoApp()
        app.run()
    except ImportError as e:
        print(f"錯(cuò)誤: 需要安裝Textual庫(kù) - {e}")
        print("請(qǐng)運(yùn)行: pip install textual")

if __name__ == "__main__":
    run_textual_demo()

#!/usr/bin/env python3
"""
Polars庫(kù)演示：高性能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庫(kù)功能演示
    """
    
    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)計(jì)
        print("\n描述性統(tǒng)計(jì):")
        print(self.df.describe())
    
    def lazy_evaluation_demo(self) -> None:
        """惰性評(píng)估演示"""
        print("\n" + "="*50)
        print("惰性評(píng)估演示")
        print("="*50)
        
        # 急切執(zhí)行（立即計(jì)算）
        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)化后計(jì)算）
        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()  # 實(shí)際執(zhí)行
        )
        lazy_time = time.time() - start_time
        
        print("急切執(zhí)行結(jié)果:")
        print(eager_result)
        print(f"\n急切執(zhí)行時(shí)間: {eager_time:.4f}秒")
        
        print("\n惰性執(zhí)行結(jié)果:")
        print(lazy_result)
        print(f"惰性執(zhí)行時(shí)間: {lazy_time:.4f}秒")
        
        print(f"\n性能提升: {eager_time/lazy_time:.2f}x")
    
    def advanced_operations(self) -> None:
        """高級(jí)操作演示"""
        print("\n" + "="*50)
        print("高級(jí)操作演示")
        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]:
        """性能對(duì)比測(cè)試"""
        print("\n" + "="*50)
        print("性能對(duì)比測(cè)試 (Polars vs Pandas)")
        print("="*50)
        
        try:
            import pandas as pd
            
            # 創(chuàng)建相同的數(shù)據(jù)
            polars_df = self.df
            pandas_df = polars_df.to_pandas()
            
            # 測(cè)試操作：分組聚合
            operations = {
                "簡(jiǎn)單分組聚合": lambda df: df.groupby("category")["value1"].mean(),
                "多列聚合": lambda df: df.groupby("category").agg({
                    "value1": ["mean", "std"],
                    "value2": ["sum", "count"]
                }),
                "條件過(guò)濾+聚合": 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():
                    # 對(duì)于polars使用惰性執(zhí)行
                    result_pl = (
                        polars_df.lazy()
                        .filter(pl.col("value1") > 100)
                        .group_by("category")
                        .agg(pl.col("value2").mean())
                        .collect()
                    )
                else:
                    # 實(shí)際執(zhí)行操作
                    if op_name == "簡(jiǎn)單分組聚合":
                        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未安裝，跳過(guò)性能對(duì)比")
            return {}
    
    def data_processing_pipeline(self) -> None:
        """完整的數(shù)據(jù)處理管道"""
        print("\n" + "="*50)
        print("完整數(shù)據(jù)處理管道")
        print("="*50)
        
        # 模擬真實(shí)的數(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ù)增強(qiáng)步驟"""
        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)建標(biāo)志列
                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:
        """趨勢(shì)分析步驟"""
        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([
                # 計(jì)算月度變化
                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庫(kù)演示")
    print("=" * 60)
    
    # 創(chuàng)建示例數(shù)據(jù)
    demo.create_sample_data(50000)
    
    # 運(yùn)行各種演示
    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("  ? 惰性評(píng)估優(yōu)化")
    print("  ? 簡(jiǎn)潔的API設(shè)計(jì)") 
    print("  ? 強(qiáng)大的并行處理")
    print("  ? 出色的內(nèi)存效率")

if __name__ == "__main__":
    main()

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

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ù)分析命令，集成多個(gè)庫(kù)的功能
    """
    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("標(biāo)準(zhǔn)差"),
            pl.col("value").count().alias("計(jì)數(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("?? 集成庫(kù)演示應(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 查看完整命令選項(xiàng)")

pip install loguru rich typer textual polars