Python3 代码片段，涵盖文件操作、数据处理、网络请求等常见场景


1. 文件操作：读取和写入 CSV 文件
读取 CSV 文件并筛选特定数据写入新文件

import csv

def process_csv(input_file, output_file):
    with open(input_file, 'r', newline='') as infile, \
         open(output_file, 'w', newline='') as outfile:
        reader = csv.DictReader(infile)
        fieldnames = reader.fieldnames  # 获取原表头
        
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()  # 写入表头
        
        # 筛选条件：保留 age 大于 30 的行
        for row in reader:
            if int(row['age']) > 30:
                writer.writerow(row)

# 使用示例
process_csv('input.csv', 'output.csv')




2. 数据处理：JSON 数据解析与转换
从 JSON 文件读取数据，提取特定字段并转换为 CSV 格式


import json
import csv

def json_to_csv(json_file, csv_file):
    with open(json_file, 'r') as f:
        data = json.load(f)  # 假设 JSON 文件是列表格式
    
    # 提取需要的字段
    fieldnames = ['name', 'age', 'email']
    
    with open(csv_file, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        
        for item in data:
            writer.writerow({
                'name': item.get('name', ''),
                'age': item.get('age', 0),
                'email': item.get('email', '')
            })

# 使用示例
json_to_csv('data.json', 'output.csv')




3. 网络请求：获取并解析网页内容
使用 requests 和 BeautifulSoup 爬取网页标题和链接

import requests
from bs4 import BeautifulSoup

def scrape_website(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        soup = BeautifulSoup(response.text, 'html.parser')
        
        # 提取所有链接
        links = []
        for a in soup.find_all('a', href=True):
            links.append({
                'text': a.get_text(strip=True),
                'url': a['href']
            })
        
        return links
    else:
        print(f"请求失败，状态码：{response.status_code}")
        return []

# 使用示例
url = 'https://example.com'

links = scrape_website(url)
for link in links[:5]:  # 打印前5个链接
    print(f"{link['text']}: {link['url']}")





4. 多线程处理：加速 IO 密集型任务
使用线程池并行下载多个文件

import requests
from concurrent.futures import ThreadPoolExecutor

def download_file(url, filename):
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        
        with open(filename, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        print(f"下载完成: {filename}")
    except Exception as e:
        print(f"下载失败 {filename}: {e}")

def download_files_parallel(urls):
    with ThreadPoolExecutor(max_workers=5) as executor:
        for url in urls:
            filename = url.split('/')[-1]
            executor.submit(download_file, url, filename)

# 使用示例
urls = [
    'https://example.com/file1.txt',
    'https://example.com/file2.txt',
    'https://example.com/file3.txt'
]
download_files_parallel(urls)




5. 数据分析：使用 Pandas 处理表格数据
读取 Excel 文件，进行数据清洗和分析

import pandas as pd

def analyze_data(file_path):
    # 读取 Excel 文件
    df = pd.read_excel(file_path)
    
    # 数据清洗：删除缺失值
    df = df.dropna(subset=['age', 'score'])
    
    # 数据转换：年龄分组
    df['age_group'] = pd.cut(df['age'], bins=[0, 18, 30, 50, 100],
                             labels=['青少年', '青年', '中年', '老年'])
    
    # 数据分析：按年龄组计算平均分数
    result = df.groupby('age_group')['score'].mean().reset_index()
    
    # 保存结果
    result.to_excel('analysis_result.xlsx', index=False)
    
    return result

# 使用示例
result = analyze_data('data.xlsx')
print(result)




6. 定时任务：周期性执行函数
使用 schedule 库实现定时任务（如每天早上 8 点执行备份）


import schedule
import time
import os

def backup_data():
    print(f"开始备份... {time.strftime('%Y-%m-%d %H:%M:%S')}")
    # 备份逻辑，例如复制文件到备份目录
    os.system('cp -r data/ backup/')
    print("备份完成")

# 设置定时任务：每天早上 8 点执行
schedule.every().day.at("08:00").do(backup_data)

# 持续运行，等待任务执行
while True:
    schedule.run_pending()
    time.sleep(60)  # 每分钟检查一次是否有待执行的任务




7. 日志记录：结构化日志输出
使用 logging 模块记录程序运行状态和错误信息

import logging

def setup_logger():
    # 创建日志器
    logger = logging.getLogger('my_app')
    logger.setLevel(logging.DEBUG)
    
    # 创建控制台处理器
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)
    
    # 创建文件处理器
    fh = logging.FileHandler('app.log')
    fh.setLevel(logging.DEBUG)
    
    # 定义日志格式
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    ch.setFormatter(formatter)
    fh.setFormatter(formatter)
    
    # 添加处理器到日志器
    logger.addHandler(ch)
    logger.addHandler(fh)
    
    return logger

# 使用示例
logger = setup_logger()

try:
    # 模拟操作
    result = 10 / 0
except Exception as e:
    logger.error(f"发生错误: {e}", exc_info=True)

logger.info("程序正常运行中")



8. 缓存机制：减少重复计算
使用 functools.lru_cache 缓存函数结果，加速递归或耗时计算

import functools

@functools.lru_cache(maxsize=128)  # 缓存最近 128 次调用结果
def fibonacci(n):
    if n <= 1:
        return n
    else:
        return fibonacci(n-1) + fibonacci(n-2)

# 使用示例
print(f"斐波那契数列第 10 项: {fibonacci(10)}")  # 首次计算，耗时较长
print(f"斐波那契数列第 10 项: {fibonacci(10)}")  # 直接从缓存获取，几乎瞬间返回

# 查看缓存信息
print(fibonacci.cache_info())  # 输出: CacheInfo(hits=1, misses=11, maxsize=128, currsize=11)




9. Web API：创建简单的 HTTP 服务器
使用 Flask 框架快速搭建 REST API

from flask import Flask, jsonify, request

app = Flask(__name__)

# 模拟数据
books = [
    {"id": 1, "title": "Python 编程", "author": "张三"},
    {"id": 2, "title": "数据分析实战", "author": "李四"}
]

# 获取所有书籍
@app.route('/api/books', methods=['GET'])
def get_books():
    return jsonify(books)

# 获取单本书籍
@app.route('/api/books/', methods=['GET'])
def get_book(book_id):
    book = next((b for b in books if b['id'] == book_id), None)
    if book:
        return jsonify(book)
    return jsonify({"error": "书籍未找到"}), 404

# 添加书籍
@app.route('/api/books', methods=['POST'])
def add_book():
    new_book = request.get_json()
    books.append({
        "id": max(b['id'] for b in books) + 1,
        "title": new_book.get('title', ''),
        "author": new_book.get('author', '')
    })
    return jsonify(books[-1]), 201

if __name__ == '__main__':
    app.run(debug=True)  # 开启调试模式，修改代码后自动重启




10. 数据加密：简单的文本加密解密

使用 cryptography 库实现 AES 加密（需安装：pip install cryptography）

from cryptography.fernet import Fernet

# 生成密钥并保存
def generate_key():
    key = Fernet.generate_key()
    with open('secret.key', 'wb') as key_file:
        key_file.write(key)
    return key

# 加载密钥
def load_key():
    try:
        return open('secret.key', 'rb').read()
    except FileNotFoundError:
        return generate_key()

# 加密文本
def encrypt_message(message, key):
    f = Fernet(key)
    encrypted = f.encrypt(message.encode())
    return encrypted

# 解密文本
def decrypt_message(encrypted_message, key):
    f = Fernet(key)
    decrypted = f.decrypt(encrypted_message).decode()
    return decrypted

# 使用示例
key = load_key()
message = "这是一条机密信息"

encrypted = encrypt_message(message, key)
print(f"加密后: {encrypted}")

decrypted = decrypt_message(encrypted, key)
print(f"解密后: {decrypted}")




 Python3 列表（list） 元组（tuple）字典（dict）集合（set）使用