每天一到公司,先被几十个文件搞晕,再对着Excel表格重复筛选复制,晚上回家还要盯着网页等更新。我用了小半年时间把这些破事全扔给Python,脚本一跑,该合并的合并,该归类的归类,该通知的通知,终于不用再做“人肉自动化”。这篇直接给出10个我用过的实用脚本,全部围绕真实业务场景,不是花架子Demo,拿来改改就能跑。适合每天被重复劳动拖住的人,也适合想通过具体项目入门Python的读者。
这些脚本覆盖文件整理、Excel处理、PDF操作、网页监控、系统运维、消息推送等常见需求。我会把每个脚本的用途、核心代码、使用注意点都写清楚,更重要的是告诉你为什么要这样写,踩过哪些坑,方便你根据自己的情况调整。
1. 十个自动化脚本,覆盖哪些场景
1.1 自动化脚本的选型理念
写自动化脚本之前,先别急着抄代码。我自己的经验是只挑三类任务做自动化:频率高、规则明确、操作机械。比如按文件类型分类、把几十张Excel合并成一张、每天定时备份目录,都属于这类。判断标准很简单:如果一件事你每周要做3次以上,每次步骤几乎一样,而且做错后容易发现,那么写脚本就划算。反过来,如果任务每次情况都不同,需要大量人为判断,就算能用脚本硬做,维护成本也会超过手工。
技术选型上,我的原则是先标准库,后第三方库。文件操作有os和shutil,路径处理有pathlib,大多数批量任务根本不需要引入重量级依赖。只有Excel、网页抓取、系统监控等场景才需要引入pandas、requests、psutil这类库。这样脚本的可移植性高,到新环境少踩依赖坑。
另一个理念是一个脚本只做好一件事。不要试图写一个万能工具,把下载、备份、发邮件全塞进去。拆成独立脚本,再通过一个总控入口或定时任务调用,出了问题能很快定位。后续维护成本低很多。
1.2 脚本清单总览
| 脚本编号 | 功能 | 主要依赖 | 典型场景 |
|---|---|---|---|
| 1 | 文件自动归类 | pathlib,shutil | 清理下载目录 |
| 2 | 批量重命名 | pathlib | 素材文件规范命名 |
| 3 | Excel多表合并 | pandas | 汇总月度报表 |
| 4 | PDF合并与文本提取 | pypdf | 电子文档归档 |
| 5 | 网页内容变化监测 | requests,hashlib | 价格或公告更新提醒 |
| 6 | 批量图片下载 | requests | 抓取公开素材 |
| 7 | RSS订阅抓取 | feedparser | 资讯聚合 |
| 8 | 磁盘与进程监控 | psutil,shutil | 服务器空间告警 |
| 9 | 定时增量备份 | zipfile,shutil | 重要目录每日备份 |
| 10 | 任务消息推送 | requests | 脚本跑完发通知 |
这张清单里的脚本不是孤立存在的,很多可以组合使用。例如脚本8监控磁盘后,可以调用脚本10发告警;脚本9备份完成后,也可以调用脚本10通知结果。下面按模块详细展开。
2. 文件与办公自动化,最值得先写的三类脚本
2.1 批量文件重命名与分类整理
我电脑的下载目录以前就是个垃圾场,截图、PDF、安装包混在一起,找文件只能靠搜索。于是写了下面这个“整理神器”。它做的事情很简单:扫描指定目录,按文件后缀归类到不同子目录,遇到重名自动加序号。
import shutil from pathlib import Path source_dir = Path.home() / "Downloads" mapping = { "图片": {".jpg", ".jpeg", ".png", ".gif", ".webp"}, "文档": {".pdf", ".docx", ".doc", ".xlsx", ".pptx", ".txt"}, "视频": {".mp4", ".avi", ".mov", ".mkv"}, "压缩包": {".zip", ".rar", ".7z", ".tar", ".gz"}, } for file in source_dir.iterdir(): if not file.is_file(): continue ext = file.suffix.lower() for folder, exts in mapping.items(): if ext in exts: target_dir = source_dir / folder target_dir.mkdir(exist_ok=True) target_path = target_dir / file.name # 处理重名:在文件名后加数字 if target_path.exists(): stem = file.stem idx = 1 while (target_dir / f"{stem}_{idx}{ext}").exists(): idx += 1 target_path = target_dir / f"{stem}_{idx}{ext}" shutil.move(str(file), str(target_path)) break这个脚本有两个设计细节值得说。第一,用Path对象而不是字符串拼接路径,跨平台不会出错,/操作符在Windows和Linux上行为统一。第二,写目标目录时用了mkdir(exist_ok=True),即使目录已经存在也不会报错。shutil.move自动处理跨盘移动,不需要自己复制再删除。
我实际运行中主要踩了三个坑:一是脚本把源目录里的子目录当成了文件,这里通过file.is_file()过滤掉了;二是某些文件名带特殊字符,Path对象能正确识别,但老式字符串拼接容易出问题;三是如果目标目录正好是源目录本身,比如把“文档”目录放在下载目录下,脚本遍历过程中会把新移动进去的文件又扫一遍,导致重复处理。解决方法是先把待处理的文件列表全部取出来,再遍历列表执行移动:
file_list = [f for f in source_dir.iterdir() if f.is_file()] for file in file_list: ...2.2 Excel多表合并与格式清洗
做数据汇总的人一定经历过这种折磨:领导发来12个月的报表,要求合并成一张总表,每个月列的顺序还不一样。我直接用pandas写脚本,几十张表秒合并,还顺便把空的单元格填上默认值。
import pandas as pd from pathlib import Path data_dir = Path("monthly_reports") all_files = list(data_dir.glob("*.xlsx")) if not all_files: print("目录下没有Excel文件") exit() df_list = [] for file in all_files: df = pd.read_excel(file, engine="openpyxl") # 统一将所有空值替换为“无” df = df.fillna("无") # 加入一列,标记数据来源文件名 df["来源文件"] = file.stem df_list.append(df) merged = pd.concat(df_list, ignore_index=True) merged.to_excel("merged_output.xlsx", index=False) print(f"合并完成,共{len(merged)}行")写这个脚本时要注意,pd.read_excel默认引擎是openpyxl,但如果你电脑只装了旧版xlrd可能会报错,最好显式指定engine="openpyxl"。另外,不同表格的列名必须一致,如果出现“销售额”和“销售金额”这种同义不同名的字段,concat后会出现两列数据。所以合并前最好先打印所有列名,统一规范。后续改进方向是加入类型转换和去重逻辑,比如把金额列统一转为数值,按订单号去掉重复行。
2.3 PDF合并与文本提取
PDF处理是办公自动化里的高频场景。我最早遇到的问题是报销时候要把多个PDF发票合并成一个文件,手动拖拽容易打乱顺序。后来用pypdf写了一个小工具,自动按文件名排序再合并,几十页文档秒级完成。
import pypdf import sys from pathlib import Path pdf_dir = Path("reports") pdf_files = sorted(pdf_dir.glob("*.pdf")) if not pdf_files: print("没有找到PDF文件") sys.exit(1) merger = pypdf.PdfMerger() for pdf in pdf_files: merger.append(str(pdf)) merger.write("merged.pdf") merger.close() print("合并完成")如果还需要提取PDF中的文本,用PdfReader:
import pypdf reader = pypdf.PdfReader("merged.pdf") for page in reader.pages: text = page.extract_text() # 按需处理text我踩过的坑主要是版本问题。老项目用PyPDF2,新版本推荐pypdf,两者的接口略有差异。另外,扫描版PDF没有文字层,extract_text()提取出来是空内容,这类文件需要OCR工具配合。所以先判断PDF是文本型还是扫描型再选工具,能省不少时间。
3. 网络数据与信息收集的实用脚本
3.1 定时网页内容变化监测与提醒
有时候需要盯着某个网页有没有更新,比如公告发布、商品降价、库存状态。手工刷新不现实,写一个轮询脚本是最优解。核心思路是定时抓取网页内容,计算哈希值,和上次比对,变了就触发通知。
import requests import hashlib import time import json url = "https://example.com/notice" local_hash_file = "page_hash.txt" webhook_url = "https://im.example.com/hook/your_key" def fetch_page(url): resp = requests.get(url, timeout=10) resp.encoding = resp.apparent_encoding return resp.text def check_change(webhook_url, msg): payload = {"msgtype": "text", "text": {"content": msg}} requests.post(webhook_url, json=payload, timeout=5) def main(): try: with open(local_hash_file, "r") as f: old_hash = f.read().strip() except FileNotFoundError: old_hash = "" html = fetch_page(url) new_hash = hashlib.md5(html.encode("utf-8")).hexdigest() if old_hash != new_hash: check_change(webhook_url, "页面内容发生变化,请及时查看") with open(local_hash_file, "w") as f: f.write(new_hash) while True: main() time.sleep(300)注意几个点。第一,必须设置timeout,否则网络异常时脚本会卡死。第二,有些网站直接抓取返回的是动态渲染后的空壳,需要用无头浏览器,那就要引入更重的工具。能用requests解决就不要上重型工具,降低维护成本。第三,resp.apparent_encoding会自动检测编码,中文页面不会出现乱码。我把这个脚本配合系统定时任务运行,每5分钟检查一次,稳定跑了几个月。
3.2 批量图片下载脚本
做内容运营时经常需要从某图库下载一批公开素材,一个个右键保存太痛苦。这个脚本用requests下载,带超时和重试机制,避免某张图片失败导致整个流程中断。
import requests from pathlib import Path save_dir = Path("images") save_dir.mkdir(exist_ok=True) # 示例URL列表,实际可从配置文件或爬虫结果读取 image_urls = [ "https://example.com/img/1.jpg", "https://example.com/img/2.jpg", ] def download_image(url, dest_path, retries=3): for attempt in range(retries): try: resp = requests.get(url, timeout=10) resp.raise_for_status() dest_path.write_bytes(resp.content) return True except (requests.RequestException, OSError) as e: if attempt == retries - 1: print(f"下载失败: {url},原因: {e}") return False for src_url in image_urls: file_name = src_url.split("/")[-1] download_image(src_url, save_dir / file_name)这里有个细节:resp.raise_for_status()会在返回状态码不是200时抛异常,避免把错误页面当图片存下来。如果图片很多,可以用ThreadPoolExecutor加多线程,但要注意目标网站的限流策略。至少我就是因为一次性开50个线程,导致IP被临时限制。后来加了每秒最多10个请求的控制,才稳定下来。
3.3 RSS订阅抓取与摘要生成
信息获取方面,我习惯用RSS聚合行业资讯,但客户端不好定制,于是写了一个脚本定时抓取RSS,生成一个简单的HTML摘要页面。
import feedparser from datetime import datetime feed_urls = [ "https://example.com/rss1.xml", "https://example.com/rss2.xml", ] articles = [] for url in feed_urls: feed = feedparser.parse(url) for entry in feed.entries[:5]: articles.append({ "title": entry.get("title", ""), "link": entry.get("link", ""), "published": entry.get("published", ""), }) html = ["<!DOCTYPE html><html><head><meta charset='utf-8'><title>资讯摘要</title></head><body>"] html.append(f"<h1>更新于 {datetime.now():%Y-%m-%d %H:%M}</h1>") for art in articles: html.append(f"<p><a href='{art['link']}'>{art['title']}</a> <small>{art['published']}</small></p>") html.append("</body></html>") with open("digest.html", "w", encoding="utf-8") as f: f.write("\n".join(html))使用feedparser解析RSS非常方便,它会自动处理命名空间和编码问题。这里有个比较坑的点:部分网站的RSS更新不需要刷新静态页面,直接抓取RSS文件内容即可,但有些站点限制了User-Agent,直接请求会返回403。解决办法是手动设置User-Agent,把自己伪装成浏览器。
4. 系统运维与效率工具脚本
4.1 磁盘空间与进程监控报警
服务器上的磁盘空间爆满是运维常见事故,与其等用户发现,不如写脚本每分钟检查一次,超过阈值立即告警。这个脚本同时监控了关键进程是否存在。
import shutil import subprocess import requests import time threshold_percent = 85 disk_path = "/" # 要监控的进程名列表 process_names = ["nginx", "mysqld"] webhook_url = "https://im.example.com/hook/alert" def send_alert(content): payload = {"msgtype": "text", "text": {"content": content}} requests.post(webhook_url, json=payload, timeout=5) def check_disk(): usage = shutil.disk_usage(disk_path) percent = usage.used / usage.total * 100 if percent > threshold_percent: send_alert(f"磁盘使用率 {percent:.1f}% 超过阈值 {threshold_percent}%") return percent def check_processes(): for name in process_names: # 通过pgrep检查进程是否存在 ret = subprocess.run(["pgrep", "-x", name], capture_output=True) if ret.returncode != 0: send_alert(f"进程 {name} 未在运行") while True: check_disk() check_processes() time.sleep(60)这里选择shutil.disk_usage而不是psutil,是因为标准库足够用,减少依赖。但如果你还要监控CPU、内存,psutil是更好的选择。pgrep是Linux工具,Windows上需要改成tasklist。所以这个脚本有平台限制,使用时根据自己的系统调整。另外,告警内容里一定要带具体指标和时间,否则收到告警还要查日志,效率很低。
4.2 定时自动备份重要文件
备份是“用时方恨少”的典型场景。我不建议直接在原目录上操作,而是把重要文件压缩成带时间戳的压缩包,再保留最近N份。
import zipfile import shutil from pathlib import Path from datetime import datetime source_dir = Path("important_files") backup_dir = Path("backups") backup_dir.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") zip_name = f"backup_{timestamp}.zip" zip_path = backup_dir / zip_name with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for file in source_dir.rglob("*"): if file.is_file(): # arcname保留相对路径,避免解压时堆到根目录 zf.write(file, arcname=file.relative_to(source_dir)) # 只保留最近10份,删除更早的备份 backup_files = sorted(backup_dir.glob("backup_*.zip"), reverse=True) for old_file in backup_files[10:]: old_file.unlink() print(f"备份完成: {zip_path}")压缩时用arcname参数让压缩包内保留完整相对目录结构,解压后不会全部混在一起。保留最近10份的逻辑,防止备份文件无限膨胀。如果你需要增量备份,可以先用shutil.copy2把变化文件同步过去,再做全量打包,效果也不错。我把这个脚本交给系统定时任务,每天凌晨执行,磁盘占用始终可控。
5. 邮件与消息自动化,减少重复沟通
5.1 自动发送带附件的邮件
月末要发十几封报表邮件,名单还不一样,手工操作又慢又容易漏。用smtplib自动发送,配合CSV里的收件人名单,一封封精确投递。
import smtplib import csv from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders import os smtp_server = "smtp.example.com" smtp_port = 465 sender = os.getenv("MAIL_USER") password = os.getenv("MAIL_PASSWORD") def send_mail(receiver, subject, body, attachment_path=None): msg = MIMEMultipart() msg["From"] = sender msg["To"] = receiver msg["Subject"] = subject msg.attach(MIMEText(body, "plain", "utf-8")) if attachment_path: with open(attachment_path, "rb") as f: part = MIMEBase("application", "octet-stream") part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", f"attachment; filename={os.path.basename(attachment_path)}" ) msg.attach(part) with smtplib.SMTP_SSL(smtp_server, smtp_port) as server: server.login(sender, password) server.sendmail(sender, receiver, msg.as_string()) with open("receivers.csv", "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: send_mail(row["email"], "月度报表", "请查收附件", "report.xlsx") print(f"已发送给 {row['email']}")强烈建议把邮箱账号和授权码放在环境变量里,而不是硬编码到脚本中。我见过有人把密码提交到代码仓库,后果相当严重。另外,发送频率较高时,可以提高server.login和sendmail之间的退避,防止被邮件服务商临时限流。如果只是给内部同事发通知,用下面这种Webhook方式更简单。
5.2 企业IM机器人消息推送
脚本跑完如何通知你?我经常用企业协作软件的Webhook机器人。实现起来很简单,本质就是向一个HTTP地址发送JSON消息。
import requests webhook_url = "https://im.example.com/hook/your_token" def send_msg(content, title="自动化任务通知"): headers = {"Content-Type": "application/json"} # 具体字段结构以你所用的协作软件文档为准 payload = { "msgtype": "text", "text": { "title": title, "content": content } } resp = requests.post(webhook_url, headers=headers, json=payload, timeout=10) resp.raise_for_status() if __name__ == "__main__": send_msg("Excel合并脚本已完成,请查看输出文件")Webhook方式的好处是真·跨语言,不懂Python也能通过curl调用。我一般习惯把这类通知函数单独写成一个模块,所有其他脚本需要告警时统一调用。这样以后换平台,只改一个文件就行。要注意的是Webhook地址本身是敏感信息,同样不要写死在代码里,建议用环境变量。
6. 常见问题与五大避坑经验
6.1 环境与依赖管理,避免“在我电脑上能跑”
Python脚本最容易出的问题就是“换个环境就跑不了”。解决方案很简单:每个项目建一个独立的虚拟环境,并把依赖清单导出。
python -m venv venv source venv/bin/activate # Windows使用 venv\Scripts\activate pip install requests pandas pypdf feedparser psutil pip freeze > requirements.txt到了新环境,执行pip install -r requirements.txt就能恢复依赖。我见过最惨的一次,是脚本在A机器上依赖的是pandas 1.x,而B机器上装的是pandas 2.x,结果合并Excel时行为都变了。所以,自动化脚本不只是写出来,还要跑得稳。把依赖固化和版本记录做好,运维成本可以低很多。
6.2 路径与编码问题,细节决定成败
路径处理优先用pathlib,库会自动处理不同操作系统的差异。不过要注意,某些第三方库接受字符串路径,不接受Path对象,看到报错时转为str()即可。
编码问题主要集中在Python 2遗留代码。现在基本都是Python 3,但读取文件时仍然建议显式指定编码,例如open("data.csv", "r", encoding="utf-8")。Windows下有些CSV文件默认是gbk,不加encoding就会乱码。如果是网页内容,使用resp.apparent_encoding可以自动检测出页面实际编码,中文乱码问题就能解决。
6.3 定时任务的正确姿势
脚本内部循环用sleep实现定时最简单,但不够可靠。脚本进程一旦退出就没人唤醒它。真正生产环境,Linux用cron,Windows用“任务计划程序”。比如每天凌晨2点备份,可以在crontab里这样写:
0 2 * * * cd /home/user/project && /home/user/project/venv/bin/python backup.py >> backup.log 2>&1写.log文件很重要,否则脚本报错你都不知道。也有人用schedule库在Python内部做定时调度,缩进、函数回调都要小心;如果定时任务需要跨重启运行,不推荐依赖它。我的习惯是脚本只负责单次执行,外部定时器负责按计划唤醒。
6.4 并发与超时控制,防止脚本挂死
写网络相关脚本时,requests不设timeout是最常见的“挂死”原因。一个请求一直等,脚本整体卡住。所以统一给所有外部请求加上timeout参数。如果做并发下载,用ThreadPoolExecutor时也不要开太多线程,保守一点,比如默认8个并发。同时要处理重试机制,用urllib3.Retry配合requests.Session()可以实现自动重试。
import requests from urllib3.util.retry import Retry from requests.adapters import HTTPAdapter session = requests.Session() retries = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) session.mount("http://", HTTPAdapter(max_retries=retries)) session.mount("https://", HTTPAdapter(max_retries=retries))这里backoff_factor的意思是每次重试间隔依次为0.5s、1s、2s,有效避免重试风暴。
6.5 日志与异常处理,不在沉默中失败
脚本不输出日志,出了问题就像“沉默的故障”。我的做法是每个重要脚本都加logging配置,并且把异常捕获打印出来。
import logging import traceback logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("script.log", encoding="utf-8"), logging.StreamHandler() ] ) try: # 业务逻辑 pass except Exception: logging.error(traceback.format_exc()) # 此处可调用消息推送脚本通知自己异常信息traceback.format_exc()能拿到完整的堆栈,光靠print(e)无法定位问题。日志同时输出到文件和控制台,调试和事后排查都方便。有了完整日志,脚本凌晨跑挂了,第二天也能根据时间点快速定位。
我个人最大的体会是,自动化脚本不是一次性交付的玩具,而是需要持续维护的“小工具”。选型时优先标准库,代码写清楚注释,把密码和地址放到环境变量,给每个脚本留好日志出口,这样它们才能真正成为“解放双手”的长久帮手。最后再分享一个小技巧:把这10个脚本统一放在同一个目录下,各自保留入口函数,再写一个带参数的main.py集中调用,这样就慢慢形成了自己的效率工具箱。后续你想加命令行交互、可视化界面,或者做更复杂的工作流,都能在这个地基上继续扩展。