导入脚本跑完,日志里只剩一句:
skip file: no valid rows
这种地方我一般不会先怀疑 pandas,也不会先怀疑编码。先看判断条件。很多 Python 代码写得啰嗦,不是因为业务复杂,是因为没用好 any() 和 all()。
any() 看的是:一组条件里,有没有一个是真的。
all() 看的是:一组条件里,是不是全都是真的。
别把它们想复杂了,就这两句话。
比如有一批订单要导入,最土的写法一般长这样:
defcan_import(row):
ifnot row.get("order_id"):
returnFalse
if row.get("amount", 0) <= 0:
returnFalse
if row.get("status") notin ("PAID", "REFUNDING"):
returnFalse
ifnot row.get("buyer_phone"):
returnFalse
returnTrue
这代码没错,但一堆 if 看着烦。条件再多两个,后面的人基本就开始乱加。
用 all() 可以收干净一点:
defcan_import(row):
return all([
bool(row.get("order_id")),
row.get("amount", 0) > 0,
row.get("status") in ("PAID", "REFUNDING"),
bool(row.get("buyer_phone")),
])
意思很直:这些检查都通过,这行数据才允许导入。
但这段我还不算最喜欢。因为这里用了列表,列表会先把里面所有表达式都算完,再交给 all()。如果里面有查库、调接口、读文件这种重操作,就亏了。
现场代码我更愿意这么写:
defcan_import(row):
checks = (
lambda x: bool(x.get("order_id")),
lambda x: x.get("amount", 0) > 0,
lambda x: x.get("status") in ("PAID", "REFUNDING"),
lambda x: bool(x.get("buyer_phone")),
)
return all(check(row) for check in checks)
all() 有短路特性。前面一个条件已经是 False,后面就不跑了。
这点在排查慢脚本时挺有用。我见过有人校验一行数据时,前面字段都空了,后面还去请求风控接口,几万行一跑,脚本慢得像在拉磨。
any() 正好反过来。它关心的是“有没有一个命中”。
比如导入前扫一遍,只要有一行金额异常,就直接打日志,不要等到入库时报错:
defhas_dirty_amount(rows):
return any(row.get("amount", 0) <= 0for row in rows)
用在业务判断里也很顺手。
defneed_manual_review(order):
flags = order.get("risk_flags", [])
return any(flag in flags for flag in (
"IP_CHANGED",
"DEVICE_CHANGED",
"HIGH_REFUND_RATE",
))
这比下面这种写法清爽:
if"IP_CHANGED"in flags or"DEVICE_CHANGED"in flags or"HIGH_REFUND_RATE"in flags:
returnTrue
条件少的时候无所谓,条件一多,or 会拉得很长,还容易漏括号。
有个坑要记一下,别线上踩。
print(all([])) # True
print(any([])) # False
第一次看到 all([]) 是 True,很多人会觉得怪。其实它的意思是:没有任何一个元素违反条件,所以认为全通过。
但业务里不能这么想。
比如你写:
defbatch_pass(rows):
return all(can_import(row) for row in rows)
如果 rows 是空列表,它会返回 True。这就很危险了。空文件被当成校验通过,后面可能还会打出“导入成功 0 条”这种很迷惑的日志。
我一般会把空数据单独拎出来:
defbatch_pass(rows):
ifnot rows:
returnFalse
return all(can_import(row) for row in rows)
再严一点,直接抛异常:
defbatch_pass(rows):
ifnot rows:
raise ValueError("import rows is empty")
return all(can_import(row) for row in rows)
还有一种写法,我在线上脚本里用得不少:一边扫,一边留下坏数据,不要只给一个 False。
defpick_bad_rows(rows):
bad_rows = []
for line_no, row in enumerate(rows, start=1):
ifnot can_import(row):
bad_rows.append({
"line_no": line_no,
"order_id": row.get("order_id"),
"reason": "required field missing or invalid amount",
})
return bad_rows
bad_rows = pick_bad_rows(rows)
if any(bad_rows):
print(f"import blocked, bad rows: {bad_rows[:3]}")
else:
print("import check passed")
这里 any(bad_rows) 其实就是判断列表里有没有元素。有人喜欢写 len(bad_rows) > 0,也可以。但我看到 any(bad_rows),第一眼知道它在表达“有没有问题数据”。
不过也别滥用。
像这种代码就有点装了:
if all([user_id]):
pass
就一个条件,用 if user_id: 就行。工具是拿来让代码变直接的,不是拿来显得自己会 Python 的。
any() 和 all() 最舒服的地方,是把一堆分散的判断收成一句业务话。
all():这一行数据是不是全部合格。
any():这一批数据里有没有一个危险信号。
写校验、过滤、开关判断、批处理脚本时,先想一下是不是能用它们。少写几层 if,后面排查问题时,眼睛也少受点罪。