一个充满反模式的文件操作模块分析与重构方法
标签： 2026-04-17 次
说下我们北京心玥软件这边的情况吧。我是开发工程师，平时就琢磨着怎么让团队代码别太烂——毕竟谁也不想接手个屎山对吧？我们组现在搞了CI、代码审查，还有静态分析、类型检查这些老几样，最关键的还是招对人，得有那两把刷子。反正最近项目推进挺顺，没出过大岔子。
早几年公司刚起步那会儿，09年吧，是从数据科学这块儿扎进去的。那时候请了一帮博士，个个算法玩得溜，写论文能发顶会，可软件工程这摊儿，压根没沾过边。他们写的代码，变量名全是x1、y2，函数一写几百行，注释？不存在的。现在回头看，当时堆的代码十有八九都废了，但架不住历史包袱重，总有些犄角旮旯的没清干净，上个月还从老系统里扒拉出段十年前的脚本，跑起来差点把测试库干崩。
重点说说我们组那个小李。她管着个关键系统，从开发到运维就她自个儿，没别人插手，也没人审过她的代码——按理说这不行，但大家都睁只眼闭只眼。为啥？她乐意啊！有时候半夜两点我还能瞅见她钉钉在线改bug，问就是“顺手的事儿”，周末也常往公司跑，说在家待着不踏实。上周六我去拿外卖，还撞见她在工位啃煎饼果子调参数，油点子溅到键盘上都没察觉。
人对工作那股劲儿，有时候我都觉得过了，跟打了鸡血似的。她是真聪明，脑子转得快，上次线上故障，别人还在翻日志，她三两句就定位到是缓存穿透，临时写了个布隆过滤器给怼上了。但论软件工程那套规范啊、架构设计啥的，也就比新手强点儿，顶多算“聪明”水平，离“专业”还差口气。比如她写的模块，功能倒都能实现，可接口设计得乱七八糟，想复用都得大改，我提过两次重构，她摆摆手说“能用就行，别整那些虚的”——嗨，随她吧，反正系统没崩，她自己也乐在其中。
file_io.py模块函数分析1
这让我们来到了一个名为utils/file_io.py的文件。从它的全名中你可以看出，这是一个充满“文件和IO”相关函数的“工具”模块。让我们看看其中的一些函数：
Python;toolbar:false">def del_rw(action, name, exc): """ See https://www.bjxykj.cn/questions/21261132/shutil-rmtree-to-remove-readonly-files Is used by shutil.rmtree to remove read only files as well """ os.chmod(name, stat.S_IWRITE) os.remove(name)
file_io.py模块函数分析2
我认为这是一个“也许这不应该是一个函数？”的案例。考虑到只有两行代码，直接包含这两行代码会更清楚你在做什么。此外，它使用了三个参数中的一个。
def safe_rmtree(remove_dir, minimum_parent_dir, min_parent_parts=2):
"""
Recursively removes all files in a directory that should contain the min parent dir
:param remove_dir:
:param minimum_parent_dir:
:param min_parent_parts: nr of folders that the parent should contain to avoid removing eg C:
:return:
"""
if not os.path.exists(remove_dir):
print(f"WARNING in safe_rmtree: {remove_dir} does not exist, nothing is removed")
return
remove_path = pathlib.Path(remove_dir)
minimum_parent_path = pathlib.Path(minimum_parent_dir)
if not minimum_parent_path.is_dir():
raise AssertionError("Min parent dir not a valid dir {}".format(minimum_parent_dir))
if len(minimum_parent_path.parts) <= min_parent_parts:
raise AssertionError(
"Unsafe removal of {}: minimum parent dir is too short {}".format(
remove_dir, minimum_parent_dir
)
)
if minimum_parent_path in remove_path.parents:
print("REMOVE {}".format(remove_dir))
shutil.rmtree(remove_dir, ｏｎｅｒｒｏｒ=del_rw) # use del_rw to also remove read only files
else:
raise AssertionError(
"Unsafe removal of {}: does not contain minimum parent dir {}".format(
remove_dir, minimum_parent_dir
)
)
time.sleep(0.1)我绝对不喜欢这个。我们至少使用了所有的参数。虽然文档并没有真正明确说明它们的作用，但如果目录包含在 minimum_parent_dir 下，并且 minimum_parent_dir 至少有 min_parent_parts 层深，那么这将删除该目录。
我大致明白为什么你希望对一个可能具有破坏性的操作进行一些限制，但我不得不怀疑为什么这个是我们添加的限制集。
另外，为什么最后要睡十分之一秒？shutil 不是在后台线程上执行操作，而是通过系统调用操纵文件系统。
现在，值得注意的是，我们正在使用pathlib.Path 与文件路径进行交互的API。这是在Python中正确的方法，这就是为什么这种方法很有趣：
def file_name_from_path(path, extension=False): """ Get the name of a file (without the extension) given a path :param path: :param extension: default false, but if true fn will include extension :return: a string with the file name (optional extension) """ basename = ntpath.basename(path) if extension is True: return basename else: return os.path.splitext(basename)[0]
这重新发明了已经存在的方法 pathlib.Path. 我还喜欢文档自相矛盾：这返回文件名（不带扩展名），除非它也返回扩展名。
def create_dir(new_dir, remove_data=False, verbose=False):
"""
Create a new directory. NOTE: this function only creates one new folder, the root
folder must already exist.
:param new_dir: directory to create
:param remove_data: if yes, remove existing data in folder.
:param verbose:
:return: a string containing the new directory
"""
if remove_data and os.path.exists(new_dir):
shutil.rmtree(new_dir)
time.sleep(0.1)
if os.path.exists(new_dir):
if verbose:
print("Output directory {} already exists".format(new_dir))
else:
os.makedirs(new_dir)
return new_dir该函数创建一个新目录并可能删除旧目录。不过，这并不是“安全”地进行的，但嘿，那个随机的sleep又回来了。
def dirs_in_dir(data_dir, contains=""):
"""
Get the paths of all directories in a directory containing a set of characters, the basename
:param data_dir: directory for which to list all paths with a base name
:param contains: characters that should be in the name of the directory
:return: a list of all directory paths
"""
if not os.path.exists(data_dir):
print("Warning in get_dirs_in_dir: directory {} does not exist".format(data_dir))
return
dir_paths = [
os.path.join(data_dir, d)
for d in os.listdir(data_dir)
if (os.path.isdir(os.path.join(data_dir, d)) and contains.lower() in d.lower())
]
return dir_paths这是对已经存在的功能的另一种重新实现pathlib.Path。我会说，“但他们已经知道这个！”但他们显然不知道；他们从StackOverflow上复制了代码，却没有阅读。
如何以最慢的方式将内容写入日志文件？
def write_to_log(text_string, file_path, level="INFO", code="I_NOTDEFINED", entry="PROJ"):
print(f"Log: {text_string}", flush=True)
if not os.path.exists(file_path):
file = open(file_path, "w")
file.close()
lookup_dict = error_lookup_dict
responsible = lookup_dict[code]["responsible"]
with open(file_path, "a") as file:
dt_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text = "{}; {}; {}; {}; {}; {}".format(
dt_string, entry, level, responsible, code, text_string + "\n"
)
file.write(log_text)大多数日志记录框架在你开始时打开一个文件，并在你完成日志记录之前一直保持文件句柄打开。这个框架不仅检查文件是否存在，而且每次写入时都会重新打开它。而且，它还会打印正在记录的内容。Python 有一个内置的日志记录框架，肯定可以处理所有这些功能。
最后，一个没有调用 shell 命令方式的库会是什么样子？file_io
def run_command_window(cmd, log_path=None):
std_output = []
updated_env = {**os.environ, "PYTHONPATH": str(config.repos_dir)}
with Popen(
cmd,
stdout=PIPE,
bufsize=1,
universal_newlines=True,
shell=sys.platform == "linux",
env=updated_env,
) as p:
for line in p.stdout:
print(line, end="", flush=True) # process line here
std_output.append(line)
if log_path:
with open(log_path, "a") as log:
log.write(line)
return p.returncode, std_output去年的一个项目上，接那个老系统自动化的时候，我碰见过一堆邪门代码，今儿想唠唠其中一个叫updated_env的文件。
这文件干的事儿特简单，就是往PYTHONPATH里塞东西——具体咋弄的呢？它是通过shell调Python代码，再起个新解释器跑。在我经手那堆烂摊子里，这算顶不错的了。
当时项目组就仨人，天天对着这代码挠头，后来有个还算痛快的结果：管项目的王哥门儿清这是个遗留垃圾，拍板要大改让它能接着用。就是不知道接这活儿的李工能不能扛住，别给逼疯了。
说实话，这代码一看就是那种“老子不懂Python但觉得自己贼聪明”的人写的。意思就是，写得是真烂，但胜在清楚。连那些压根不该存在的破代码，都码得整整齐齐，一眼能看明白。
我就琢磨啊，这不是更坑人嘛。现在想挑出该删的玩意儿，比预想的费劲多了。你想啊，总不能跟烧垃圾似的“一把火全扬了”，得先瞅清楚扔的是啥，别把有用的顺带燎了。
哦对，那文件里还有段注释写着“临时方案，别动”——结果这“临时”都三年了，估计写注释的人早跑路了。你说这叫什么事儿？
