我跟你说啊,最近真的是被公司内部小工具救了命,不然早被各种“帮我导个数”“搞个接口”“查下日志”给折磨秃了。你们是不是也一样,整天被人拉小黑屋:“这个能不能自动化一下呀,小小一个需求,很简单的。”
结果一看,简单个鬼。
我后来干脆认命:既然躲不掉,那就用 Python 把这些活全干了,顺手还搞了一套“内部工具军火库”。今天就边唠嗑边跟你说 7 个库,真的是,配合起来能把公司内部工具从零拉满。
先说最常用的那个场景:领导突然在走廊撞到你,“那个啥,有没有个命令,一键清一下测试环境脏数据?”你嘴上说没问题,心里骂骂咧咧回工位,打开 PyCharm。
我一般第一反应就是上命令行工具,手搓脚本是肯定要后悔的。这时候 click/typer 就登场了。
我给你看个最常用的那种“小黑盒”脚本,专门清理测试库的垃圾订单,跑之前一定要三思人生那种:
import typer
import sqlalchemy as sa
app = typer.Typer()
def get_db():
engine = sa.create_engine("mysql+pymysql://user:pwd@127.0.0.1:3306/test")
return engine
@app.command()
def clean_orders(env: str = "test", dry_run: bool = True):
"""清理环境里的脏订单,慎用,慎用,再慎用"""
engine = get_db()
with engine.begin() as conn:
sql = "select id from orders where status = 'DIRTY'"
ids = [row.id for row in conn.execute(sa.text(sql))]
typer.echo(f"准备删除 {len(ids)} 条订单,环境={env}")
if dry_run:
typer.echo("dry_run = True,只打印不删除")
return
conn.execute(sa.text("delete from orders where id in :ids"), {"ids": tuple(ids)})
typer.echo("删除完成")
if __name__ == "__main__":
app()
就这种东西,给测试同学一个命令:
python tools.py clean-orders --env test --dry-run False
你就从“数据库苦力”变成了“内部平台开发”。typer 语法比 click 轻松一点,反正俩都是一家的,看你心情用谁就行。
然后你很快会发现:命令行太丑了,别人一看黑底白字就害怕,说你这是“黑客工具”用不来。那咋办?我一般第二个就把 rich 抄出来,终端美化神器,进度条、表格、彩色日志,全给你整明白。
比如刚才那个删除工具,顺手加个进度条,用户一看就放心很多:
from rich.progress import track
from rich.console import Console
console = Console()
def delete_orders(conn, ids):
for order_id in track(ids, description="疯狂删除中..."):
conn.execute(sa.text("delete from orders where id=:id"), {"id": order_id})
console.log(f"[green]删除完了,一共搞掉 {len(ids)} 条[/green]")
你只要在公司里多跑几次这种“带进度条”的脚本,大家自动会觉得:哎,这玩意儿是“系统级”的,不是你瞎写的小脚本。气质立马就不一样。
第三种就是那种“大家都在抢你的接口”的场景。比如运营要查某个用户最近 30 天行为,客服要模糊查手机号,测试要造数据,每个人都想让你写个 SQL。人一多,你就会想:要不我搞个内部查询系统算了。
这时候我一般直接用 FastAPI 搭个小服务,比你手写 Flask 少吵一点——不是说 Flask 不好啊,就是 FastAPI 那个自动文档真的太省嘴皮子了。
随手写个内部用户查询接口,大概这样:
from fastapi import FastAPI, Query
from pydantic import BaseModel
import sqlalchemy as sa
app = FastAPI(title="内部查询工具")
engine = sa.create_engine("mysql+pymysql://user:pwd@127.0.0.1:3306/company")
class User(BaseModel):
id: int
name: str
phone: str | None = None
@app.get("/user", response_model=list[User])
def search_user(keyword: str = Query(..., min_length=2, max_length=20)):
sql = """
select id, name, phone from user
where name like :kw or phone like :kw
limit 50
"""
with engine.begin() as conn:
rows = conn.execute(sa.text(sql), {"kw": f"%{keyword}%"}).mappings()
return [User(**row) for row in rows]
你把这个挂在内网,大家自己打开 /docs 玩去,谁还来找你要 SQL。到后面你再慢慢加查询、导出、权限啥的,一不小心就长成半个“内部运营平台”了,简历上直接写“内部中台系统”,完全不过分。
说完接口,肯定要聊报表。只要公司有 Excel,就跑不掉。每次月初财务喊你导个“汇总表”,你手动拉一次就会发誓:下一次一定要用 pandas 和 openpyxl 搞定,不然真的会疯。
典型场景:从数据库里把订单拉出来,按部门、按月份聚合一下,再塞回 Excel 给别人玩。大概是这个味道:
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine("mysql+pymysql://user:pwd@127.0.0.1:3306/company")
sql = """
select dept_name, pay_amount, date(pay_time) as pay_date
from orders
where pay_time >= :start and pay_time < :end
"""
with engine.begin() as conn:
df = pd.read_sql(sql, conn, params={"start": "2024-01-01", "end": "2024-02-01"})
report = (
df.assign(month=lambda d: d["pay_date"].astype("datetime64[ns]").dt.to_period("M"))
.groupby(["dept_name", "month"])["pay_amount"]
.sum()
.reset_index()
)
report.to_excel("部门月度收入报表.xlsx", index=False)
你发给财务之后,对方一般会说一句:“哎这个不错,后面每个月记得帮我跑一下哈。” 这个时候你千万别嘴快说“好”。你要装作若有所思:“我给你做个小程序,点一下自己下。” 然后默默在脚本最后加一个 if __name__ == '__main__':,再顺手接一个 typer 命令。以后只需要教 TA 输入:
python monthly_report.py --month 2024-01
你就从“报表生成器”升级成“报表平台支持”。
说到数据库,刚才其实已经偷偷用了 SQLAlchemy 了。这个库吧,说实话一开始看着有点啰嗦,但你只要踩过几次“纯 SQL + 复制粘贴”的坑,比如改库地址忘了一个脚本,线上和测试混着写,那你会立刻爱上这种“统一入口”。
最典型的一个用法就是——把数据库连接配成环境变量,然后所有内部工具都走同一个 get_engine(),迁库给自己省掉无数眼泪:
import os
import sqlalchemy as sa
def get_engine():
url = os.getenv("COMPANY_DB_URL", "mysql+pymysql://user:pwd@localhost:3306/company")
return sa.create_engine(url, pool_pre_ping=True, pool_recycle=3600)
以后你任何小工具,不管是清数据、导报表、查配置,全都 import 这一句。领导突然说要从 MySQL 换成 Postgres,你改一个环境变量就全好了,脸不红气不喘(内心还是骂人的)。
上面这些都还是“人点一下就执行”的。但公司里还有一类很烦的小事:每天定时拉个第三方接口、刷新一下某张统计表、清理七天前的日志。这些之前都是我手工点,后来某天半夜被运营电话吵醒,让我“帮忙重跑一下脚本”,我一气之下把 schedule 也搬进来了。
这个库比那种正儿八经的 APScheduler 要轻很多,搞内部小工具刚刚好:
import schedule
import time
def sync_third_party():
print("开始同步外部系统数据...")
# 省略一堆调用逻辑
print("同步完成")
schedule.every().day.at("02:00").do(sync_third_party)
if __name__ == "__main__":
print("小调度器已启动")
while True:
schedule.run_pending()
time.sleep(1)
你把这个脚本丢一台不太重要的服务器上,用个 supervisor 或者 systemd 拉起来,就相当于自己在公司里造了个迷你版“任务调度中心”。虽然土点,但好用,关键是你能睡完整个觉。
最后,还得说一个经常被忽略的小心机:你这些内部工具写着写着,迟早会遇到“数据结构越来越复杂”的问题——比如某个配置要 30 多个字段,从不同地方拼出来。这个时候一定要上 pydantic 把数据规格钉死,不然全是 if else 判空,早晚要出事。
举个暴力点的例子:生成一个发给其他系统的“工单”对象,先用模型描述清楚:
from pydantic import BaseModel, Field, validator
class Ticket(BaseModel):
id: int
title: str = Field(min_length=3)
creator: str
priority: int = Field(ge=1, le=5)
tags: list[str] = []
@validator("title")
def strip_title(cls, v):
return v.strip()
def send_ticket(raw: dict):
ticket = Ticket(**raw)
# 这里再调用内部接口发出去
print("准备发送工单:", ticket.json(ensure_ascii=False))
有了这个,你就可以很放心地把“工单生成”这事交给别的同事,哪怕他乱传点字段,至少你的脚本会在一开始就抛错,而不是半夜在生产环境给你整一堆奇怪的脏数据。
唉我扯了半天,发现差不多把我这两年写内部工具最常抱团取暖的那几位都点名了: 命令行这边靠 typer/click,好看一点拉上 rich,对外服务扔给 FastAPI,数据全交给 pandas + openpyxl + SQLAlchemy,定时任务丢给 schedule,数据结构再用 pydantic 扎个笼子。
你随便捡三四个组合一下,就能搞一个“还挺像那么回事”的公司内部工具。 别人看你就是“搞平台的”,只有你自己知道——就是几堆库、几百行 Python,再加一点点不想加班的决心。
行了,我这会儿得去改一个昨天自己写的脚本 bug……你先挑一个库试着在公司整一个小工具出来,回头踩坑了再来跟我吐槽。