跳转到内容
新建笔记

Python os:路径、文件读写与进程环境

os 提供进程环境、目录和操作系统接口;os.path 处理本机风格的路径;文件内容通常通过内置 open() 或 pathlib 读写。本页将路径计算、实际文件操作和启动子进程分开,避免把字符串处理结果当作文件已经存在或操作已经成功。

示例以 Python 3.11 为基线。写入、复制和删除演示都只操作示例自己创建的临时文件;迁移到真实数据目录时,应先确定目标、覆盖策略和失败后的处理方式。

路径拼接不等于访问文件

跳转到“路径拼接不等于访问文件”
API作用容易误解的地方
os.getcwd()当前进程工作目录不一定是脚本所在目录
os.path.join()按本机路径规则组合片段后续绝对片段可能替换前面的部分
os.path.abspath()通常按当前工作目录形成规范化绝对路径不要求路径存在,也不保证解析符号链接
os.path.dirname() / basename()拆分目录部分与最后一部分末尾有分隔符时 basename() 可为空
os.path.split()返回 (目录部分, 最后一部分)不检查文件系统
os.path.splitext()拆出最后一个扩展名archive.tar.gz 的扩展名是 .gz
os.path.relpath(path, start)计算相对表示不要求目标存在;Windows 跨盘符可能报错
exists() / isfile() / isdir()查询当前文件系统状态查询与后续操作之间仍可能发生变化

os.path 随运行平台选择规则。分析另一平台的路径字符串时,用 posixpath 或 ntpath 明确指定,不能用本机一次实验推断所有系统。

import ntpath
import posixpath
assert posixpath.join("/data/run", "/tmp/result.txt") == "/tmp/result.txt"
assert posixpath.split("/home/user/file.txt") == ("/home/user", "file.txt")
assert posixpath.splitext("archive.tar.gz") == ("archive.tar", ".gz")
assert posixpath.basename("/data/run/") == ""
assert posixpath.relpath("/a/b/c/file.txt", "/a/b") == "c/file.txt"
assert ntpath.join(r"C:\work", r"D:\data\x.txt") == r"D:\data\x.txt"
assert ntpath.join(r"C:\work", r"\logs") == r"C:\logs"
assert ntpath.join("C:", "logs") == "C:logs"
print("POSIX and Windows lexical path examples passed")

Windows 的 C:logs 是盘符相对路径,不能当作 C:\logs。只有根分隔符而没有盘符的片段还有自己的规则。os.sep 才是本机主路径分隔符,原稿的 os.rep 不存在;通常无需手工拼接它。

abspath() 和 normpath() 不展开 ~ 或环境变量,分别需要 expanduser()、expandvars();展开结果也不保证存在。普通脚本中可用 Path(__file__).resolve().parent 得到源文件附近的目录,但 Notebook 不一定定义 __file__。如果路径涉及符号链接,词法规范化与实际文件解析也可能产生不同含义。Python os.path 文档

os.chdir(path) 成功返回 None,失败抛出 OSError 的子类,如 FileNotFoundError、NotADirectoryError、PermissionError;它不会用 True/False 表示结果。工作目录是进程状态,线程和使用相对路径的代码会共同受到影响。

此示例只进入自己创建的临时目录,并在清理临时目录前恢复原目录:

import os
from pathlib import Path
from tempfile import TemporaryDirectory
before = Path.cwd()
with TemporaryDirectory(prefix="vitalogos-cwd-") as directory:
target = Path(directory).resolve()
try:
result = os.chdir(target)
assert result is None
assert Path.cwd().resolve() == target
finally:
os.chdir(before)
assert Path.cwd() == before
print("working directory restored")

这是单线程演示,不是可供并发服务使用的隔离机制。应用通常可以直接保存基准目录并使用绝对路径,从而减少对 chdir() 的依赖。

遍历与文件操作的准确语义

跳转到“遍历与文件操作的准确语义”

os.listdir(path) 返回直属条目的名称,不自动补上父目录,也不保证排序;可能同时包含文件、目录和链接。os.walk() 逐层给出 (root, dirs, files),可在自顶向下模式修改 dirs[:] 来剪枝。它默认不跟随目录符号链接,默认也会忽略某些遍历错误;需要完整清单时应提供 onerror 并处理失败。

操作成功后的含义
os.mkdir(path)新建单级目录;缺少父目录时失败
os.makedirs(path, exist_ok=True)按需新建父目录;已有普通文件仍不能冒充目录
os.remove(path) / unlink(path)删除文件或链接目录项,不用于删除普通目录
os.rmdir(path)删除空目录
os.removedirs(path)删除末端空目录,再尝试移除空的父目录,影响范围可能向上扩展
shutil.rmtree(path)删除整个目录树;与删除一个空目录的语义不同
os.rename(src, dst)重命名/移动;目标已存在时的行为存在平台差异
os.replace(src, dst)用明确的替换语义重命名,仍可能因跨文件系统、权限等原因失败

这些接口通常成功返回 None,失败抛异常,不会返回 Lua 风格的 nil + 错误信息。先 exists() 再操作也不能消除并发变化;真正操作的异常仍要处理。这里不提供对任意传入目录递归删除的示例。

以下可运行示例验证目录枚举、文本写入、空目录删除和文件删除,所有对象都在独立临时目录内:

import os
from pathlib import Path
from tempfile import TemporaryDirectory
def raise_walk_error(error):
raise error
with TemporaryDirectory(prefix="vitalogos-files-") as directory:
root = Path(directory)
nested = root / "data"
os.mkdir(nested)
sample = nested / "sample.txt"
with sample.open("x", encoding="utf-8", newline="\n") as stream:
assert stream.write("第一行\nsecond\n") == 11
assert sorted(os.listdir(root)) == ["data"]
rows = list(os.walk(root, onerror=raise_walk_error))
assert len(rows) == 2
assert rows[0][1] == ["data"]
assert rows[1][2] == ["sample.txt"]
with sample.open("r", encoding="utf-8") as stream:
assert stream.readline() == "第一行\n"
assert stream.readline() == "second\n"
assert stream.readline() == ""
os.remove(sample)
os.rmdir(nested)
assert os.listdir(root) == []
print("temporary file operations passed")

文件模式、字符和字节

跳转到“文件模式、字符和字节”
模式部分作用
r读取已有文件;默认基础模式
w写入;已有文件会被截断,不存在则创建
x仅新建;已存在时抛出 FileExistsError
a追加;不存在则创建,写入追加到末尾
+在所选基础模式上启用读写
t文本模式,默认;读写 str,有编码和换行转换
b二进制模式;读写 bytes,不传 encoding

例如 rb 读取字节,w+ 先截断再读写,r+ 不会先截断,ab+ 可读取并追加字节。追加模式中 seek(0) 可以改变读取位置,不能把后续写入改成覆盖文件开头。原稿的 U 模式在 Python 3.11 已移除;文本读取默认的通用换行处理由 newline 参数管理。内置 open() 文档

read(size) 的 size 在文本模式中以字符计,在二进制模式中以字节计;不传会尝试读取剩余全部内容,没有“超过两倍内存才有问题”的固定阈值。readline() 返回一行并通常保留换行符;末行没有换行时不会补上。文本 EOF 返回 "",二进制 EOF 返回 b""。readlines(hint) 的 hint 不是行数限制;逐行处理大文件可以直接 for line in stream。

with 负责在正常结束或异常路径关闭文件;块内无需额外 close()。缓冲减少细小系统调用的开销,但 flush() 不等于保证断电后数据落盘。文本写入的返回值是写入字符数,不能当作 UTF-8 字节数。JSON 读写另见 json,二进制记录布局见 struct。

保留 JPG 收集场景,拒绝覆盖已有目标

跳转到“保留 JPG 收集场景,拒绝覆盖已有目标”

原示例从数据集目录收集 .jpg 文件。下面只复制第一层普通文件,扩展名大小写不敏感;跳过链接和子目录,目标文件采用 xb 独占创建。复制的是文件内容,不承诺保留权限、时间戳或扩展属性。

from pathlib import Path
import shutil
from tempfile import TemporaryDirectory
def collect_jpegs(source, destination):
source = Path(source).resolve(strict=True)
destination = Path(destination).resolve()
if not source.is_dir():
raise NotADirectoryError(source)
if source == destination:
raise ValueError("源目录与目标目录必须不同")
destination.mkdir(parents=True, exist_ok=True)
copied = []
for item in sorted(source.iterdir(), key=lambda value: value.name):
if item.is_symlink() or not item.is_file() or item.suffix.lower() != ".jpg":
continue
target = destination / item.name
with item.open("rb") as incoming, target.open("xb") as outgoing:
shutil.copyfileobj(incoming, outgoing)
copied.append(target.name)
return copied
with TemporaryDirectory(prefix="vitalogos-jpegs-") as directory:
root = Path(directory)
source = root / "source"
source.mkdir()
(source / "one.jpg").write_bytes(b"fixture-one")
(source / "two.JPG").write_bytes(b"fixture-two")
(source / "ignore.png").write_bytes(b"fixture-ignore")
(source / "folder.jpg").mkdir()
target = root / "output"
assert collect_jpegs(source, target) == ["one.jpg", "two.JPG"]
assert (target / "one.jpg").read_bytes() == b"fixture-one"
try:
collect_jpegs(source, target)
except FileExistsError:
pass
else:
raise AssertionError("已有文件不能被覆盖")
print("JPG filename selection and non-overwrite behavior passed")

测试字节只用于验证文件复制,不是真实 JPEG,也不证明图像可解码。此函数面向本地批处理;它不是事务,途中失败时先前复制项及当前不完整目标可能保留。遇到并发修改、需要回滚、链接安全边界或大量数据恢复时,应另行设计暂存与提交方案,而不是在异常时递归删除整个目标目录。

数字文件名可以用 f"{number:04d}" 得到最小宽度 4 的十进制形式,例如 [0, 10, 20, 120, 121, 1121] 对应 0000、0010、0020、0120、0121、1121;宽度不会截断更长的数,也不自动拒绝负数。f"{3.1415926:.2f}"、f"{'hello':<10}" 分别控制小数位和对齐;日期格式、旧式 % 格式和 IPython %matplotlib 魔法命令属于不同语法。原例中不要用 str 作为变量名遮蔽内置类型。

os.environ 是当前进程环境的字符串映射,通常在导入 os 时建立。直接通过它赋值会更新当前进程环境,后续子进程通常可以继承;外部进程不能借此反向修改父进程的环境。原生代码或直接 os.putenv() 造成的变化也未必同步到这个 Python 映射。取可选值可用 get(),下标取不存在的键会抛出 KeyError。Python os 文档

原文的 MASTER_ADDR、MASTER_PORT 只是分布式训练程序读取的约定变量,设值本身不会创建服务或完成初始化。HOMEPATH 也不是跨平台的完整用户主目录接口;通常用 Path.home(),并仍需处理目标环境没有合适配置的情况。

os.system() 通过命令解释器执行字符串,返回值含义有平台差异。参数来自程序时,使用 subprocess.run() 的列表形式可避免把它们当作额外 shell 语法。下面只启动同一 Python 解释器,向子进程传入一个演示变量,不修改父进程环境:

import os
import subprocess
import sys
key = "VITALOGOS_OS_DEMO"
before = os.environ.get(key)
child_environment = os.environ.copy()
child_environment[key] = "example-value"
assert sys.executable
completed = subprocess.run(
[sys.executable, "-c", "import os; print(os.environ['VITALOGOS_OS_DEMO'])"],
env=child_environment,
text=True,
capture_output=True,
check=True,
timeout=10,
)
assert completed.stdout.strip() == "example-value"
assert os.environ.get(key) == before
print("child environment passed; parent unchanged")

check=True 会在非零退出码时抛出 CalledProcessError,timeout 为等待设置上限。若要运行原来的 run.py --type rendering,参数应作为独立元素传入,且脚本必须真实存在、来源明确;本页没有启动渲染工程。解释器诊断见 sys。