首页  

python3 实用代码2     所属分类 python 浏览量 17
一、数据处理类

1. collections - 容器数据类型

from collections import Counter, defaultdict, deque, namedtuple, OrderedDict

# Counter - 计数器
cnt = Counter(['a', 'b', 'c', 'a', 'b', 'a'])
print(cnt)  # Counter({'a': 3, 'b': 2, 'c': 1})

# defaultdict - 带默认值的字典
dd = defaultdict(list)
dd['key'].append(1)  # 无需先检查key是否存在

# namedtuple - 命名元组
Point = namedtuple('Point', ['x', 'y'])
p = Point(1, 2)
print(p.x, p.y)

# deque - 双端队列
dq = deque([1, 2, 3])
dq.appendleft(0)  # 左侧添加
dq.pop()          # 右侧弹出

2. itertools - 迭代器工具

import itertools

# 排列组合
list(itertools.combinations('ABC', 2))  # [('A','B'), ('A','C'), ('B','C')]
list(itertools.permutations('AB', 2))   # [('A','B'), ('B','A')]

# 无限迭代器
for i in itertools.count(10, 2):  # 从10开始,步长为2
    if i > 20: break
    print(i)

# 循环迭代
for item in itertools.cycle(['A', 'B', 'C']):
    # 无限循环
    pass




二、文件与目录操作
1. pathlib - 路径操作(Python3.4+推荐)
from pathlib import Path

# 创建Path对象
p = Path('/home/user/documents/file.txt')

# 常用操作
print(p.name)      # file.txt
print(p.stem)      # file
print(p.suffix)    # .txt
print(p.parent)    # /home/user/documents
print(p.exists())  # True/False

# 遍历目录
for file in Path('.').glob('*.py'):
    print(file)

# 创建目录
Path('new_dir').mkdir(parents=True, exist_ok=True)

# 读写文件
content = p.read_text()
p.write_text('Hello World')


2. os 和 shutil - 操作系统接口

import os, shutil

# 文件操作
os.path.exists('/path/to/file')
os.path.getsize('/path/to/file')
os.path.join('folder', 'file.txt')

# 目录操作
os.listdir('.')
os.makedirs('path/to/dir', exist_ok=True)

# 文件复制、移动、删除
shutil.copy('src.txt', 'dst.txt')
shutil.move('old.txt', 'new.txt')
shutil.rmtree('dir_to_remove')  # 递归删除目录




三、日期时间处理
datetime - 日期时间
from datetime import datetime, date, timedelta
import time

# 当前时间
now = datetime.now()
print(now.strftime('%Y-%m-%d %H:%M:%S'))

# 时间计算
tomorrow = now + timedelta(days=1)
last_week = now - timedelta(weeks=1)

# 时间戳
timestamp = time.time()
dt_from_timestamp = datetime.fromtimestamp(timestamp)

# 字符串转时间
dt = datetime.strptime('2023-01-01', '%Y-%m-%d')




四、JSON处理

import json
import csv

# JSON读写
data = {'name': 'John', 'age': 30}

# 写入文件
with open('data.json', 'w', encoding='utf-8') as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 读取文件
with open('data.json', 'r', encoding='utf-8') as f:
    loaded_data = json.load(f)

# 字符串转换
json_str = json.dumps(data)
parsed_data = json.loads(json_str)



五、网络请求
requests - HTTP请求
import requests

# GET请求
response = requests.get('https://api.example.com/data')
print(response.status_code)
print(response.json())  # 如果是JSON数据

# 带参数的请求
params = {'key': 'value'}
response = requests.get('https://api.example.com', params=params)

# POST请求
data = {'name': 'John', 'age': 30}
response = requests.post('https://api.example.com', json=data)

# 设置请求头
headers = {'User-Agent': 'my-app'}
response = requests.get(url, headers=headers)

# 会话保持
session = requests.Session()
session.get('https://example.com/login')



六、并发处理
1. concurrent.futures - 线程/进程池

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * n

# 线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    # 提交单个任务
    future = executor.submit(task, 5)
    print(future.result())
    
    # 批量提交
    results = executor.map(task, range(5))
    for result in results:
        print(result)

# 进程池(适合CPU密集型)
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(task, range(10)))



2. asyncio - 异步IO(Python3.5+)

import asyncio

async def fetch_data(url):
    # 模拟异步操作
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    # 并发执行多个协程
    tasks = [
        fetch_data('url1'),
        fetch_data('url2'),
        fetch_data('url3')
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

# 运行
asyncio.run(main())




七、日志记录
import logging

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

# 创建logger
logger = logging.getLogger(__name__)

# 记录日志
logger.debug('Debug message')
logger.info('Info message')
logger.warning('Warning message')
logger.error('Error message')
logger.critical('Critical message')



八、配置管理
configparser - 配置文件

import configparser

# 创建配置
config = configparser.ConfigParser()
config['DEFAULT'] = {'Server': 'localhost', 'Port': '8080'}
config['DATABASE'] = {'Host': 'db.local', 'Name': 'mydb'}

# 写入文件
with open('config.ini', 'w') as f:
    config.write(f)

# 读取配置
config.read('config.ini')
server = config['DEFAULT']['Server']
db_host = config['DATABASE']['Host']




九、数据验证
pydantic - 数据验证和设置管理
from pydantic import BaseModel, ValidationError, Field
from typing import List, Optional
from datetime import datetime

class User(BaseModel):
    id: int
    name: str = Field(..., min_length=2, max_length=50)
    email: str
    age: Optional[int] = Field(None, ge=0, le=150)
    friends: List[int] = []
    created_at: datetime = datetime.now()

# 验证数据
try:
    user = User(
        id=1,
        name='John',
        email='john@example.com',
        age=30,
        friends=[2, 3, 4]
    )
    print(user.dict())  # 转换为字典
except ValidationError as e:
    print(e.errors())




十、其他实用工具类
1. 装饰器工具

from functools import wraps, lru_cache
import time

# 缓存装饰器
@lru_cache(maxsize=128)
def expensive_function(n):
    time.sleep(2)
    return n * n

# 计时装饰器
def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end-start:.2f} seconds")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)


2. 单例模式

class Singleton:
    _instance = None
    
    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, value):
        if not hasattr(self, 'initialized'):  # 防止重复初始化
            self.value = value
            self.initialized = True



十一、第三方库推荐
pandas - 数据分析
numpy - 数值计算
Pillow - 图像处理
openpyxl - Excel操作
SQLAlchemy - 数据库ORM
click - 命令行工具
rich - 终端美化
tqdm - 进度条




上一篇     下一篇
N字价格运动与结构

基于深度学习和计算机视觉的K线图交易信号识别

Python f-string 格式化字符串字面量

《走进我的交易室》精华笔记