凌晨 1 点,脚本又在那台 机器上卡住了。
日志只剩一行:
[WARN] report_2026_06_25.xlsx not found, retry 5就是这种活儿, 在我瞅见的第一时间, 手就不想去点它。点一回, 那叫配合业务, 可要是天天都点, 这不就等于是把人当成那种按期执行, 无需思考的任务来使唤。而最令人觉得畅快的地方恰恰就在这儿, 虽说它或许并没有多高端, 可难得的是它能够把这些不那么光鲜好看的活儿承接过去。
如下这九个物件, 于平常之际我着实会加以运用。并非是用以充当黑客, 主要成因在于当它们执行任务之时, 屏幕呈现出接连不断的闪动状态, 的确略微带有那般特别的感觉。
1、:先把文件摸清楚
前些时候去写文件路径, 我特别厌烦那好多字符串进行拼接, 当中有斜杠, 还有反斜杠, 并且目录不存在, 大半夜的时候, 最容易被这种小事情给恶心到一回。
现在我基本先用 把目录扫一遍。
from pathlib import Path root = Path(r"D:\daily_report") bad_files = for f in root.rglob("*.csv"): if f.stat.st_size == 0: bad_files.append(f.name) print("空文件:", bad_files)这个物件瞅着小巧, 切实很强势。特别是大量地寻觅日志, 搜寻报表, 找寻导入失败的文件, 切莫先去动用复杂的框架, 连目录都还没弄清楚, 后续全是盲目地查找。
2、:文件归档别再手拖
运营丢过来的文件经常长这样:
order_20260625.csv order_20260624.csv refund_20260625.csv手动建目录、复制、改名,很容易漏。
from pathlib import Path import shutil src = Path(r"D:\receive") dst = Path(r"D:\archive") for file in src.glob("*.csv"): day = file.stem.split("_")[-1] target_dir = dst / day target_dir.mkdir(parents=True, exist_ok=True) target = target_dir / file.name shutil.move(str(file), str(target)) print("moved:", file.name, "=>", target_dir)这般脚本, 无需写得花哨, 重点在于能够重复运行, 运行完毕存有输出, 遇到问题能够知晓卡在何处。
3、:接口别靠浏览器点
有些后台页面, 看上去呈现为按钮,然而实际上其背后所包含的是一个接口, 在遭遇批量补数据这一情况时, 我通常最先去抓取一番请求, 随后再进行相关操作的调用。
import requests token = "替换成自己的内部token" ids = ["A1021", "A1022", "A1023"] for order_id in ids: resp = requests.post( "https://internal.example.com/api/retry", json={"orderNo": order_id, "source": "ops_script"}, headers={"Authorization": f"Bearer {token}"}, timeout=8 ) print(order_id, resp.status_code, resp.text[:80])我相对而言比较嫌弃那种没有特定内容的代码, 线上网络一旦出现抽风状况, 脚本就会停在那里好似装死一般, 对于接口自动化来说, 特定内容是必须要携带的。
4、:Excel 不是人肉数据库
只要和财务、运营、客服打交道,Excel 迟早绕不开。
比如每天给一份退款表,要把状态列补出来:
from openpyxl import load_workbook wb = load_workbook("refund.xlsx") sheet = wb["Sheet1"] for row in range(2, sheet.max_row + 1): amount = sheet[f"C{row}"].value reason = sheet[f"D{row}"].value if amount and amount > 5000: sheet[f"E{row}"] = "需要复核" elif reason and"重复"in reason: sheet[f"E{row}"] = "疑似重复退款" else: sheet[f"E{row}"] = "自动通过" wb.save("refund_checked.xlsx")别小瞧这类活, 省下来的并非仅仅是几分钟, 而是能少背一些锅, 人工去改Excel, 最怕的是改了错一行却浑然不知。
5、:脏数据先洗一遍
有空格存在于CSV里, 手机号带有横杠, 金额之中混着逗号, 这般数据直接进入数据库, 我通常对其不太相信, 是这样的情况。
import pandas as pd df = pd.read_csv("users.csv") df["phone"] = ( df["phone"] .astype(str) .str.replace("-", "", regex=False) .str.strip ) df["amount"] = ( df["amount"] .astype(str) .str.replace(",", "", regex=False) .astype(float) ) dirty = df[df["phone"].str.len != 11] dirty.to_csv("bad_users.csv", index=False) df[df["phone"].str.len == 11].to_csv("clean_users.csv", index=False)放到这儿, 我通常会把那些有问题的数据单独挑出来, 给它弄出去。可别那个脚本一运行起来, 就悄无声息地全给收纳进去了, 到后面要是业务方面找你核对账目, 那情况可就会变得很糟糕哇。
6、:目录一有文件就开干
有些系统特别土气, 上游把文件扔到目录里, 下游会定时进行扫描, 你得盯着文件是否到达, 使用起来很不方便。
from pathlib import Path from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import time classCsvArrived(FileSystemEventHandler): defon_created(self, event): file = Path(event.src_path) if file.suffix == ".csv": print("新文件到了:", file.name) # 这里可以接清洗、校验、入库脚本 time.sleep(1) print("处理完成:", file.name) observer = Observer observer.schedule(CsvArrived, r"D:\incoming", recursive=False) observer.start try: whileTrue: time.sleep(3) except KeyboardInterrupt: observer.stop observer.join这个东西运行起来真的好似监控程序一般。窗口一动不动, 文件一旦到来即刻触发, 相较于人去盯着文件夹要可靠许多。
7、:别再手动跑日报
在Linux上面, 我更加倾向于喜欢。然而, 针对一些体积较小的机器, 以及临时性质的任务而言, 使用 已然足够了。
import schedule import time from datetime import datetime defmake_daily_report: print("开始生成日报:", datetime.now.strftime("%F %T")) # 查库、读CSV、生成Excel都可以塞这里 print("日报完成") schedule.every.day.at("08:40").do(make_daily_report) whileTrue: schedule.run_pending time.sleep(1)留意, 这类脚本别当作核心调度系统来使用, 它适宜于轻任务、小场景, 真要是到了生产关键链路, 那就得规规矩矩地启用调度平台。
8、:批量上机器看日志
有时候问题不在代码,在机器上。
依次对十几台服务器, 一台一台地通过SSH上去查看日志, 当点到第三台的时候, 人就有一种异常难受的感觉了。这件事情是能够被否决掉的。
import paramiko hosts = ["10.8.1.21", "10.8.1.22", "10.8.1.23"] for host in hosts: ssh = paramiko.SSHClient ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy) ssh.connect( hostname=host, username="ops", key_filename=r"C:\keys\ops_id_rsa", timeout=6 ) cmd = "tail -n 20 /data/app/logs/error.log" _, stdout, stderr = ssh.exec_command(cmd, timeout=10) print("\n====", host, "====") print(stdout.read.decode("utf-8", errors="ignore")) err = stderr.read.decode("utf-8", errors="ignore") if err: print("ERR:", err) ssh.close此处千万别把密码固定写在脚本之中, 倘若能够运用密钥那就使用密钥, 脚本一旦被扔出去之后, 明文形式的密码早晚都会变成事故。
9、:最后才用的笨办法
能调接口就别模拟鼠标。能读文件就别截图识别。
但有些老系统就是没接口,只能点。那就让 点。
import pyautogui import time pyautogui.hotkey("ctrl", "l") pyautogui.write("https://internal.example.com/report", interval=0.01) pyautogui.press("enter") time.sleep(3) pyautogui.click(1280, 360) # 导出按钮位置 time.sleep(1) pyautogui.click(1180, 620) # 确认下载这玩意儿通常于末尾放置, 鉴于分辨率出现变动、窗口位置产生变化、网页加载速度减缓, 它均存在出现问题的可能性, 然而针对那些由来已久的后台情况, 它确实能够起到应急的作用。
自动化真正爽的地方,不是代码多漂亮。
是你将一项存在重复情况、显得无聊且极易出现差错的工作, 转化成一个脚本。在运行之前查看一下输入内容, 运行完毕之后查看一下日志记录, 而它们二者之间那一堆机械性质的劳动就消失不见了。
屏幕上刷刷刷输出:
moved: order_20260625.csv clean_users.csv generated 10.8.1.21 error.log checked这时候你会有点错觉:像黑客。
其实不是。
只是终于不像人工客服了。