ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

10个Python自动化脚本:文件整理、数据抓取与运维巡检实战

10个Python自动化脚本:文件整理、数据抓取与运维巡检实战 直接说结论这十个脚本不是我从什么开源项目里抄来充数的而是我这些年真正在电脑上跑过、帮自己和同事省下大量重复劳动的代码。标题叫“解放双手”一点都不夸张——每天整理文件、汇总报表、测设备、盯日志、抓页面数据这些活儿的共同点是规则明确、步骤固定、做起来不费脑但就是特别浪费时间。今天这篇我按使用频率最高的四个方向来拆文件办公、数据采集、系统运维、浏览器自动化。每个脚本我都尽量控制在二三十行内附上核心代码、运行前提和最容易踩的坑。文章偏实战建议你打开电脑跟着敲一遍。如果还没装Python先看第1章十分钟就能搞定环境。1. 准备工作环境、依赖与运行习惯1.1 Python环境安装与PATH配置先解决一个最常见的卡点很多人打开命令行输入python系统直接提示“不是内部或外部命令”。这基本就是安装时漏勾了“Add Python to PATH”。Windows安装就两步去官网下载64位安装包双击后第一屏务必勾选“Add python.exe to PATH”然后再点Install Now。装完打开新的命令行窗口输入python --version能打印出版本号就说明环境OK。Linux和macOS一般自带Python 3但版本可能偏旧建议通过系统包管理器装一个新的稳定版。我写脚本基本都基于Python 3.9以上如果你用的是3.6或更老有些语法和依赖可能会出问题能升就升。环境配好后我强烈建议再搭建虚拟环境。每个项目建一个独立环境避免不同项目依赖互相冲突。操作很简单python -m venv venv venv\Scripts\activate # Windows source venv/bin/activate # Linux / macOS后面所有pip install都装进这个虚拟环境里以后删了目录就能整体卸载不会把系统搞乱。1.2 常用依赖库清单下面所有脚本会用到的库我先汇总成一张表装的时候直接按需拿用途依赖库安装命令Excel合并/清洗pandas, openpyxlpip install pandas openpyxlPDF处理pypdfpip install pypdfSSH网络设备paramikopip install paramiko数据库查询sqlalchemy 对应驱动pip install sqlalchemy pymysql网页表格抓取pandas, requests, lxmlpip install pandas requests lxml浏览器自动化seleniumpip install selenium定时任务schedulepip install schedule安装速度慢或者超时的时候可以选一个离自己网络比较近的PyPI镜像源把命令里的下载地址换成镜像地址即可效果立竿见影。1.3 写自动化脚本的三个好习惯参考我之前写的那些脚本有几个习惯一直坚持第一脚本入口统一用ifname main:包起来。好处是既能直接运行也能被别人import后调用函数不会一导入就执行。第二头一次跑脚本优先做“只读”操作。比如文件归档先把move改成copy磁盘清理先只打印不删除确认无误再放开真正的动作。自动化脚本最怕的就是误操作安全永远排在效率前面。第三打印和日志一定要有。脚本跑完不是结束你得知道它干了什么、有没有出错。简单项目用print就好长期后台运行的脚本建议用logging写进文件。2. 文件与办公效率类先把整理工作干掉2.1 脚本一批量文件智能归档场景不用多说下载文件夹里堆了几百个文件简历、安装包、截图、PDF、压缩包混在一起。手动新建文件夹再拖动没个把小时下不来。核心逻辑是用扩展名分类让脚本自动创建子文件夹并移动文件from pathlib import Path import shutil src_dir Path(D:/downloads) # 改成你的目录 for item in src_dir.iterdir(): if not item.is_file(): continue ext item.suffix.lstrip(.).lower() or no_ext sub_dir src_dir / ext sub_dir.mkdir(exist_okTrue) target sub_dir / item.name if target.exists(): # 重名时加时间戳避免覆盖 target sub_dir / f{item.stem}_{item.stat().st_mtime:.0f}{item.suffix} shutil.move(str(item), str(target))为什么我推荐用pathlib而不是os.path因为pathlib的Path对象把路径拼接、后缀提取、目录创建都封装得特别直观跨平台也不容易出斜杠问题。代码里item.suffix拿扩展名、item.stem拿文件名主体比手动字符串split更可靠。如果按扩展名分类还不够想按日期归档也很简单用datetime.fromtimestamp(item.stat().st_mtime)取出修改日期再按年/月建文件夹即可。踩坑提示在这头一次运行一定要先把shutil.move改成shutil.copy跑一遍看看文件都去了哪里。我就见过有同事脚本里目录拼错把所有文件挪进了同名覆盖的深渊后悔都来不及。2.2 脚本二Excel报表自动合并与去重做运营、做数据分析的朋友应该深有体会每周从各个渠道导出十几张Excel格式差不多但要手动复制到一张总表里。合并本身不难难在合并前要对齐列名、去重、处理空值。用pandas十行以内解决import pandas as pd from pathlib import Path frames [] for f in Path(reports).glob(*.xlsx): df pd.read_excel(f, dtypestr) # 统一按字符串读入 df[来源文件] f.name # 记录来自哪个文件 frames.append(df) result pd.concat(frames, ignore_indexTrue) result result.drop_duplicates() result.to_excel(merged_report.xlsx, indexFalse)这里有两个细节我觉得特别关键第一是dtypestr。Excel里那些长编号、银行卡号、订单号如果不强制读成字符串pandas会自动识别成数字结果就是科学计数法好好的编号变得面目全非。除非你确认某列必须做数值计算否则先按字符串读最安全。第二是pd.concat之前要检查各表的列名是否一致。有些表叫“客户名称”有些叫“客户名”合并后这些列会变成两列数据就乱了。所以脚本前面加一步先打印每个文件的columns或者统一做一次重命名再进concat。空文件也会让脚本报错。我在循环里习惯加一句if df.empty: continue空表直接跳过避免.concat的时候报错。2.3 脚本三PDF批量合并与拆分PDF的合并拆分是行政、财务、法务岗的刚需。比如把几十份单页合同合成一本或者把一本大报告拆出指定页数发给不同的人。合并直接看代码from pypdf import PdfReader, PdfWriter from pathlib import Path file_list sorted(Path(pdfs).glob(*.pdf)) writer PdfWriter() for path in file_list: reader PdfReader(str(path)) for page in reader.pages: writer.add_page(page) with open(merged.pdf, wb) as f: writer.write(f)这里我特意用pypdf而不是老牌的PyPDF2。PyPDF2已经很久没大更新了pypdf是它的延续接口几乎一样但维护更积极新项目直接选pypdf就好。拆分更简单把指定范围的页塞进新writer再写出来即可。比如只保留第2页到第5页writer PdfWriter() reader PdfReader(big.pdf) for page in reader.pages[1:5]: writer.add_page(page) with open(part.pdf, wb) as f: writer.write(f)提示一点如果PDF设置了密码读取时传password参数reader PdfReader(encrypted.pdf, passwordyour_password)至于合并后书签丢失、压缩体积这类进阶需求pypdf不太擅长需要的话可以直接用其他专业PDF工具没必要让脚本硬扛。3. 数据采集与报表类让数据自己找上门3.1 脚本四数据库自动查询并生成Excel很多公司系统虽然能导数据但导一次要点五六次菜单还经常要登录超时。与其手动点不如让脚本直接连数据库把报表拉出来。我用SQLAlchemy做统一连接这样换数据库类型不用改太多代码import os import pandas as pd from sqlalchemy import create_engine engine create_engine( mysqlpymysql://{user}:{pwd}{host}:3306/{db}.format( useros.getenv(DB_USER), pwdos.getenv(DB_PASSWORD), hostos.getenv(DB_HOST), dbos.getenv(DB_NAME), ) ) sql SELECT department, COUNT(*) AS cnt FROM orders WHERE create_time CURDATE() GROUP BY department df pd.read_sql(sql, engine) df.to_excel(daily_summary.xlsx, indexFalse)直接把账号密码写在代码里是大忌。先不说代码会不会被同事传到网上光是把脚本放到Git仓库就是一个隐患。我习惯把敏感信息放到环境变量里操作系统里配好几个变量脚本通过os.getenv读取别人拿到代码也看不到密码。数据库相关有两个特别值得注意的点第一个查询数据量大的时候别一次全拉很容易把内存打爆。可以先加上LIMIT 10000验证脚本逻辑确认没问题再去掉限制或者分批按天/按月查询再合并。第二个涉及个人隐私或核心业务数据时先考虑是否需要脱敏。脚本处理完后如果只是本地自用也要注意报表别随手发到公共群聊。这是职业习惯更是责任问题。如果公司数据库是Oracle驱动改成oracledb或cx_Oracle连接字符串换上对应格式逻辑一样跑。3.2 脚本五公开网页表格一键抓取遇到那种网页上排好的表格数据如果没有下载接口很多人会手动复制粘贴。其实静态网页里的表格pandas自带一个非常方便的函数read_html可以直接把它解析成DataFrame。import pandas as pd url https://example.com/table.html # 替换成公开数据页面 tables pd.read_html(url) print(f页面中发现 {len(tables)} 个表格) for i, table in enumerate(tables): table.to_csv(ftable_{i}.csv, indexFalse, encodingutf-8-sig)为什么特别好用因为read_html底层帮你处理了HTML里table标签的解析不用自己写一堆BeautifulSoup查找代码一个函数就拿到了结构化数据。编码用utf-8-sig是给Excel准备的。如果直接存utf-8Excel打开中文会乱码加上BOM头之后Windows记事本和Excel都能正确识别。但read_html只能处理静态HTML。页面内容靠JavaScript动态加载的它会报错或抓到空表这时候就要上第5章的Selenium方案。另外强调的是合规性。这个脚本只适合抓取公开开放的页面数据抓之前看一眼网站的robots.txt请求间隔不要设得太短更不要拿自动脚本去采集用户隐私信息。自动化是为了提高效率不是让自己惹上麻烦。3.3 脚本六接口轮询与结果自动落地接口轮询最常见的场景是系统没有主动推送只能隔几分钟调一次接口发现有新数据就把结果记录下来。代码很直白import requests import time from datetime import datetime def pull_once(): response requests.get( https://api.example.com/status, headers{Authorization: Bearer your_token}, timeout15, ) response.raise_for_status() data response.json() row f{datetime.now().isoformat()},{data[status]},{data[count]}\n with open(status_log.csv, a, encodingutf-8) as f: f.write(row) while True: try: pull_once() print(f[{datetime.now()}] 拉取成功) except requests.RequestException as e: print(f[{datetime.now()}] 拉取失败: {e}) time.sleep(60)这里每个选择都有原因timeout15必须加否则接口假死时requests会无限等下去脚本就永远卡在那个网络请求上。raise_for_status的作用是返回码是4xx/5xx时立即抛异常而不是保留一个看似成功的响应继续处理。写文件用的是追加模式“a”不是覆盖模式“w”否则每次执行都会把上一次的数据清掉。如果脚本会长时间运行建议定期做日志轮转或者按日期生成文件避免单个文件无限增长。token这类鉴权信息同样别硬编码放环境变量比较稳妥。4. 系统运维与网络自动化巡检不再靠手动4.1 脚本七日志关键字扫描与告警排查问题时最枯燥的就是盯着日志找关键字。脚本做这件事非常合适而且可以做得比人还细。先写一个最简版本from pathlib import Path from collections import Counter log_path Path(app.log) keywords [ERROR, TimeoutException, OutOfMemory] def scan_log(): counter Counter() with log_path.open(r, encodingutf-8, errorsignore) as f: for line in f: for kw in keywords: if kw in line: counter[kw] 1 return counter result scan_log() for kw, count in result.items(): print(f[{kw}] 出现 {count} 次)逐行读取而不是一次性读整个文件是为了照顾上GB的大日志。readlines会把所有内容都塞进内存机器可能直接卡死for line in f则是流式读取内存占用很稳定。errorsignore也要养成习惯。日志文件编码经常不统一遇到个别乱码字节导致整个脚本报错就很冤。生产环境更常用的是“增量扫描”记录上次读到的文件位置每次只看新增部分。实现思路是维护一个偏移量文件offset_path Path(scan_offset.txt) offset int(offset_path.read_text()) if offset_path.exists() else 0 with log_path.open(r, encodingutf-8) as f: f.seek(offset) new_lines f.readlines() offset f.tell() offset_path.write_text(str(offset))这样脚本每分钟跑一次不会重复扫描全量日志效率高一个量级。配合第5章的schedule定时调度就能做成一个轻量巡检告警机器人。4.2 脚本八SSH批量巡检网络设备做网络运维的朋友应该最有共鸣。几十台交换机、路由器每天巡检登录一遍敲命令手都要断了。paramiko是Python里最成熟的SSH库用它写批量巡检是经典操作。import paramiko from datetime import datetime from concurrent.futures import ThreadPoolExecutor devices [ {ip: 10.0.0.1, user: admin, password: pass1, cmd: display version}, {ip: 10.0.0.2, user: admin, password: pass2, cmd: display version}, ] def run_one(device): client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( device[ip], usernamedevice[user], passworddevice[password], timeout10, ) stdin, stdout, stderr client.exec_command(device[cmd]) output stdout.read().decode(utf-8, errorsignore) with open(freport_{device[ip]}.txt, a, encodingutf-8) as f: f.write(f\n--- {datetime.now()} ---\n{output}\n) print(f{device[ip]} 巡检完成) finally: client.close() with ThreadPoolExecutor(max_workers10) as pool: pool.map(run_one, devices)为什么用ThreadPoolExecutor因为SSH连接是典型的IO密集操作大部分时间都耗在网络上多线程能让几十台设备同时巡检比一台一台串行快很多。加max_workers限制并发数不要无限开线程把设备连爆。set_missing_host_key_policy这个方法测试环境图方便用AutoAddPolicy可以但生产环境建议使用RejectPolicy并在known_hosts里预先记录设备指纹否则等于把所有主机都当可信主机存在中间人攻击风险。账号密码放在脚本里的问题我再强调一次。网络设备的密码泄露后果比数据库还严重建议至少放到独立的配置文件里并且设置严格的文件权限。另外想提醒的是虽然脚本能跑通思科、华为、H3C等主流设备的命令但每个厂商的交互方式、命令差异很大。如果设备数量多且品牌复杂直接用netmiko库会更省事它对主流厂商做了大量适配能自动处理分页和命令提示符我实际用下来稳定性比纯paramiko高不少。4.3 脚本九磁盘空间检查与过期文件清理服务器磁盘满是我们经常半夜被叫醒的原因之一。这个脚本干两件事检查磁盘剩余空间找出超过指定时间的旧文件。import shutil import time from pathlib import Path # 检查磁盘使用率 usage shutil.disk_usage(C:/) percent usage.used / usage.total * 100 print(f磁盘总容量:{usage.total/1024**3:.1f}GB,已用:{usage.used/1024**3:.1f}GB,使用率:{percent:.1f}%) # 找出超过30天的文件 base_dir Path(D:/data) max_keep_seconds 30 * 24 * 3600 for f in base_dir.rglob(*): if f.is_file(): age time.time() - f.stat().st_mtime if age max_keep_seconds: print(f旧文件: {f} ({age/86400:.1f} 天))rglob(*)会递归查找所有文件包括子目录里的内容。注意这里只做打印不真正删除。我强烈建议在自动删除这件事上保持克制。实际生产环境里文件删错了基本没有后悔药。我的习惯做法是先打印出所有待清理文件人工扫一眼确认没有重要内容再把f.unlink()那行的注释打开执行。如果条件允许更稳妥的做法是先把文件移动到统一的归档目录或回收站确认一周没问题后再彻底清理。还有一点经验磁盘检查脚本要尽量早跑、勤跑。放在每天凌晨执行早上上班前扫一眼昨天的报告磁盘要满也早就处理了不用等报警响了才手忙脚乱。5. 浏览器与任务调度把解放双手进行到底5.1 脚本十浏览器自动填表与提交最后一个脚本讲网页系统里反复填表的自动化。有些内部系统没有开放接口每周都要登录后台填一堆内容不用Selenium真不知道还能怎么办。Selenium是一个真正的浏览器自动化框架它会启动一个浏览器窗口像真人一样操作页面。from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC driver webdriver.Chrome() try: driver.get(https://example.com/login) wait WebDriverWait(driver, 10) username wait.until( EC.presence_of_element_located((By.ID, username)) ) username.send_keys(my_user) driver.find_element(By.NAME, password).send_keys(my_password) driver.find_element(By.CSS_SELECTOR, button[typesubmit]).click() # 等页面出现某个关键元素再继续下一步 wait.until(EC.presence_of_element_located((By.ID, welcome_msg))) print(driver.title) finally: driver.quit()为什么必须用WebDriverWait而不是time.sleep(5)固定等待时间短了不稳定长了浪费时间而且页面加载快慢受网络影响很大。显式等待是等条件成立就继续页面上元素一出现立刻往下走既稳又高效。定位元素时我一般优先用ID、Name、CSS选择器最后才考虑XPath。ID是页面唯一标识最可靠CSS选择器简洁直观XPath虽然功能强大但写不好就特别脆弱页面稍微变动就找不到元素。Selenium只做自动化操作不做灰产。这个脚本的适用范围是你自己有权限、日常要重复操作的业务系统比如测试环境数据准备、后台信息录入。千万别拿它做批量注册、刷单、恶意采集这类事情脚本无罪用法要合规。Chrome版本更新后经常出现浏览器驱动不匹配报错这个放在第6章常见问题里解决。5.2 chedule定时执行让脚本每天自己跑脚本写得再好如果每天还得手动双击运行那也只解放了一半。用schedule这个库可以像设闹钟一样让脚本按时执行import schedule import time def job(): print(定时任务开始……) # 这里调用你写的自动化函数 schedule.every().day.at(09:30).do(job) schedule.every(30).minutes.do(check_temp) schedule.every().monday.at(10:00).do(weekly_report) while True: schedule.run_pending() time.sleep(1)schedule的语法非常有表达力every().day.at指定每日时间every(30).minutes指定间隔分钟数every().monday指定周一。这个库只适合短任务如果需要复杂的集群调度、任务依赖、失败重试还是得用Celery或系统级任务调度。脚本需要长期挂机运行的话不能直接关掉终端。我在服务器上一般这样做用systemd或Windows任务计划程序注册成服务开机自启崩溃自动拉起。核心命令就一句把schedule脚本作为启动项运行即可。定时任务有个坑第一次部署时间设好后并不会马上执行很多人误以为脚本坏了。我的做法是第一次手动跑一遍完整流程再让它进入定时循环这样能确认脚本本身没问题排除时间设置错误的干扰。5.3 配置管理与敏感信息保护多个脚本共用一套数据库密码、接口地址、账号信息时你肯定不想改一个密码就翻遍所有脚本。我习惯用一个独立的config模块统一管理import os from pathlib import Path BASE_DIR Path(__file__).parent DB_HOST os.getenv(DB_HOST, localhost) DB_USER os.getenv(DB_USER) DB_PASSWORD os.getenv(DB_PASSWORD) API_TOKEN os.getenv(API_TOKEN)如果再配上python-dotenv可以把这些配置放在本地.env文件里而.env文件明确写进.gitignore不上传到代码仓库。这种做法的最大好处是开发环境用一套配置生产环境用另一套配置脚本代码不用改。我还建议在config里保留一个DEBUG开关。DEBUG True时脚本只打印执行计划和中间结果不执行真正的写入删除操作。写自动化脚本这几年这个习惯帮我避免了至少三次灾难性的误操作。6. 常见问题与避坑实录这部分直接把大家高频踩过的问题整理成一个速查表都是我实际见过甚至自己犯过的错问题现象根本原因解决办法import xxx 报ModuleNotFoundErrorpip装到了别的Python环境用python -m pip install xxx代替pip install虚拟环境内执行中文输出乱码控制台编码不匹配脚本开头加sys.stdout.reconfigure(encodingutf-8)Excel打开乱码文件没带BOMto_csv加encodingutf-8-sig中文路径找不到文件pathlib编码问题或路径写错用绝对路径打印Path对象确认实际路径Chorme无法启动浏览器版本和ChromeDriver不匹配使用webdriver-manager自动管理驱动版本页面元素定位失败页面还没加载完成用WebDriverWait替代time.sleeppip下载超时网络问题使用离自己近的镜像源加--timeout参数脚本跑到一半退出没有异常处理包try/except并写入错误日志还有一个我自己很有印象的问题pandas读取Excel时有些单元格会解析成NaN合并之后部分行变得空白去重也失效。解决方式是在读取后把所有NaN统一填充成空字符串比如调用df.fillna(, inplaceTrue)这样后续处理逻辑会更稳定。文件路径的问题也是高频重灾区。路径里带了空格、反斜杠或用了相对路径导致脚本换目录就找不到文件。我现在的习惯是所有脚本都用pathlib的Path来写路径尽量用绝对路径或基于脚本所在目录定位避免“从命令行运行正常双击脚本运行就报错”这种奇怪问题。结尾的一点个人经验写自动化脚本这件事我的体会是“先追求能跑再追求优雅”。刚开始你写的代码可能又笨又长这没问题功能先实现后面再一点点优化。真正重要的不是代码多漂亮而是能不能稳定解决实际问题、敢不敢交给它处理真数据。最后再分享一个小习惯我会在本地建一个scripts目录每个脚本旁边放一个README.txt记录这个脚本是干什么的、依赖哪些库、参数怎么改。时间一长你手头的脚本会越来越多回头翻查的时候这几行备忘比代码注释还顶用。祝大家少加班多摸鱼用脚本把重复劳动清零。
返回列表