Python 脚本自动化实战:让重复工作一键完成

import os
from pathlib import Path

def batch_rename(directory, prefix="file"):
    """批量重命名目录下的所有文件"""
    path = Path(directory)
    
    for i, file in enumerate(path.iterdir(), 1):
        if file.is_file():
            # 保留扩展名
            ext = file.suffix
            new_name = f"{prefix}_{i:03d}{ext}"
            new_path = file.parent / new_name
            file.rename(new_path)
            print(f"{file.name} → {new_name}")

# 使用
batch_rename("/path/to/photos", prefix="trip2026")

from PIL import Image
from pathlib import Path

def compress_images(directory, quality=70):
    """批量压缩图片到指定目录"""
    path = Path(directory)
    output_dir = path / "compressed"
    output_dir.mkdir(exist_ok=True)
    
    for img_file in path.glob("*.jpg"):
        img = Image.open(img_file)
        # 等比缩放
        max_size = (1920, 1080)
        img.thumbnail(max_size)
        
        output_path = output_dir / img_file.name
        img.save(output_path, "JPEG", quality=quality, optimize=True)
        
        # 对比大小
        old_size = img_file.stat().st_size / 1024
        new_size = output_path.stat().st_size / 1024
        print(f"{img_file.name}: {old_size:.1f}KB → {new_size:.1f}KB (-{100*(1-new_size/old_size):.1f}%)")

compress_images("/path/to/photos")

import pandas as pd

# 读取
df = pd.read_excel("data.xlsx", sheet_name="Sheet1")

# 数据处理
df['年龄'] = pd.to_numeric(df['年龄'], errors='coerce')
df = df.dropna(subset=['年龄'])

# 分组统计
summary = df.groupby('部门')['工资'].agg(['mean', 'sum', 'count'])
print(summary)

# 输出到新的 Excel
with pd.ExcelWriter('output.xlsx') as writer:
    df.to_excel(writer, sheet_name='原始数据', index=False)
    summary.to_excel(writer, sheet_name='汇总统计')

from docx import Document
from docx.shared import Pt, RGBColor

# 创建 Word
doc = Document()

# 添加标题
doc.add_heading('Python 自动生成报告', 0)

# 添加段落
p = doc.add_paragraph('这是一个自动生成的 ')
run = p.add_run('粗体')
run.bold = True
run.font.size = Pt(14)

# 添加表格
table = doc.add_table(rows=3, cols=3)
table.style = 'Light Grid Accent 1'
for i in range(3):
    for j in range(3):
        cell = table.cell(i, j)
        cell.text = f'行 {i+1} 列 {j+1}'

doc.save('report.docx')

import requests
from bs4 import BeautifulSoup
import csv

def scrape_jobs(keyword, pages=1):
    """抓取招聘信息"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    
    results = []
    
    for page in range(1, pages + 1):
        url = f"https://example.com/jobs?kw={keyword}&page={page}"
        resp = requests.get(url, headers=headers, timeout=10)
        resp.raise_for_status()
        
        soup = BeautifulSoup(resp.text, 'html.parser')
        
        for job_card in soup.select('.job-card'):
            title = job_card.select_one('.title').text.strip()
            company = job_card.select_one('.company').text.strip()
            salary = job_card.select_one('.salary').text.strip()
            
            results.append({
                'title': title,
                'company': company,
                'salary': salary
            })
    
    # 保存到 CSV
    with open(f'jobs_{keyword}.csv', 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=['title', 'company', 'salary'])
        writer.writeheader()
        writer.writerows(results)
    
    print(f"已抓取 {len(results)} 条数据")

scrape_jobs("Python", pages=5)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

def send_email(to, subject, body, attachments=None):
    """发送邮件"""
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    username = "your_email@gmail.com"
    password = "app_password"  # 开启两步验证后用应用专用密码
    
    msg = MIMEMultipart()
    msg['From'] = username
    msg['To'] = to
    msg['Subject'] = subject
    
    msg.attach(MIMEText(body, 'html'))
    
    # 附件
    if attachments:
        for file_path in attachments:
            with open(file_path, 'rb') as f:
                attach = MIMEApplication(f.read())
                attach.add_header('Content-Disposition', 'attachment', 
                                 filename=file_path.split('/')[-1])
                msg.attach(attach)
    
    # 发送
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(username, password)
        server.send_message(msg)
    
    print(f"邮件已发送到 {to}")

# 使用
send_email(
    "recipient@example.com",
    "每日数据报告",
    "<h1>今日数据</h1><p>详见附件</p>",
    attachments=["/path/to/report.pdf"]
)

import schedule
import time
from datetime import datetime

def daily_report():
    """每天早上 9 点生成报告"""
    print(f"[{datetime.now()}] 生成每日报告...")
    # 实际报告生成逻辑
    generate_report()

def backup_database():
    """每天凌晨 2 点备份数据库"""
    print(f"[{datetime.now()}] 备份数据库...")
    # 备份逻辑

# 配置任务
schedule.every().day.at("09:00").do(daily_report)
schedule.every().day.at("02:00").do(backup_database)
schedule.every(30).minutes.do(check_server_status)

print("定时任务已启动,等待执行...")

while True:
    schedule.run_pending()
    time.sleep(60)

import psutil
import requests
import time

def monitor_server(threshold_cpu=80, threshold_mem=80, threshold_disk=90):
    """监控服务器 CPU/内存/磁盘"""
    
    cpu = psutil.cpu_percent(interval=1)
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage('/').percent
    
    status = {
        'time': time.strftime('%Y-%m-%d %H:%M:%S'),
        'cpu': cpu,
        'memory': mem,
        'disk': disk
    }
    
    # 告警
    alerts = []
    if cpu > threshold_cpu:
        alerts.append(f"⚠️ CPU 使用率过高: {cpu}%")
    if mem > threshold_mem:
        alerts.append(f"⚠️ 内存使用率过高: {mem}%")
    if disk > threshold_disk:
        alerts.append(f"⚠️ 磁盘使用率过高: {disk}%")
    
    if alerts:
        # 发钉钉/飞书/邮件通知
        send_alert('\n'.join(alerts))
    
    return status

def send_alert(message):
    """发送告警到钉钉"""
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=XXX"
    requests.post(webhook, json={
        "msgtype": "text",
        "text": {"content": message}
    })

# 每 5 分钟检测一次
while True:
    status = monitor_server()
    print(status)
    time.sleep(300)

import pymysql
from contextlib import contextmanager

@contextmanager
def get_conn():
    """上下文管理器:自动管理连接"""
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='password',
        database='mydb',
        charset='utf8mb4'
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# 批量插入
def batch_insert(data_list):
    with get_conn() as conn:
        with conn.cursor() as cursor:
            sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
            cursor.executemany(sql, data_list)
            print(f"插入了 {cursor.rowcount} 条记录")

# 使用
users = [
    ('张三', 'zhangsan@example.com', 25),
    ('李四', 'lisi@example.com', 30),
    ('王五', 'wangwu@example.com', 28),
]
batch_insert(users)

# 批量更新
def update_user_ages():
    with get_conn() as conn:
        with conn.cursor() as cursor:
            cursor.execute("UPDATE users SET age = age + 1 WHERE active = 1")
            print(f"更新了 {cursor.rowcount} 条记录")

import requests
import json

def call_api(endpoint, method='GET', data=None, headers=None):
    """通用 API 调用"""
    base_url = "https://api.example.com/v1"
    url = f"{base_url}/{endpoint}"
    
    default_headers = {
        'Authorization': 'Bearer YOUR_API_KEY',
        'Content-Type': 'application/json'
    }
    if headers:
        default_headers.update(headers)
    
    try:
        if method == 'GET':
            resp = requests.get(url, headers=default_headers, timeout=10)
        elif method == 'POST':
            resp = requests.post(url, headers=default_headers, 
                               data=json.dumps(data), timeout=10)
        elif method == 'PUT':
            resp = requests.put(url, headers=default_headers,
                              data=json.dumps(data), timeout=10)
        elif method == 'DELETE':
            resp = requests.delete(url, headers=default_headers, timeout=10)
        
        resp.raise_for_status()
        return resp.json()
    
    except requests.exceptions.RequestException as e:
        print(f"API 调用失败: {e}")
        return None

# 获取用户列表
users = call_api("users")
print(json.dumps(users, indent=2, ensure_ascii=False))

# 创建订单
new_order = call_api("orders", method='POST', data={
    "product_id": 123,
    "quantity": 2,
    "user_id": 456
})

from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
import PyPDF2

# 1. 生成 PDF
def create_pdf(filename, title, content_lines):
    """生成 PDF 报告"""
    c = canvas.Canvas(filename, pagesize=A4)
    width, height = A4
    
    # 中文字体(需要下载字体文件)
    pdfmetrics.registerFont(TTFont('SimSun', 'SimSun.ttf'))
    c.setFont('SimSun', 16)
    
    # 标题
    c.drawString(100, height - 100, title)
    
    # 内容
    c.setFont('SimSun', 12)
    y = height - 150
    for line in content_lines:
        c.drawString(100, y, line)
        y -= 20
        if y < 50:
            c.showPage()
            y = height - 50
    
    c.save()
    print(f"PDF 已生成: {filename}")

# 2. 合并 PDF
def merge_pdfs(output, inputs):
    """合并多个 PDF"""
    merger = PyPDF2.PdfMerger()
    for pdf in inputs:
        merger.append(pdf)
    merger.write(output)
    merger.close()
    print(f"PDF 已合并: {output}")

# 3. 提取 PDF 文本
def extract_text(pdf_path):
    """提取 PDF 文本"""
    reader = PyPDF2.PdfReader(pdf_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text()
    return text

# 使用
create_pdf("report.pdf", "测试报告", ["第一行", "第二行", "第三行"])
merge_pdfs("merged.pdf", ["file1.pdf", "file2.pdf"])

import click

@click.group()
def cli():
    """文件处理工具集"""
    pass

@cli.command()
@click.argument('directory')
@click.option('--prefix', default='file', help='文件名前缀')
@click.option('--ext', default=None, help='只处理指定扩展名')
def rename(directory, prefix, ext):
    """批量重命名文件"""
    from pathlib import Path
    path = Path(directory)
    files = path.glob(f"*.{ext}") if ext else path.iterdir()
    
    for i, f in enumerate(files, 1):
        if f.is_file():
            new_name = f"{prefix}_{i:03d}{f.suffix}"
            f.rename(f.parent / new_name)
            click.echo(f"✓ {f.name} → {new_name}")

@cli.command()
@click.argument('source')
@click.argument('target')
@click.option('--quality', default=70, help='压缩质量 1-100')
def compress(source, target):
    """压缩图片"""
    from PIL import Image
    img = Image.open(source)
    img.save(target, "JPEG", quality=quality, optimize=True)
    click.echo(f"✓ 压缩完成: {target}")

@cli.command()
@click.argument('url')
@click.option('--output', '-o', default='output.html')
def fetch(url, output):
    """下载网页内容"""
    import requests
    resp = requests.get(url)
    with open(output, 'w', encoding='utf-8') as f:
        f.write(resp.text)
    click.echo(f"✓ 已保存到 {output}")

if __name__ == '__main__':
    cli()

"""
每日销售报告自动化:
1. 从数据库拉取数据
2. 用 pandas 分析
3. 生成 Excel + 图表
4. 生成 PDF 报告
5. 邮件发送给管理层
"""

import pandas as pd
import pymysql
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

class DailyReport:
    def __init__(self):
        self.db_config = {
            'host': 'localhost',
            'user': 'report_user',
            'password': 'password',
            'database': 'sales'
        }
        self.email_config = {
            'smtp': 'smtp.gmail.com',
            'port': 587,
            'user': 'report@company.com',
            'password': 'app_password'
        }
    
    def fetch_data(self):
        """从数据库拉取昨日销售数据"""
        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        
        conn = pymysql.connect(**self.db_config)
        query = f"""
        SELECT order_id, product_name, category, amount, customer_id, created_at
        FROM orders
        WHERE DATE(created_at) = '{yesterday}'
        """
        df = pd.read_sql(query, conn)
        conn.close()
        return df, yesterday
    
    def analyze(self, df):
        """数据分析"""
        analysis = {
            '总订单数': len(df),
            '总销售额': df['amount'].sum(),
            '平均订单金额': df['amount'].mean(),
            '独立客户数': df['customer_id'].nunique(),
        }
        
        # 分类销售统计
        category_stats = df.groupby('category')['amount'].agg(['sum', 'count', 'mean'])
        
        # TOP 10 商品
        top_products = df.groupby('product_name')['amount'].sum().nlargest(10)
        
        return analysis, category_stats, top_products
    
    def generate_charts(self, df, date):
        """生成图表"""
        fig, axes = plt.subplots(2, 2, figsize=(14, 10))
        
        # 1. 各类别销售额
        category_sum = df.groupby('category')['amount'].sum()
        axes[0, 0].pie(category_sum.values, labels=category_sum.index, autopct='%1.1f%%')
        axes[0, 0].set_title('各类别销售额占比')
        
        # 2. 时段分布
        df['hour'] = pd.to_datetime(df['created_at']).dt.hour
        hourly = df.groupby('hour')['amount'].sum()
        axes[0, 1].bar(hourly.index, hourly.values)
        axes[0, 1].set_title('时段销售分布')
        axes[0, 1].set_xlabel('小时')
        
        # 3. 订单金额分布
        axes[1, 0].hist(df['amount'], bins=30, edgecolor='black')
        axes[1, 0].set_title('订单金额分布')
        axes[1, 0].set_xlabel('金额')
        
        # 4. TOP 10 商品
        top = df.groupby('product_name')['amount'].sum().nlargest(10)
        axes[1, 1].barh(top.index, top.values)
        axes[1, 1].set_title('TOP 10 商品')
        
        plt.tight_layout()
        chart_path = f'charts_{date}.png'
        plt.savefig(chart_path, dpi=100, bbox_inches='tight')
        plt.close()
        return chart_path
    
    def generate_excel(self, df, analysis, category_stats, top_products, date):
        """生成 Excel 报告"""
        excel_path = f'report_{date}.xlsx'
        
        with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
            # 概览
            pd.DataFrame([analysis]).T.to_excel(
                writer, sheet_name='概览', header=['数值'])
            
            # 明细
            df.to_excel(writer, sheet_name='订单明细', index=False)
            
            # 分类统计
            category_stats.to_excel(writer, sheet_name='分类统计')
            
            # TOP 商品
            top_products.to_excel(writer, sheet_name='TOP 商品')
        
        return excel_path
    
    def send_email(self, date, attachments, recipients):
        """发送邮件"""
        msg = MIMEMultipart()
        msg['Subject'] = f'每日销售报告 - {date}'
        
        html = f"""
        <h2>每日销售报告</h2>
        <p><strong>报告日期:</strong> {date}</p>
        <p>详细数据请查看附件。</p>
        <p><small>本报告由系统自动生成,请勿直接回复。</small></p>
        """
        msg.attach(MIMEText(html, 'html'))
        
        for file_path in attachments:
            with open(file_path, 'rb') as f:
                attach = MIMEApplication(f.read())
                attach.add_header('Content-Disposition', 'attachment',
                                 filename=file_path.split('/')[-1])
                msg.attach(attach)
        
        with smtplib.SMTP(self.email_config['smtp'], self.email_config['port']) as server:
            server.starttls()
            server.login(self.email_config['user'], self.email_config['password'])
            for recipient in recipients:
                msg['To'] = recipient
                server.send_message(msg)
                del msg['To']
        
        print(f"报告已发送给 {len(recipients)} 位收件人")
    
    def run(self):
        """执行完整流程"""
        print("开始生成每日报告...")
        
        # 1. 拉取数据
        df, date = self.fetch_data()
        print(f"  ✓ 拉取 {len(df)} 条订单数据")
        
        # 2. 分析
        analysis, category_stats, top_products = self.analyze(df)
        print(f"  ✓ 总销售额: ¥{analysis['总销售额']:.2f}")
        
        # 3. 图表
        chart_path = self.generate_charts(df, date)
        print(f"  ✓ 图表已生成: {chart_path}")
        
        # 4. Excel
        excel_path = self.generate_excel(df, analysis, category_stats, top_products, date)
        print(f"  ✓ Excel 已生成: {excel_path}")
        
        # 5. 邮件
        recipients = ['manager@company.com', 'ceo@company.com']
        self.send_email(date, [excel_path, chart_path], recipients)
        print("✓ 报告流程完成")

# 定时执行
if __name__ == '__main__':
    report = DailyReport()
    report.run()

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

def safe_operation(func):
    """装饰器:统一异常处理"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(f"{func.__name__} 执行失败: {e}", exc_info=True)
            # 发告警
            send_alert(f"任务失败: {func.__name__}")
    return wrapper

@safe_operation
def my_task():
    # 业务逻辑
    pass

import yaml

with open('config.yaml') as f:
    config = yaml.safe_load(f)

import logging

logger = logging.getLogger(__name__)

# DEBUG:调试信息
logger.debug("变量 x = %s", x)

# INFO:正常流程
logger.info("开始处理 %s 条记录", count)

# WARNING:可恢复异常
logger.warning("网络超时,准备重试")

# ERROR:严重错误
logger.error("数据库连接失败")

# CRITICAL:致命错误
logger.critical("系统无法继续运行")

from typing import List, Dict, Optional

def process_data(items: List[Dict[str, str]], 
                filter_key: Optional[str] = None) -> List[Dict]:
    """处理数据(带类型注解)"""
    if filter_key:
        return [item for item in items if filter_key in item]
    return items

import unittest

def add(a, b):
    return a + b

class TestMath(unittest.TestCase):
    def test_add(self):
        self.assertEqual(add(1, 2), 3)
        self.assertEqual(add(-1, 1), 0)
    
    def test_add_negative(self):
        with self.assertRaises(TypeError):
            add("1", 2)

if __name__ == '__main__':
    unittest.main()