# 變量與運(yùn)算
price = 19.95
quantity = 3
total = price * quantity
print(f"總價: {total}")  # 輸出: 總價: 59.85
# 類型轉(zhuǎn)換與檢查
num_str = "123"
num_int = int(num_str)
print(isinstance(num_int, int))  # 輸出: True

# 列表操作
fruits = ['apple', 'banana']
fruits.append('orange')  # 增加
fruits[1] = 'grape'      # 修改
last_fruit = fruits.pop() # 刪除并返回最后一個元素
print(fruits)  # 輸出: ['apple', 'grape']
# 字典操作
scores = {'Math': 85, 'English': 92}
scores['Science'] = 88   # 增加鍵值對
print(scores['Math'])    # 通過key訪問: 85
for subject, score in scores.items(): # 遍歷
    print(f"{subject}: {score}")

# 條件與循環(huán)處理考試成績
grade_dict = {'張三': 78, '李四': 92, '王五': 58}
for name, score in grade_dict.items():
    if score >= 90:
        level = '優(yōu)秀'
    elif score >= 60:
        level = '及格'
    else:
        level = '不及格'
    print(f"{name}: {score}分 -> {level}")
# 異常處理
try:
    user_input = int(input("請輸入一個數(shù)字: "))
    result = 100 / user_input
except ValueError:
    print("輸入的不是有效數(shù)字！")
except ZeroDivisionError:
    print("除數(shù)不能為零！")
else:
    print(f"結(jié)果是: {result}")

# 定義帶默認(rèn)參數(shù)和類型提示的函數(shù)
def greet(name: str, greeting: str = "Hello") -> str:
    """返回一個問候字符串。"""
    return f"{greeting}, {name}!"
print(greet("Alice"))  # 使用默認(rèn)參數(shù)
print(greet("Bob", "Hi"))  # 提供參數(shù)
# 可變參數(shù)示例：計算任意個數(shù)的和
def dynamic_sum(*args, **kwargs):
    normal_sum = sum(args)
    print(f"位置參數(shù)和: {normal_sum}")
    print(f"關(guān)鍵字參數(shù): {kwargs}")
dynamic_sum(1, 2, 3, name='Tom', age=10)

class Animal:
    def __init__(self, name):  # 構(gòu)造方法
        self.name = name  # 實例變量
    def speak(self):  # 方法
        raise NotImplementedError("子類必須實現(xiàn)此方法")
class Dog(Animal):  # 繼承
    def speak(self):  # 多態(tài)：重寫父類方法
        return f"{self.name} says: Woof!"
my_dog = Dog("Buddy")
print(my_dog.speak())  # 輸出: Buddy says: Woof!

# 使用with安全地讀寫文件
with open('note.txt', 'w', encoding='utf-8') as f:
    f.write("Hello, World!\n這是第二行。")
with open('note.txt', 'r', encoding='utf-8') as f:
    content = f.read()
    print(content)
# 處理JSON數(shù)據(jù)（常用于API和配置）
import json
data = {'name': '小明', 'hobbies': ['閱讀', '游泳']}
json_str = json.dumps(data, ensure_ascii=False)  # 轉(zhuǎn)為JSON字符串
loaded_data = json.loads(json_str)  # 解析JSON字符串

def timer(func):
    import time
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 執(zhí)行耗時: {end-start:.2f}秒")
        return result
    return wrapper
@timer
def some_task():
    time.sleep(1)
some_task()

def fibonacci(limit):
    a, b = 0, 1
    while a < limit:
        yield a  # 每次產(chǎn)生一個值，函數(shù)狀態(tài)會暫停保留
        a, b = b, a + b
for num in fibonacci(10):
    print(num)  # 輸出: 0 1 1 2 3 5 8

# 變量命名規(guī)范
name = "Python"          # 字符串
version = 3.9           # 整數(shù)
rating = 9.5           # 浮點(diǎn)數(shù)
is_awesome = True      # 布爾值
# 多重賦值
x, y, z = 1, 2, 3
# 鏈?zhǔn)劫x值
a = b = c = 100
# 變量交換（Python特有）
m, n = 10, 20
m, n = n, m  # 交換后 m=20, n=10

# 單行注釋
"""
多行注釋（實際是三引號字符串）
通常用于模塊、函數(shù)說明
"""
# 代碼塊通過縮進(jìn)定義
if True:
    print("縮進(jìn)4個空格")  # 屬于if代碼塊
    print("同一代碼塊")
print("已退出代碼塊")     # 不屬于if代碼塊

# 整數(shù)
int_num = 42          # 十進(jìn)制
binary_num = 0b1010   # 二進(jìn)制: 10
hex_num = 0xFF        # 十六進(jìn)制: 255
# 浮點(diǎn)數(shù)
float_num = 3.14159
scientific = 1.23e-4  # 0.000123
# 復(fù)數(shù)
complex_num = 3 + 4j
# 數(shù)值運(yùn)算
print(10 / 3)     # 浮點(diǎn)除法: 3.333...
print(10 // 3)    # 整除: 3
print(10 % 3)     # 取模: 1
print(2 ** 3)     # 冪運(yùn)算: 8

# 列表操作
fruits = ['apple', 'banana', 'orange']
fruits.append('grape')      # 末尾添加
fruits.insert(1, 'pear')    # 指定位置插入
fruits.remove('banana')     # 刪除元素
popped = fruits.pop()       # 移除并返回最后一個
print(fruits[0:2])          # 切片: ['apple', 'pear']
# 列表推導(dǎo)式（高效創(chuàng)建列表）
squares = [x**2 for x in range(10) if x % 2 == 0]
# 結(jié)果: [0, 4, 16, 36, 64]
# 元組
coordinates = (10.5, 20.3)
x, y = coordinates  # 解包
single_tuple = (42,)  # 單元素元組需要逗號
# 字符串操作
text = "Python Programming"
print(text.upper())         # 轉(zhuǎn)大寫
print(text.find("Pro"))     # 查找位置: 7
print(text.split(" "))      # 分割: ['Python', 'Programming']
print("Hello, {}!".format(name))  # 格式化
print(f"Version: {version}")      # f-string (Python 3.6+)

# 字典（鍵值對集合）
student = {
    "name": "Alice",
    "age": 20,
    "courses": ["Math", "Physics"]
}
student["grade"] = "A"          # 添加鍵值對
print(student.get("age"))       # 安全獲取: 20
print(student.get("score", "N/A"))  # 默認(rèn)值: N/A
# 遍歷字典
for key, value in student.items():
    print(f"{key}: {value}")
# 字典推導(dǎo)式
squared_dict = {x: x**2 for x in range(5)}
# 結(jié)果: {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}
# 集合（無序不重復(fù)元素）
set_a = {1, 2, 3, 3, 4}  # 自動去重: {1, 2, 3, 4}
set_b = {3, 4, 5, 6}
print(set_a | set_b)   # 并集: {1, 2, 3, 4, 5, 6}
print(set_a & set_b)   # 交集: {3, 4}
print(set_a - set_b)   # 差集: {1, 2}

# if-elif-else結(jié)構(gòu)
score = 85
if score >= 90:
    grade = "A"
elif score >= 80:
    grade = "B"
elif score >= 70:
    grade = "C"
else:
    grade = "D"
# 三元表達(dá)式
status = "及格" if score >= 60 else "不及格"
# 多條件判斷
age = 25
is_student = True
if 18 <= age <= 30 and is_student:
    discount = 0.5
elif age > 60 or not is_student:
    discount = 0.2
else:
    discount = 0

# for循環(huán)遍歷序列
colors = ["red", "green", "blue"]
for i, color in enumerate(colors):  # enumerate獲取索引和值
    print(f"顏色{i}: {color}")
# 遍歷字典
for key in student.keys():
    print(key)
# while循環(huán)
count = 0
while count < 5:
    print(f"計數(shù): {count}")
    count += 1
    if count == 3:
        break  # 跳出循環(huán)
# 循環(huán)控制
for i in range(10):
    if i % 2 == 0:
        continue  # 跳過偶數(shù)
    print(f"奇數(shù): {i}")
# 循環(huán)中的else子句（循環(huán)正常結(jié)束時執(zhí)行）
for i in range(3):
    print(i)
else:
    print("循環(huán)完成")

def greet(name, greeting="Hello"):
    """
    問候函數(shù)
    參數(shù):
    name: 姓名
    greeting: 問候語，默認(rèn)為'Hello'
    返回:
    問候字符串
    """
    return f"{greeting}, {name}!"
# 函數(shù)調(diào)用
print(greet("Alice"))                # 位置參數(shù)
print(greet(greeting="Hi", name="Bob"))  # 關(guān)鍵字參數(shù)
# 返回多個值（實際返回元組）
def min_max(numbers):
    return min(numbers), max(numbers)
low, high = min_max([3, 1, 4, 1, 5])
print(f"最小值: {low}, 最大值: {high}")

def flexible_func(a, b=10, *args, **kwargs):
    """
    演示各種參數(shù)類型
    a: 位置參數(shù)（必須）
    b: 默認(rèn)參數(shù)（可選）
    *args: 可變位置參數(shù)（元組）
    **kwargs: 可變關(guān)鍵字參數(shù)（字典）
    """
    print(f"a: {a}, b: ")
    print(f"args: {args}")
    print(f"kwargs: {kwargs}")
# 調(diào)用示例
flexible_func(1)                     # a:1, b:10, args:(), kwargs:{}
flexible_func(1, 2, 3, 4, x=5, y=6)  # a:1, b:2, args:(3,4), kwargs:{'x':5,'y':6}
# 參數(shù)解包
params = {'name': 'Charlie', 'age': 25}
def person_info(name, age):
    return f"{name} is {age} years old"
print(person_info(**params))  # 字典解包

# lambda表達(dá)式（匿名函數(shù)）
add = lambda x, y: x + y
print(add(5, 3))  # 8
# 常用于排序
students = [
    {'name': 'Alice', 'score': 85},
    {'name': 'Bob', 'score': 92},
    {'name': 'Charlie', 'score': 78}
]
# 按分?jǐn)?shù)排序
sorted_students = sorted(students, key=lambda s: s['score'], reverse=True)
# map函數(shù)：對序列每個元素應(yīng)用函數(shù)
numbers = [1, 2, 3, 4]
squared = list(map(lambda x: x**2, numbers))  # [1, 4, 9, 16]
# filter函數(shù)：過濾序列
evens = list(filter(lambda x: x % 2 == 0, numbers))  # [2, 4]
# 函數(shù)作為返回值
def make_multiplier(n):
    def multiplier(x):
        return x * n
    return multiplier
double = make_multiplier(2)
print(double(5))  # 10

# 全局變量與局部變量
global_var = "全局"
def scope_test():
    local_var = "局部"
    print(global_var)  # 可訪問全局變量
    print(local_var)   # 可訪問局部變量
    # 修改全局變量需要聲明
    global global_var
    global_var = "已修改"
scope_test()
# print(local_var)  # 錯誤！不能訪問函數(shù)內(nèi)的局部變量
# 閉包：函數(shù)記住其外部作用域
def counter():
    count = 0
    def increment():
        nonlocal count  # 聲明非局部變量
        count += 1
        return count
    return increment
my_counter = counter()
print(my_counter())  # 1
print(my_counter())  # 2 - count被記住了

class StudentManager:
    """學(xué)生成績管理類"""
    def __init__(self):
        self.students = {}
    def add_student(self, name, scores):
        """添加學(xué)生及其成績"""
        self.students[name] = scores
    def calculate_average(self, name):
        """計算學(xué)生平均分"""
        if name not in self.students:
            return None
        scores = self.students[name]
        return sum(scores) / len(scores) if scores else 0
    def get_top_student(self):
        """獲取平均分最高的學(xué)生"""
        if not self.students:
            return None
        # 使用lambda和max函數(shù)
        return max(self.students.items(), 
                  key=lambda item: self.calculate_average(item[0]))
    def analyze_scores(self):
        """成績分析"""
        all_scores = []
        for scores in self.students.values():
            all_scores.extend(scores)
        if not all_scores:
            return "暫無數(shù)據(jù)"
        analysis = {
            'total_students': len(self.students),
            'total_scores': len(all_scores),
            'average_all': sum(all_scores) / len(all_scores),
            'max_score': max(all_scores),
            'min_score': min(all_scores)
        }
        return analysis
# 使用示例
manager = StudentManager()
# 添加學(xué)生數(shù)據(jù)
manager.add_student("Alice", [85, 92, 78])
manager.add_student("Bob", [76, 88, 91])
manager.add_student("Charlie", [92, 95, 89])
# 分析數(shù)據(jù)
print(f"Alice的平均分: {manager.calculate_average('Alice'):.2f}")
top_student = manager.get_top_student()
print(f"最高分學(xué)生: {top_student[0]}, 平均分: {manager.calculate_average(top_student[0]):.2f}")
analysis = manager.analyze_scores()
print(f"總學(xué)生數(shù): {analysis['total_students']}")
print(f"所有科目平均分: {analysis['average_all']:.2f}")

class Date:
    """日期類演示類方法與靜態(tài)方法"""
    def __init__(self, year, month, day):
        self.year = year
        self.month = month
        self.day = day
    # 實例方法 - 操作實例屬性
    def display(self):
        return f"{self.year}-{self.month:02d}-{self.day:02d}"
    @classmethod  # 類方法 - 操作類本身
    def from_string(cls, date_string):
        """從字符串創(chuàng)建Date實例"""
        year, month, day = map(int, date_string.split('-'))
        return cls(year, month, day)  # cls代表類本身
    @classmethod
    def is_leap_year(cls, year):
        """判斷是否為閏年"""
        return year % 400 == 0 or (year % 4 == 0 and year % 100 != 0)
    @staticmethod  # 靜態(tài)方法 - 工具函數(shù)，不依賴類或?qū)嵗?
    def days_in_month(month, year):
        """返回指定月份的天數(shù)"""
        if month in [1, 3, 5, 7, 8, 10, 12]:
            return 31
        elif month in [4, 6, 9, 11]:
            return 30
        elif month == 2:
            return 29 if Date.is_leap_year(year) else 28
        else:
            raise ValueError("無效月份")
# 使用示例
d1 = Date(2024, 5, 20)
print(d1.display())  # 實例方法
d2 = Date.from_string("2024-05-21")  # 類方法作為替代構(gòu)造器
print(d2.display())
print(Date.is_leap_year(2024))  # True
print(Date.days_in_month(2, 2024))  # 29

class Temperature:
    """使用屬性裝飾器控制屬性訪問"""
    def __init__(self, celsius=0):
        self._celsius = celsius  # 私有屬性
    @property
    def celsius(self):
        """獲取攝氏溫度"""
        print("獲取攝氏溫度")
        return self._celsius
    @celsius.setter
    def celsius(self, value):
        """設(shè)置攝氏溫度"""
        if value < -273.15:
            raise ValueError("溫度不能低于絕對零度(-273.15°C)")
        print(f"設(shè)置攝氏溫度為: {value}")
        self._celsius = value
    @property
    def fahrenheit(self):
        """計算華氏溫度（只讀屬性）"""
        return self._celsius * 9/5 + 32
    @fahrenheit.setter
    def fahrenheit(self, value):
        """通過華氏溫度設(shè)置攝氏溫度"""
        self._celsius = (value - 32) * 5/9
# 描述符類
class ValidatedAttribute:
    """自定義描述符驗證屬性值"""
    def __init__(self, min_value=None, max_value=None):
        self.min_value = min_value
        self.max_value = max_value
        self.data = {}
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return self.data.get(id(instance))
    def __set__(self, instance, value):
        if self.min_value is not None and value < self.min_value:
            raise ValueError(f"值不能小于 {self.min_value}")
        if self.max_value is not None and value > self.max_value:
            raise ValueError(f"值不能大于 {self.max_value}")
        self.data[id(instance)] = value
class Product:
    price = ValidatedAttribute(min_value=0, max_value=10000)
    quantity = ValidatedAttribute(min_value=0, max_value=1000)
    def __init__(self, name, price, quantity):
        self.name = name
        self.price = price  # 觸發(fā)描述符的__set__
        self.quantity = quantity
# 使用示例
temp = Temperature(25)
print(f"攝氏: {temp.celsius}°C")  # 觸發(fā)@property
print(f"華氏: {temp.fahrenheit}°F")
temp.celsius = 30  # 觸發(fā)@celsius.setter
temp.fahrenheit = 100  # 觸發(fā)@fahrenheit.setter
try:
    p = Product("手機(jī)", -100, 10)
except ValueError as e:
    print(f"錯誤: {e}")  # 值不能小于 0
p2 = Product("筆記本", 5000, 5)
print(f"{p2.name}: 價格{p2.price}, 庫存{p2.quantity}")

import time
from functools import wraps
from typing import Callable, Any
def retry(max_attempts: int = 3, delay: float = 1.0):
    """
    參數(shù)化重試裝飾器
    Args:
        max_attempts: 最大重試次數(shù)
        delay: 重試間隔(秒)
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    print(f"嘗試第 {attempt + 1} 次...")
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        print(f"失敗，{delay}秒后重試...")
                        time.sleep(delay)
            print(f"所有 {max_attempts} 次嘗試均失敗")
            raise last_exception
        return wrapper
    return decorator
# 創(chuàng)建緩存裝飾器
def cache(maxsize: int = 128):
    """帶LRU緩存的參數(shù)化裝飾器"""
    def decorator(func: Callable) -> Callable:
        cache_dict = {}
        cache_keys = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 創(chuàng)建緩存鍵
            key = (args, frozenset(kwargs.items()))
            if key in cache_dict:
                print(f"緩存命中: {func.__name__}{args}")
                return cache_dict[key]
            print(f"計算: {func.__name__}{args}")
            result = func(*args, **kwargs)
            # LRU緩存邏輯
            cache_dict[key] = result
            cache_keys.append(key)
            if len(cache_dict) > maxsize:
                oldest_key = cache_keys.pop(0)
                del cache_dict[oldest_key]
            return result
        wrapper.clear_cache = lambda: (cache_dict.clear(), cache_keys.clear())
        return wrapper
    return decorator
# 使用參數(shù)化裝飾器
@retry(max_attempts=5, delay=0.5)
def unstable_api_call():
    """模擬不穩(wěn)定的API調(diào)用"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("API連接失敗")
    return "API調(diào)用成功"
@cache(maxsize=2)
def expensive_computation(n):
    """模擬昂貴計算"""
    print(f"執(zhí)行昂貴計算: {n}")
    time.sleep(1)
    return n * n
# 測試
try:
    result = unstable_api_call()
    print(result)
except Exception as e:
    print(f"最終失敗: {e}")
print(expensive_computation(5))  # 計算
print(expensive_computation(5))  # 從緩存獲取
print(expensive_computation(3))  # 計算
print(expensive_computation(4))  # 計算，觸發(fā)LRU淘汰
print(expensive_computation(5))  # 重新計算（已被淘汰）

from dataclasses import dataclass, field
from typing import List, ClassVar
import json
# 類裝飾器：為類添加單例模式
def singleton(cls):
    """單例模式裝飾器"""
    instances = {}
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance
@singleton
class AppConfig:
    def __init__(self):
        self.settings = {"theme": "dark", "language": "zh-CN"}
    def update(self, key, value):
        self.settings[key] = value
# 使用@dataclass自動生成方法
@dataclass(order=True, frozen=False)  # frozen=True創(chuàng)建不可變實例
class Person:
    """使用dataclass自動生成__init__, __repr__, __eq__等方法"""
    name: str
    age: int
    email: str = ""  # 默認(rèn)值
    hobbies: List[str] = field(default_factory=list)  # 可變默認(rèn)值的正確寫法
    id: ClassVar[int] = 0  # 類變量
    def __post_init__(self):
        """初始化后自動調(diào)用"""
        Person.id += 1
    def to_dict(self):
        return {"name": self.name, "age": self.age, "hobbies": self.hobbies}
# 測試
config1 = AppConfig()
config2 = AppConfig()
print(f"是否是同一個實例: {config1 is config2}")  # True
config1.update("theme", "light")
print(f"config2的主題: {config2.settings['theme']}")  # light
# dataclass使用
p1 = Person("Alice", 25, "alice@example.com", ["reading", "swimming"])
p2 = Person("Bob", 30)
print(p1)  # Person(name='Alice', age=25, email='alice@example.com', hobbies=['reading', 'swimming'])
print(p1 == p2)  # False
print(p1.to_dict())  # {'name': 'Alice', 'age': 25, 'hobbies': ['reading', 'swimming']}

import sqlite3
from contextlib import contextmanager
from typing import Optional, Iterator
# 方法1：使用類實現(xiàn)上下文管理器
class DatabaseConnection:
    """數(shù)據(jù)庫連接上下文管理器"""
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
    def __enter__(self) -> sqlite3.Connection:
        """進(jìn)入上下文時調(diào)用"""
        print(f"連接數(shù)據(jù)庫: {self.db_path}")
        self.connection = sqlite3.connect(self.db_path)
        self.connection.row_factory = sqlite3.Row  # 返回字典樣式的行
        return self.connection
    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出上下文時調(diào)用"""
        if self.connection:
            if exc_type is None:
                print("提交事務(wù)并關(guān)閉連接")
                self.connection.commit()
            else:
                print(f"發(fā)生錯誤: {exc_val}，回滾事務(wù)")
                self.connection.rollback()
            self.connection.close()
        print("數(shù)據(jù)庫連接已關(guān)閉")
        return False  # 不抑制異常，True表示抑制
# 方法2：使用@contextmanager裝飾器
@contextmanager
def temp_table(connection: sqlite3.Connection, table_name: str) -> Iterator:
    """創(chuàng)建臨時表的上下文管理器"""
    cursor = connection.cursor()
    try:
        print(f"創(chuàng)建臨時表: {table_name}")
        cursor.execute(f"CREATE TEMP TABLE {table_name} (id INTEGER, name TEXT)")
        yield cursor  # 這里將控制權(quán)交給with塊內(nèi)的代碼
    finally:
        print(f"清理臨時表: {table_name}")
        cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
        cursor.close()
# 方法3：處理多個資源的上下文管理器
from contextlib import ExitStack
class MultiResourceManager:
    """管理多個資源的上下文管理器"""
    def __init__(self):
        self.stack = ExitStack()
    def add_file(self, filepath, mode='r'):
        """添加文件到資源棧"""
        file_obj = self.stack.enter_context(open(filepath, mode, encoding='utf-8'))
        return file_obj
    def add_connection(self, db_path):
        """添加數(shù)據(jù)庫連接到資源棧"""
        conn = self.stack.enter_context(DatabaseConnection(db_path))
        return conn
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stack.close()
# 使用示例
print("=== 示例1: 數(shù)據(jù)庫連接上下文 ===")
with DatabaseConnection(":memory:") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    with temp_table(conn, "temp_data") as temp_cursor:
        temp_cursor.execute("INSERT INTO temp_data VALUES (1, '測試')")
        temp_cursor.execute("SELECT * FROM temp_data")
        print(temp_cursor.fetchall())
print("\n=== 示例2: 多資源管理 ===")
with MultiResourceManager() as manager:
    # 自動管理多個資源
    conn = manager.add_connection(":memory:")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE test (id INTEGER)")
    # 可以繼續(xù)添加其他資源
    # file = manager.add_file('test.txt', 'w')
    # file.write('test content')
print("\n=== 示例3: 使用contextlib實用工具 ===")
from contextlib import suppress, redirect_stdout
import io
# suppress忽略指定異常
with suppress(FileNotFoundError):
    with open("不存在的文件.txt") as f:
        content = f.read()
# redirect_stdout重定向輸出
output = io.StringIO()
with redirect_stdout(output):
    print("這條信息不會顯示在控制臺")
    print("而是被重定向到StringIO")
print(f"捕獲的輸出: {output.getvalue()}")

class SingletonMeta(type):
    """單例模式元類"""
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            print(f"創(chuàng)建 {cls.__name__} 的唯一實例")
            cls._instances[cls] = super().__call__(*args, **kwargs)
        else:
            print(f"返回 {cls.__name__} 的現(xiàn)有實例")
        return cls._instances[cls]
class ValidateAttributesMeta(type):
    """驗證類屬性的元類"""
    def __new__(mcs, name, bases, attrs):
        print(f"\n創(chuàng)建類: {name}")
        # 驗證屬性名不能以下劃線開頭（示例規(guī)則）
        invalid_attrs = []
        for attr_name in attrs:
            if attr_name.startswith('_') and not attr_name.startswith('__'):
                invalid_attrs.append(attr_name)
        if invalid_attrs:
            print(f"警告: 屬性 {invalid_attrs} 不建議以下劃線開頭")
        # 自動添加類版本信息
        if '__version__' not in attrs:
            attrs['__version__'] = '1.0.0'
        return super().__new__(mcs, name, bases, attrs)
# 使用元類
class Database(metaclass=SingletonMeta):
    def __init__(self, name):
        self.name = name
        self.connected = False
    def connect(self):
        self.connected = True
        print(f"{self.name} 數(shù)據(jù)庫已連接")
class UserModel(metaclass=ValidateAttributesMeta):
    """用戶模型類，屬性會被元類驗證"""
    _private_data = "不應(yīng)直接訪問"  # 這會觸發(fā)警告
    name = "默認(rèn)用戶"
    email = ""
    def get_info(self):
        return f"{self.name} <{self.email}>"
# 測試
print("=== 元類示例1: 單例模式 ===")
db1 = Database("主數(shù)據(jù)庫")
db1.connect()
db2 = Database("主數(shù)據(jù)庫")  # 返回同一個實例
print(f"db1 is db2: {db1 is db2}")
print(f"db2.name: {db2.name}")
print("\n=== 元類示例2: 屬性驗證 ===")
user = UserModel()
print(f"UserModel版本: {UserModel.__version__}")
print(f"用戶信息: {user.get_info()}")
# 動態(tài)創(chuàng)建類
print("\n=== 動態(tài)創(chuàng)建類 ===")
def class_factory(class_name, base_classes, attributes):
    """動態(tài)創(chuàng)建類的工廠函數(shù)"""
    return type(class_name, base_classes, attributes)
# 動態(tài)創(chuàng)建類
Animal = class_factory(
    'Animal',
    (),
    {
        'species': '未知',
        '__init__': lambda self, name: setattr(self, 'name', name),
        'speak': lambda self: print(f"{self.name} 發(fā)出聲音")
    }
)
cat = Animal("貓咪")
cat.species = "貓科"
cat.speak()
print(f"物種: {cat.species}")

import threading
import time
import concurrent.futures
from queue import Queue, Empty
from typing import List
import random
# 線程安全的計數(shù)器
class ThreadSafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()
    def increment(self):
        with self._lock:  # 自動獲取和釋放鎖
            self._value += 1
            return self._value
    @property
    def value(self):
        with self._lock:
            return self._value
# 生產(chǎn)者-消費(fèi)者模式
class ProducerConsumer:
    def __init__(self, max_size=5):
        self.queue = Queue(maxsize=max_size)
        self.stop_event = threading.Event()
    def producer(self, producer_id: int):
        """生產(chǎn)者線程函數(shù)"""
        while not self.stop_event.is_set():
            item = f"產(chǎn)品-{producer_id}-{time.time():.2f}"
            try:
                self.queue.put(item, timeout=1)
                print(f"生產(chǎn)者{producer_id} 生產(chǎn): {item}")
                time.sleep(random.uniform(0.1, 0.5))
            except Exception as e:
                break
    def consumer(self, consumer_id: int):
        """消費(fèi)者線程函數(shù)"""
        while not self.stop_event.is_set() or not self.queue.empty():
            try:
                item = self.queue.get(timeout=1)
                print(f"消費(fèi)者{consumer_id} 消費(fèi): {item}")
                self.queue.task_done()
                time.sleep(random.uniform(0.2, 0.8))
            except Empty:
                continue
    def run(self, num_producers=2, num_consumers=3, duration=5):
        """運(yùn)行生產(chǎn)消費(fèi)過程"""
        threads = []
        # 創(chuàng)建生產(chǎn)者線程
        for i in range(num_producers):
            t = threading.Thread(target=self.producer, args=(i,))
            t.daemon = True
            threads.append(t)
            t.start()
        # 創(chuàng)建消費(fèi)者線程
        for i in range(num_consumers):
            t = threading.Thread(target=self.consumer, args=(i,))
            t.daemon = True
            threads.append(t)
            t.start()
        # 運(yùn)行指定時間
        time.sleep(duration)
        self.stop_event.set()
        # 等待隊列清空
        self.queue.join()
        print("所有任務(wù)完成")
# 使用ThreadPoolExecutor
def process_task(task_id: int) -> str:
    """模擬處理任務(wù)"""
    sleep_time = random.uniform(0.5, 2.0)
    print(f"任務(wù){(diào)task_id} 開始執(zhí)行，預(yù)計耗時{sleep_time:.1f}秒")
    time.sleep(sleep_time)
    result = f"任務(wù){(diào)task_id} 完成"
    return result
def execute_with_threadpool():
    """使用線程池執(zhí)行任務(wù)"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # 提交任務(wù)
        future_to_task = {
            executor.submit(process_task, i): i 
            for i in range(10)
        }
        # 收集結(jié)果
        results = []
        for future in concurrent.futures.as_completed(future_to_task):
            task_id = future_to_task[future]
            try:
                result = future.result()
                results.append(result)
                print(result)
            except Exception as e:
                print(f"任務(wù){(diào)task_id} 生成異常: {e}")
        print(f"\n總共完成 {len(results)} 個任務(wù)")
# 測試
print("=== 線程安全計數(shù)器 ===")
counter = ThreadSafeCounter()
def worker(counter: ThreadSafeCounter, iterations: int):
    for _ in range(iterations):
        counter.increment()
threads = []
for _ in range(10):
    t = threading.Thread(target=worker, args=(counter, 100))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"最終計數(shù)值: {counter.value} (期望值: 1000)")
print("\n=== 生產(chǎn)者-消費(fèi)者模式 ===")
pc = ProducerConsumer(max_size=3)
pc.run(duration=3)
print("\n=== 線程池執(zhí)行器 ===")
execute_with_threadpool()

import asyncio
import aiohttp
import asyncpg
from datetime import datetime
# 異步上下文管理器
class AsyncDatabaseConnection:
    """異步數(shù)據(jù)庫連接"""
    async def __aenter__(self):
        print("異步連接數(shù)據(jù)庫...")
        self.conn = await asyncpg.connect(
            user='user', password='password',
            database='test', host='localhost'
        )
        return self.conn
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("異步關(guān)閉數(shù)據(jù)庫連接...")
        await self.conn.close()
# 異步迭代器
class AsyncDataStreamer:
    """異步數(shù)據(jù)流"""
    def __init__(self, limit=10, delay=0.5):
        self.limit = limit
        self.delay = delay
        self.current = 0
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(self.delay)
        data = f"數(shù)據(jù)塊-{self.current}-{datetime.now().timestamp()}"
        self.current += 1
        return data
# 異步任務(wù)示例
async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
    """異步獲取URL內(nèi)容"""
    try:
        print(f"開始獲取: {url}")
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            return f"{url}: {len(content)} 字符"
    except Exception as e:
        return f"{url}: 錯誤 - {str(e)}"
async def process_data_stream():
    """處理異步數(shù)據(jù)流"""
    print("開始處理數(shù)據(jù)流...")
    streamer = AsyncDataStreamer(limit=5, delay=0.3)
    async for data in streamer:
        print(f"接收到: {data}")
        # 模擬數(shù)據(jù)處理
        await asyncio.sleep(0.1)
    print("數(shù)據(jù)流處理完成")
async def concurrent_url_fetcher():
    """并發(fā)獲取多個URL"""
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://jsonplaceholder.typicode.com/posts/1",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print("\n獲取結(jié)果:")
        for result in results:
            print(f"  {result}")
async def main():
    """主異步函數(shù)"""
    print("=== 異步編程示例 ===\n")
    # 示例1: 異步數(shù)據(jù)流
    await process_data_stream()
    print("\n" + "="*50 + "\n")
    # 示例2: 并發(fā)URL獲取
    await concurrent_url_fetcher()
    print("\n" + "="*50 + "\n")
    # 示例3: 異步任務(wù)控制
    print("異步任務(wù)控制示例:")
    # 創(chuàng)建任務(wù)
    task1 = asyncio.create_task(fetch_data("任務(wù)1", 2))
    task2 = asyncio.create_task(fetch_data("任務(wù)2", 1))
    # 等待特定任務(wù)完成
    done, pending = await asyncio.wait([task1, task2], timeout=1.5)
    print(f"已完成: {len(done)} 個任務(wù)")
    print(f"仍在進(jìn)行: {len(pending)} 個任務(wù)")
    # 取消剩余任務(wù)
    for task in pending:
        task.cancel()
    try:
        await asyncio.gather(*pending, return_exceptions=True)
    except asyncio.CancelledError:
        print("已取消剩余任務(wù)")
async def fetch_data(name: str, delay: float) -> str:
    """模擬異步數(shù)據(jù)獲取"""
    await asyncio.sleep(delay)
    return f"{name} 完成，延遲 {delay}秒"
# 運(yùn)行異步程序
if __name__ == "__main__":
    # 對于異步程序，需要特殊方式運(yùn)行
    print("注意: 異步代碼需要合適的運(yùn)行環(huán)境")
    print("可以使用以下方式運(yùn)行:")
    print("1. 在Jupyter中使用: await main()")
    print("2. 在腳本中使用: asyncio.run(main())")
    # 實際運(yùn)行代碼
    # asyncio.run(main())

import sys
from dataclasses import dataclass
from pympler import asizeof  # 需要安裝: pip install pympler
class RegularUser:
    """普通類，使用__dict__存儲屬性"""
    def __init__(self, user_id, name, email, age):
        self.user_id = user_id
        self.name = name
        self.email = email
        self.age = age
class SlotUser:
    """使用__slots__優(yōu)化內(nèi)存的類"""
    __slots__ = ('user_id', 'name', 'email', 'age')
    def __init__(self, user_id, name, email, age):
        self.user_id = user_id
        self.name = name
        self.email = email
        self.age = age
@dataclass
class DataClassUser:
    """使用dataclass"""
    user_id: int
    name: str
    email: str
    age: int
def memory_comparison():
    """比較不同類的內(nèi)存使用"""
    # 創(chuàng)建大量實例
    num_instances = 10000
    # 普通類
    regular_users = [RegularUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                     for i in range(num_instances)]
    # slots類
    slot_users = [SlotUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                  for i in range(num_instances)]
    # dataclass
    dataclass_users = [DataClassUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                       for i in range(num_instances)]
    # 計算內(nèi)存使用
    reg_memory = sum(asizeof.asizeof(u) for u in regular_users[:100]) * (num_instances / 100)
    slot_memory = sum(asizeof.asizeof(u) for u in slot_users[:100]) * (num_instances / 100)
    dc_memory = sum(asizeof.asizeof(u) for u in dataclass_users[:100]) * (num_instances / 100)
    print(f"內(nèi)存使用對比 ({num_instances}個實例):")
    print(f"普通類:    {reg_memory / 1024:.1f} KB")
    print(f"Slots類:   {slot_memory / 1024:.1f} KB")
    print(f"Dataclass: {dc_memory / 1024:.1f} KB")
    print(f"節(jié)省比例:  {(1 - slot_memory/reg_memory)*100:.1f}%")
# 測試
if __name__ == "__main__":
    print("=== 內(nèi)存優(yōu)化示例 ===")
    # 單個實例對比
    reg = RegularUser(1, "Alice", "alice@example.com", 25)
    slot = SlotUser(1, "Alice", "alice@example.com", 25)
    print(f"單個RegularUser大小: {sys.getsizeof(reg)} 字節(jié) + __dict__")
    print(f"單個SlotUser大小: {sys.getsizeof(slot)} 字節(jié)")
    # 嘗試動態(tài)添加屬性
    try:
        reg.new_attr = "可以動態(tài)添加"  # 成功
        print("普通類可以動態(tài)添加屬性")
    except:
        print("普通類無法動態(tài)添加屬性")
    try:
        slot.new_attr = "嘗試動態(tài)添加"  # 失敗
        print("Slots類可以動態(tài)添加屬性")
    except AttributeError:
        print("Slots類無法動態(tài)添加屬性")
    print("\n批量內(nèi)存對比:")
    memory_comparison()