# mypy: 允许未类型化定义
导入
内置函数
导入
导入库
导入 importlib.machinery
导入
检查
导入
输入/输出
导入
linecache 模块
导入
操作系统
导入
系统
导入
类型
from collections.abc 导入
迭代器
from contextlib 导入 contextmanager
from 打字
导入
任何,
可调用,
角色,
可选,
类型检查,
联合
from 弱引用
导入
弱值字典
导入
火炬
from torch 序列化
导入
_获取恢复位置,
可能解码 ASCII
from torch.types 导入 FileLike
from .目录读取器
导入
目录读取器
from ._importlib 导入 (
_calc___package__,
_normalize_line_endings,
_normalize_path,
_resolve_name,
稳定性检查,
)
from _重命名
导入
反重命名,
包重命名器
from ._package_unpickler 导入
包反序列化器
from .文件结构表示
导入
从文件列表创建目录,
目录
from .导入器
导入
导入器
如果
类型检查:
from .glob_group 导入
全局模式
全部 = [
"包导入器"]
这是一个列表，其中包含即使没有显式允许也隐式允许的导入项
已被标记为 extern。这是为了解决 Torch 隐式地
依赖 numpy 包，无法追踪该包。
# https://github.com/pytorch/MultiPy/issues/46
隐式导入允许列表:
迭代器[
字符串] = [
numpy,
numpy.core,
numpy.core._multiarray_umath,
# FX GraphModule 可能依赖于内置模块，用户通常
# 不外部引入内置模块。这里我们默认导入它。
内建函数,
]
# 兼容性名称映射，以方便外部模块的升级。
# 主要动机是启用许多模块的 Numpy 升级。
依赖于。Numpy 的最新版本移除了`numpy.str`和
`numpy.bool` 导致许多模块反序列化失败。
外部导入兼容名称映射:
字典[
字符串,
字典[
字符串,
任何]] = {
numpy: {
"字符串":
字符串,
"布尔":
布尔,
},
}
[文档]
类
包导入器(
导入器):
导入器允许您加载由 :class:`PackageExporter` 编写的包代码。
代码以密封方式加载，使用包中的文件
而不是正常的 Python 导入系统。这允许
将 PyTorch 模型代码和数据打包，以便可以运行
在服务器上或在未来的迁移学习中使用。
包导入器确保模块中的代码只能从包内部加载，
除了在导出时明确列为外部的模块之外。
压缩包中的文件 `extern_modules` 列出了包外部依赖的所有模块。
这防止了“隐式”依赖，因为包在本地运行是因为它导入了
本地安装的包，但然后将包复制到另一台机器上时失败。
"文档"
“该包中已加载的模块的字典，相当于 `sys.modules`，但仅限于此导入器。
本地化。
"文档"
模块:
字典[
字符串,
类型.
模块类型]
[文档]
定义 __init__(
我,
文件或缓冲区。:
联盟[
文件类,
火把._C.
PyTorch 文件读取器
]
模块允许:
可调用[[
字符串
]
布尔] = lambda
模块名称:
是,
):
打开 ``file_or_buffer`` 以导入。这会检查导入的包是否只允许由 ``module_allowed``
允许的 ``file_or_buffer``
参数：
file_or_buffer：一个文件对象（必须实现 :meth:`read`，:meth:`readline`，:meth:`tell` 和 :meth:`seek`）
字符串，或包含文件名的 ``os.PathLike`` 对象。
module_allowed (Callable[[str], bool], 可选): 用于确定外部提供的模块是否允许的方法
应该被允许。可以用来确保加载的包不依赖于服务器不支持模块。默认为允许任何内容。
默认为允许任何内容。
抛出异常：
导入错误：如果该包将使用不允许的模块。
"文档"
火把._C._log_api_usage_once(
torch.package.PackageImporter)
我.
zip 读取器:
任何
如果 isinstance(
文件或缓冲区,
火把._C.
PyTorch 文件读取器):
我.
文件名 =
我.
压缩文件读取器 =
文件或缓冲区
elif isinstance(文件或缓冲区, (os.PathLike,
字符串)):
我.
文件名 = os.
文件系统路径(
文件或缓冲区)
如果
不 os.
路径.isdir(
我.
文件名):
我.
压缩文件读取器 =
火把._C.
PyTorch 文件读取器(
我.
文件名)
else:
我.
压缩文件读取器 =
目录读取器(
我.
文件名)
else:
我.
文件名 =
"<二进制>"
我.
压缩文件读取器 =
火把._C.
PyTorch 文件读取器(
文件或缓冲区)
火把._C.
记录 API 使用元数据(
"torch.package.PackageImporter.metadata",
{
序列化 ID:
我.
zip 读取器.
序列化 ID(),
"文件名":
我.
文件名,
},
)
我.
根 =
包节点(
无)
我.
模块 = {}
我.
外部模块 =
我.
读取外部()
for 外部模块
在
我.
外部模块:
如果
不
允许的模块(
外部模块):
raise 导入错误(
f"包 '{
文件或缓冲区}
需要外部模块{
外部模块}' "
f但该模块已被禁止
)
我.
添加外部(
外部模块)
for 文件名
在
我.
zip 读取器.
获取所有记录():
我.
添加文件(fname)
我.
补丁内置函数 = builtins.
字典.
复制()
我.
修补过的内置函数[
__import__] =
我.__import__
允许打包模块引用其包导入器
我.
模块[
torch_package_importer] = self
# 类型：忽略[赋值]
我._mangler =
包裹搅拌器()
用于减少反序列化
我.
存储上下文:
任何 =
无
我.
最后的映射位置 =
无
用于 torch.serialization._load
我.
反挑选器 = lambda *
参数, **kwargs:
包卸载器(
我, *
参数, **kwargs)
[文档] def 导入模块(self, 名称: str, 包=None):
"""从包中加载模块（如果尚未加载），然后返回模块。模块是本地加载的
the module. Modules are loaded locally
导入器中，并将出现在 `self.modules` 而不是 `sys.modules` 中。
Args:
name (str): 要加载的模块的完全限定名称。
package ([type], optional): 未使用，但存在以匹配 importlib.import_module 的签名。默认为 `None`。
返回值：
types.ModuleType：已（可能已）加载的模块。
"""
# 我们应该始终能够支持从这个包中导入模块。
这是为了支持类似：
导入器加载 pickle 对象...
导入模块(obj.__module__) <- 这行字符串将被修改
#
# _mangler.demangle 不会解构任何模块名称
# 由不同的 PackageImporter 实例产生。
# name = self._mangler.demangle(name)
# return self._gcd_import(name)
[文档] def load_binary(self, package: str, resource: str) -> bytes:
"""加载原始字节。
Args:
package (str): 模块包的名称（例如 ``"my_package.my_subpackage"``）。
资源（str）：资源的唯一名称。
返回值：
bytes：加载的数据。
"""
path = self._zipfile_path(package, resource)
return self.zip_reader.get_record(path)
[文档] def load_text(
self,
package: 字符串,
resource: 字符串,
encoding: 字符串 = "utf-8",
errors: 字符串 = "strict",
def __init__(self):
"""加载一个字符串。
参数:
package (str): 模块包名（例如：``"my_package.my_subpackage"``）。
资源（str）：资源的唯一名称。
编码（str，可选）：传递给 ``decode``。默认为 ``'utf-8'``。
错误处理（str，可选）：传递给 ``decode``。默认为 ``'strict'``。
返回值：
加载的文本
"""
data = self.load_binary(package, resource)
返回 data.decode(encoding, errors)
[文档]
定义
加载 pickle(
我,
包:
字符串,
资源:
字符串,
地图位置=
无) ->
任何:
从包中反序列化资源，加载构建对象所需的任何模块
使用 :meth:`import_module`。
参数：
包 (str)：模块包的名称（例如，``"my_package.my_subpackage"``）。
资源 (str)：资源的唯一名称。
map_location：传递给 `torch.load` 以确定张量如何映射到设备。默认为 ``None``。
返回：
任何：未反序列化的对象。
"文档"
Pickle 文件 =
我._zipfile_path(
包,
资源)
恢复位置 =
_获取恢复位置(
地图位置)
已加载存储 = {}
loaded_reduces = {}
storage_context = 火把._C.
反序列化存储上下文()
定义
加载张量(
数据类型,
尺寸,
键,
位置,
恢复位置):
名称 = f"{
键}
.存储
如果
存储上下文.
具有存储(
名称):
存储 =
存储上下文.
获取存储(
名称,
数据类型)._typed_storage()
else:
张量 =
我.
zip 读取器.
从记录中获取存储(
.data/ +
名称,
尺寸, dtype
)
如果 isinstance(
我.
zip 读取器,
火把._C.
PyTorch 文件读取器):
存储上下文.
添加存储(
名称,
张量)
存储 =
张量._typed_storage()
已加载存储[
键] =
恢复位置(
存储,
位置)
定义
持久化加载(
已保存的 ID):
断言 isinstance(
已保存的 ID,
元组)
类型名 =
可能解码 ASCII(
已保存的 ID[0])
数据 =
已保存的 ID[1
]
如果
类型名 ==
存储:
存储类型,
键,
位置,
大小 =
数据
如果
存储类型
是
火把.
未类型化存储:
dtype = 火把.uint8
else:
dtype = 存储类型.dtype
如果 key
不
在
已加载存储:
加载张量(
数据类型,
尺寸,
键,
可能解码 ASCII(
位置),
恢复位置,
)
存储 =
已加载存储[
键]
一旦我们决定中断序列化功能，我们就可以
# 停止使用 TypedStorage 进行包装
返回
火把.
存储.
类型化存储(
包装存储=
存储.
未类型化存储,
数据类型=
数据类型,
内部=
真实
)
elif 类型名 ==
"减少包":
# 修复 BC 破坏性变更，此加载路径上的对象
# 将错误地多次加载
如果
长度(
数据) == 2:
函数, args =
数据
返回
函数(
我, *
参数)
减少 ID,
函数, args =
数据
如果
减少 ID
不
在
已加载减少:
已加载减少[
减少 ID] =
函数(
我, *
参数)
返回
已加载的减少[
减少 ID]
else:
f"未知持久加载的 typename，期望为'storage'或'reduce_package'，但得到' "{
类型名}
# 加载数据（这可能会进一步使用 `persistent_load` 来加载数据）
数据文件 =
输入/输出.BytesIO(
我.
zip 读取器.
获取记录(
Pickle 文件))
反序列化器 =
我.
反序列化器(
数据文件)
反序列化器.
持久加载 =
持久加载
# 类型：忽略[赋值]
@contextmanager
定义
设置反序列化上下文():
允许 reduce_package 访问反序列化上下文
我.
存储上下文 =
存储上下文
我.
上次地图位置 =
地图位置
尝试:
产生
最后:
我.
存储上下文 =
无
我.
上次地图位置 =
无
与
设置反序列化上下文():
结果 =
反序列化器.
加载()
# TODO from zdevito:
# 这个有状态且奇怪的函数将在我们的努力中需要被移除
将格式统一。如果多个 python 同时运行，则存在竞争条件。
线程尝试读取独立文件
火把.
_工具.
验证已加载的稀疏张量()
返回
结果
[文档] def id(self):
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
返回 torch.package 用于区分：class:`PackageImporter`实例的内部标识符。
看起来像::
"""
return self._mangler.parent_name()
[文档] def file_structure(
self, *, include: "GlobPattern" = "**", exclude: "GlobPattern" = ()
) -> 目录:
返回包的 zip 文件结构的表示。
参数：
include（Union[List[str], str]）：一个可选字符串，例如 ``"my_package.my_subpackage"``, 或者可选字符串列表
用于指定要包含在 zip 文件表示中的文件名。这也可以是
一个 glob 样式的模式，如 :meth:`PackageExporter.mock` 中所述
exclude（Union[List[str], str]）：一个可选的模式，用于排除与该模式匹配的文件名。
返回：
class:`Directory`
"""
返回从文件列表创建目录的函数(
self.filename, self.zip_reader.get_all_records(), include, exclude
)
[文档] def python_version(self):
"""返回创建此包所使用的 Python 版本。
注意：此函数为实验性，且不具有向前兼容性。计划稍后将其移动到锁定文件中。
注意：此函数为实验性，且不具有向前兼容性。计划稍后将其移动到锁定文件中。
返回值：
class:`Optional[str]` 一个 Python 版本，例如 3.8.9，或者如果没有与该软件包一起存储版本则为 None
"""
python_version_path = ".data/python_version"
return (
self.zip_reader.get_record(python_version_path).decode("utf-8").strip()
if self.zip_reader.has_record(python_version_path)
else None
)
定义
读取外部(
我):
返回 (
我.
zip 读取器.
获取记录(
".data/外部模块")
.解码(
utf-8)
.按行分割(
保留结束符=
错误)
)
定义
创建模块(
我,
名称:
字符串,
文件名:
可选[
字符串
]
是否为包:
布尔,
父节点:
字符串
):
混乱文件名 =
我.
_搅拌器.
搅拌(
文件名)
如果
文件名
否则
无
规范 = importlib.
机器.
模块规范(
名称,
我, # type: ignore[arg-type]
原始=
"<包导入器>",
是否为包=
是否为包,
)
模块 = importlib.
工具.
从规范导入模块(
规格)
我.
模块[
名称] =
模块
模块.__name__ =
我.
_变形器.
混乱(
名称)
命名空间 =
模块.
字典
ns["__spec__"] = 规范
ns["__加载器__"] = self
ns["__文件__"] =
混乱文件名
ns["__缓存的__"] =
无
ns["__内置__"] =
我.
补丁内置函数
ns["__torch 包__"] =
真实
将此模块添加到我们的私有全局注册表中。由于混淆，它应该是唯一的。
断言
模块.__name__
不
在
_已导入的包模块
_已导入的包模块[
模块.__name__] =
模块
预先在父模块中安装，以防止从父模块尝试导入
访问 sys.modules
我.
在父节点上安装(
父节点,
名称,
模块)
如果
文件名
是
不
无:
断言
乱序文件名
是
不
无
预先在 `linecache` 中安装源代码，以便 `inspect`、堆栈跟踪等正常工作。
处理后的文件名
断言
文件名
不
在 linecache.
缓存
# 类型: 忽略[attr-defined]
linecache.懒加载缓存(
_编译源代码, ns)
代码 =
我.
_编译源代码(
文件名,
混乱文件名)
执行(
代码, ns)
返回
模块
定义
_加载模块(
我,
名称:
字符串,
父节点:
字符串):
cur: _路径节点 =
我.
根
for 原子
在
名称.
分割(
“。”):
如果
不 isinstance(cur,
_包节点)
或者
原子
不
在 cur.
儿童:
如果
名称
在
隐式导入允许列表:
模块 =
我.
模块[
名称] = importlib.
导入模块(
名称)
返回
模块
raise 模块未找到错误(
f没有名为 "{
名称}
" 的模块在自包含归档中 "{
我.
文件名}
''
f" 并且该模块也不在允许的外部模块列表中：{
我.
外部模块}",
名称=
名称,
)
cur = cur.儿童[
原子]
如果 isinstance(cur,
_外部节点):
模块 =
我.
模块[
名称] = importlib.
导入模块(
名称)
如果
兼容映射 :=
外部导入兼容名称映射.
获取(
名称):
for 旧名称,
新名称
在
兼容映射.
项目():
模块.
字典.setdefault(
旧名称,
新名称)
返回
模块
返回
我.
_创建模块(
名称, cur.
源文件, isinstance(cur,
_包节点),
父节点)
# 类型: 忽略[attr-defined]
定义
_编译源代码(
我,
完整路径:
字符串,
混乱的文件名:
字符串):
源 =
我.
zip 读取器.
获取记录(
完整路径)
源 =
_标准化行结束符(
源)
返回
编译(
源,
混乱的文件名,
执行,
不继承=
是)
# 注意：命名为 `get_source` 以便 linecache 可以找到源
# 当这是模块的 __loader__ 时。
定义
获取源(
我,
模块名称) ->
字符串:
# linecache 使用 `module.__name__` 作为参数调用 `get_source`，因此我们必须在这里解混淆。
模块 =
我.
导入模块(demangle(
模块名称))
返回
我.
zip 读取器.
获取记录(demangle(
模块.__file__)).
解码(
utf-8)
# note: 命名为 `get_resource_reader` 以便 importlib.resources 可以找到它。
# 这通常被认为是一个内部方法。
定义
获取资源读取器(
我,
全名):
尝试:
包 =
我.
_获取包(
全名)
除了
导入错误:
返回
无
如果
软件包.__loader__
是
不
我:
返回
无
返回
_包资源读取器(
我,
全名)
定义
在父级上安装(
我,
父节点:
字符串,
名称:
字符串,
模块:
类型.
模块类型):
如果
不
父节点:
返回
# 将模块作为属性设置在其父级上。
parent_module = 我.
模块[
父节点]
如果
父模块.__loader__
是
我:
setattr(父模块,
名称.
分割(
“。”
)2
]
模块)
# note: 从 cpython 的导入代码复制而来，将创建模块的调用替换为_make_module
定义
查找并加载(
我,
名称):
父级 =
名称.
分割(
“。”
)0]
模块名称无父级 =
名称.
分割(
“。”
)-1]
如果
父节点:
如果
父级
不
在
我.
模块:
我._gcd_import(
父节点)
# 疯狂的副作用！
如果
名称
在
我.
模块:
返回
我.
模块[
名称]
parent_module = 我.
模块[
父节点]
尝试:
父模块.__path__
# 类型: 忽略[attr-defined]
除了
属性错误:
# 当我们尝试导入仅包含 pybinded 文件的包时，
# 父目录不一定被定义为 Python 包，
# 因此我们在调用错误之前先搜索该包是否实际存在。
如果 isinstance(
父模块.__loader__,
importlib.machinery.扩展文件加载器,
):
如果
名称
不
在
我.
外部模块:
msg = (
_错误消息
+ ;{!r}
是一个未导出的 C 扩展模块。C 扩展模块\
需要通过 PackageExporter 导出才能使用，因为我们不支持内部化它们。}"
).格式(
名称,
名称)
raise 模块未找到错误(
信息,
名称=
名称) from
无
如果
不 isinstance(
父模块.
字典.
获取(module_name_no_parent),
类型.
模块类型,
):
msg = (
_ERR_MSG
+ ;
原文
是一个不包含的 C 扩展包
原文
。
).格式(
名称,
父节点,
名称)
raise 模块未找到错误(
信息,
名称=
名称) from
无
else:
msg = (_ERR_MSG + ;
源代码
不是一个包).
格式(
名称,
父节点)
raise 模块未找到错误(
信息,
名称=
名称) from
无
模块 =
我.
_加载模块(
名称,
父节点)
我.
_在父级上安装(
父节点,
名称,
模块)
返回
模块
# note: 从 cpython 的导入代码复制
定义 _find_and_load(
我,
名称):
模块 =
我.
模块.
获取(
名称, _NEEDS_LOADING)
如果
模块
是
需要加载:
返回
我.
查找并加载(
名称)
如果
模块
是
无:
消息 = f
"导入"{
名称}
停止；sys.modules 中无 None
raise 模块未找到错误(
消息,
名称=
名称)
处理 https://github.com/pytorch/pytorch/issues/57490，其中 std 的
通过对 sys.modules 的黑客攻击创建假子模块不是导入
友好的
如果
名称 ==
os:
我.
模块[
os.path] =
角色(
任何,
模块).
路径
elif 名称 ==
打字:
如果
系统模块.version_info < (3, 13):
我.
模块[
typing.io] =
角色(
任何,
模块).
输入/输出
我.
模块[
typing.re] =
角色(
任何,
模块).
正则表达式
返回
模块
定义 _gcd_import(
我,
名称,
软件包=
无,
级别=0):
根据名称、调用所在的包以及级别调整导入并返回模块
此函数代表了 import_module 和__import__之间的功能最大公约数。这包括设置__package__如果
该函数表示 import_module 和__import__之间功能的最小公倍数。这包括如果需要设置__package__
则设置__package__。
加载器没有。
"文档"
_健康检查(
名称,
软件包,
级别)
如果
等级 > 0:
名称 =
_解析名称(
名称,
软件包,
级别)
返回
我.
_查找和加载(
名称)
# 来自 cpython 的导入代码
定义
_从列表处理(
我,
模块, fromlist, *,
递归=
错误):
确定应该返回什么 __import__。
import_ 参数是一个可调用的对象，它接受要导入的模块的名称作为参数。
导入。需要将函数与假设 importlib 的导入实现解耦。
如果希望导入实现独立。
"文档"
模块名称 = demangle(
模块.__name__)
# fromlist...的炼狱...
尝试从 fromlist 导入内容
如果
有属性(
模块,
"__路径__"):
for x 在 fromlist:
如果
不 isinstance(x,
字符串):
如果
递归:
其中 =
模块名称 +
"__all__"
else:
其中 = "``from list''"
raise 类型错误(
f"项目在"{
哪里}
必须为 str 类型，而不是{
类型(x).__name__}"
)
elif x == "*":
如果
不
递归
并且
有属性(
模块, "__all__"):
我.
_处理列表(
模块,
模块.__all__,
递归=
是)
elif 不
有属性(
模块, x):
发件人 = f"{
模块名称}.{x}"
尝试:
我.
_gcd 导入(
发件人)
除了
模块未找到错误
是
异常:
# 向后兼容性规定我们忽略由 fromlist 触发的失败导入
# 不存在的模块触发的 fromlist 导入
#
如果 (
异常.
名称 == from_name
并且
我.
模块.
获取(from_name,
需要加载)
是
不
无
):
continue
raise
返回
模块
定义
__导入__(
我,
名称,
全局变量=
无, locals=
无, fromlist=(),
级别=0):
如果
等级 == 0:
模块 =
我.
最大公约数导入(
名称)
else:
全局变量 =
全局变量
如果
全局变量
是
不
无
否则 {}
包 =
_calc___包__(
全局_)
模块 =
我.
_gcd 导入(
名称,
包,
级别)
如果
不 fromlist:
返回 'name' 中的第一个点之前的内容。这很复杂，因为 'name' 可能是相对的。
那个 'name' 可能是相对的。
如果
等级 == 0:
返回
我._gcd_import(
名称.partition(
“。”
)0])
elif 不
名称:
返回
模块
else:
确定在哪里将模块的名称切片到第一个点。
# in '名称'.
截断 =
长度(
名称) -
长度(
名称.partition(
“。”
)0])
# 切片结束需要为正数以减轻需要特殊处理的情况
# 当 ``'.' 不在名称'' 中时。
模块名称 = demangle(
模块.__name__)
返回
我.
模块[
模块名称
[
长度(
模块名称) - cut_off]]
else:
返回
我._handle_fromlist(
模块, fromlist)
定义 _get_package(
我,
包):
"获取包名或模块对象并返回模块。"
如果是名称，则导入模块。如果传入或导入的对象不是包，则抛出异常。
如果传入或导入的对象不是包，则抛出异常。
"文档"
如果
有属性(
包, "__spec__"):
如果
包.__spec__.
子模块搜索位置
是
无:
raise 类型错误(f"{
包.__spec__.
名称!r}
不是包)
否则:
返回
包
else:
模块 =
我.
导入模块(
包)
如果
模块.__spec__.
模块搜索位置
是
无:
raise 类型错误(f"{
软件包!r}
不是一个包")
else:
返回
模块
定义 _zipfile_path(
我,
软件包,
资源=
无):
包 =
我.
_获取包(
软件包)
断言
软件包.
__加载器__
是 self
名称 = demangle(
软件包.__name__)
如果
资源
是
不
无:
资源 = _normalize_path(
资源)
返回 f"{
名称.
替换(
点,
根目录)}/{
资源}"
else:
返回 f"{
名称.
替换(
点,
根目录)}"
定义
获取或创建包(
我,
原子:
列表[
字符串]
) -> "联合[_PackageNode, _ExternNode]":
cur = 我.
根
for i, 原子
在
列举(
原子):
节点 = cur.
儿童.
获取(
原子,
无)
如果
节点
是
无:
节点 = cur.
儿童[
原子] =
_包节点(
无)
如果 isinstance(node,
_外部节点):
返回
节点
如果 isinstance(node,
_模块节点):
名称 =
“。”.
连接(
原子
[i])
raise 导入错误(
f模块结构不一致。模块{
名称}
不是一个包，但有子模块
)
断言 isinstance(node,
_包节点)
cur = 节点
返回 cur
定义
_添加文件(
我,
文件名:
字符串):
将给定的文件组装成 Python 模块。将忽略.data 目录中的文件。
参数：
filename (str): 要添加到包归档中的文件名
"文档"
*前缀,
最后一个 =
文件名.
分割(
根目录)
如果
长度(
前缀) > 1
并且
前缀[0] == ".data":
返回
包 =
我._get_or_create_package(
前缀)
如果 isinstance(
包, _ExternNode):
raise 导入错误(
f"模块结构不一致。包中包含一个模块文件，"{
文件名}"
f"它是标记为外部的模块的子包。"
)
如果
最后一个 == "__init__.py":
包.
源文件 =
文件名
elif 最后.
以...结尾(".py"):
包名 =
最后
[ -
长度(".py")]
软件包.
儿童[
包名] =
模块节点(
文件名)
定义
添加外部(
我,
外部名称:
字符串):
*前缀,
最后一个 =
外部名称.
分割(
“。”)
包 =
我._get_or_create_package(
前缀)
如果 isinstance(
软件包, _ExternNode):
返回
# 简短的 extern 涵盖了此 extern 情况
软件包.
儿童[
最后] =
外部节点()
需要加载 =
对象()
错误信息前缀 =
没有名为 " 的模块
错误消息 =
错误消息前缀 + "
格式化字符串"
类
路径节点:
通过
类
_包节点(
_路径节点):
定义 __init__(
我,
源文件:
可选[
字符串
)]
我.
源文件 =
源文件
我.
儿童:
字典[
字符串,
_路径节点] = {}
类
_模块节点(
_路径节点):
__slots__ = ["源文件"]
定义 __init__(
我,
源文件:
字符串):
我.
源文件 =
源文件
类
外部节点(
路径节点):
通过
# 所有已通过包导入的模块的私有全局注册表。
_package_imported_modules: 弱值字典 =
弱值字典()
默认情况下，`inspect` 只在 `sys.modules` 中查找类的源文件。
将它修改为检查我们的私有包导入模块注册表。
_orig_getfile = 检查.getfile
定义 _patched_getfile(
对象):
如果
检查.
是否为类(
对象):
如果
对象.__module__
在 _package_imported_modules:
返回 _package_imported_modules[
对象.
__模块__].__file__
返回 _orig_getfile(
对象)
检查.
获取文件 =
_修改版_getfile
类 _PackageResourceReader:
用于支持 PackageImporter.get_resource_reader()的私有类。
符合 importlib.abc.ResourceReader 接口。允许访问
包导入器的内部结构。
"文档"
定义 __init__(
我,
导入器,
全名):
我.
导入器 =
导入器
我.
全名 =
全名
定义
打开资源(
我,
资源):
from 输入/输出
导入 BytesIO
返回 BytesIO(
我.
导入器.
加载二进制(
我.
全名,
资源))
定义
资源路径(
我,
资源):
# 资源路径的合约是它要么返回一个具体的
# 文件系统路径，要么引发 FileNotFoundError 异常。
如果 isinstance(
我.
导入器.
zip 读取器,
目录读取器
) 并且
我.
导入器.
zip 读取器.
有记录(
os.路径.
连接(
我.
全名,
资源)
):
返回 os.
路径.
连接(
我.
导入器.
zip 读取器.
目录,
我.
全名,
资源
)
raise 文件未找到错误
定义
资源(
我,
名称):
路径 =
我.
导入器.
压缩文件路径(
我.
全名,
名称)
返回
我.
导入器.
zip 读取器.
有记录(
路径)
定义
内容(
我):
from pathlib 导入 Path
文件名 =
我.
全名.
替换(
“。”,
根目录)
全名路径 =
路径(
我.
导入器.
_zip 文件路径(
我.
全名))
文件 =
我.
导入器.
zip 读取器.
获取所有记录()
子目录已见 =
设置()
for 文件名
在
文件:
尝试:
相对 =
路径(
文件名).
相对于(
完整路径名)
除了 ValueError:
continue
# 如果文件的路径（相对于 zip 文件的顶部，即相对于
# 命名空间），相对于在提供资源时指定的包
# 读者被创建，有父级，那么它就是一个子目录中的名称，因此我们跳过它。
# 子目录，因此我们跳过它。
父级名称 =
相对.
父节点.
名称
如果
长度(
父亲姓名) == 0:
产生
亲戚.
名称
elif 父级名称
不
在
已见子目录:
已见子目录.
添加(
父级名称)
产生
父名称
