`updatedb`¶

更新数据库

这个模块提供了一些和更新数据库有关的函数

class p115client.tool.updatedb.P115QueryDB(con, /)[source]¶

Bases: object

封装了一些常用的数据库查询方法，针对 updatedb 产生的 SQLite 数据库

Note

默认情况下，只有 “id”、”parent_id”、”updated_at” 有索引，所以请自行添加其它需要的索引

get_ancestors(id: int = 0, /) → list[dict][source]¶

获取某个文件或目录的祖先节点信息，包括 “id”、”parent_id” 和 “name” 字段

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id

Returns:

当前节点的祖先节点列表，从根目录开始（id 为 0）直到当前节点

get_attr(id: int = 0, /) → dict[source]¶

获取某个文件或目录的信息

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id

Returns:

当前节点的信息字典

get_id(pickcode: str = '', sha1: str = '', path: str | Sequence[str] = '', parent_id: int = 0, is_alive: bool = True) → int[source]¶

查询匹配某个字段的文件或目录的 id

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
pickcode – 当前节点的提取码，优先级高于 sha1
sha1 – 当前节点的 sha1 校验哈希值，优先级高于 path
path – 当前节点的路径
parent_id – 仅用于 path 参数，用来限定搜索的顶层路径

Returns:

当前节点的 id

get_parent_id(id: int = 0, /, default: None | int = None) → int[source]¶

获取某个 id 对应的父 id

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id

Returns:

当前节点的父 id

get_path(id: int = 0, /) → str[source]¶

获取某个文件或目录的路径

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id

Returns:

当前节点的路径

get_patht(id: int = 0, /) → list[str][source]¶

获取某个文件或目录的路径节点元组

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id

Returns:

当前节点的路径节点元组

get_pickcode(id: int = -1, sha1: str = '', path: str | Sequence[str] = '', parent_id: int = 0, is_alive: bool = True) → str[source]¶

查询匹配某个字段的文件或目录的提取码

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id，优先级高于 sha1
sha1 – 当前节点的 sha1 校验哈希值，优先级高于 path
path – 当前节点的路径
parent_id – 仅用于 path 参数，用来限定搜索的顶层路径

Returns:

当前节点的提取码

get_sha1(id: int = -1, pickcode: str = '', path: str | Sequence[str] = '', parent_id: int = 0, is_alive: bool = True) → str[source]¶

查询匹配某个字段的文件的 sha1

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id，优先级高于 pickcode
pickcode – 当前节点的提取码，优先级高于 path
path – 当前节点的路径
parent_id – 仅用于 path 参数，用来限定搜索的顶层路径

Returns:

当前节点的 sha1 校验哈希值

has_id(id: int, /, is_alive: bool = True) → int[source]¶

是否存在某个 id

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
id – 当前节点的 id
is_alive – 是否存活

Returns:

如果是 1，则是 True；如果是 0，则是 False

iter_children(parent_id: int = 0, /, fields: Collection[str] = (), ensure_file: None | bool = None) → Iterator[dict][source]¶

Caution

当 fields 为空时，获取全部字段

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
parent_id – 父目录的 id
fields – 需要获取的字段
ensure_file –
是否仅输出文件
- 如果为 True，仅输出文件
- 如果为 False，仅输出目录
- 如果为 None，全部输出

Returns:

迭代器，产生一组信息的字典，大概包含如下字段（具体由你的数据库 data 表确定）：

(
    "id", "parent_id", "name", "sha1", "size", "pickcode",
    "is_dir", "is_alive", "extra", "created_at", "updated_at",
)

iter_count_dir(parent_id: int = 0, /) → Iterator[dict][source]¶

迭代获取所有指定 id 下所有目录节点（包括自己）直属的文件数和目录数

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
parent_id – 顶层目录的 id

Returns:

迭代器，返回字典

{
    "id": int,
    "parent_id": int,
    "dir_count": int,
    "file_count": int,
}

iter_count_tree(parent_id: int = 0, /) → Iterator[dict][source]¶

迭代获取所有指定 id 下所有目录节点（包括自己）直属的文件数和目录数，以及子树下的文件数合计和目录数合计

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
parent_id – 顶层目录的 id

Returns:

迭代器，返回字典

{
    "id": int,
    "parent_id": int,
    "dir_count": int,
    "file_count": int,
    "tree_dir_count": int,
    "tree_file_count": int,
}

iter_dangling_ids() → Iterator[int][source]¶

罗列所有悬空的文件或目录的 id

Note

悬空的 id，即祖先节点中，存在一个节点，它的 parent_id 是悬空的

Parameters:: self – P115QueryDB 实例或者数据库连接或游标
Returns:: 迭代器，一组目录的 id

iter_dangling_parent_ids() → Iterator[int][source]¶

罗列所有悬空的 parent_id

Note

悬空的 parent_id，即所有的 parent_id 中，，不为 0 且不在 data 表中的部分

Parameters:: self – P115QueryDB 实例或者数据库连接或游标
Returns:: 迭代器，一组目录的 id

iter_descendants(parent_id: int = 0, /, min_depth: int = 1, max_depth: int = -1, *, fields: str, escape: None | bool | Callable[[str], str] = True, ensure_file: None | bool = None, topdown: None | bool = True) → Iterator[Any][source]¶

iter_descendants(parent_id: int = 0, /, min_depth: int = 1, max_depth: int = -1, *, fields: Collection[str] | bool = True, escape: None | bool | Callable[[str], str] = True, ensure_file: None | bool = None, topdown: None | bool = True) → Iterator[dict]

获取某个目录之下的所有节点信息

Caution

当 fields 为空时，获取全部字段

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
parent_id – 顶层目录的 id
min_depth – 最小深度
max_depth – 最大深度。如果小于 0，则无限深度
fields –
需要获取的字段
- 如果为 str，直接获取这个字段的值（不返回字典）
- 如果为 True，获取所有字段，且包括（”ancestors”, “path”, “relpath”, “depth”），返回字典
- 如果为 False，获取所有字段，但除了（”ancestors”, “path”, “relpath”, “depth”），返回字典
- 否则，获取所指定的这组字段，返回字典
escape –
对文件名进行转义
- 如果为 None，则不处理；否则，这个函数用来对文件名中某些符号进行转义，例如 “/” 等
- 如果为 True，则使用 posixpatht.escape，会对文件名中 “/”，或单独出现的 “.” 和 “..” 用 “" 进行转义
- 如果为 False，则使用 posix_escape_name 函数对名字进行转义，会把文件名中的 “/” 转换为 “|”
- 如果为 Callable，则用你所提供的调用，以或者转义后的名字
ensure_file –
是否仅输出文件
- 如果为 True，仅输出文件
- 如果为 False，仅输出目录
- 如果为 None，全部输出
topdown –
是否自顶向下深度优先遍历
- 如果为 True，则自顶向下深度优先遍历
- 如果为 False，则自底向上深度优先遍历
- 如果为 None，则自顶向下宽度优先遍历

Returns:

迭代器，产生一组信息的字典，大概包含如下字段：

(
    # NOTE: 这些具体由你的数据库 ``data`` 表确定
    "id", "parent_id", "name", "sha1", "size", "pickcode",
    "is_dir", "is_alive", "extra", "created_at", "updated_at",
    # NOTE: 这些是另外附加的字段
    "ancestors", "path", "relpath", "depth",
)

iter_descendants_bfs(parent_id: int = 0, /, *, fields: str, escape: None | bool | Callable[[str], str] = True) → Iterator[Any][source]¶

iter_descendants_bfs(parent_id: int = 0, /, *, fields: Collection[str] | bool = True, escape: None | bool | Callable[[str], str] = True) → Iterator[dict]

获取某个目录之下的所有节点信息

Caution

当 fields 为空时，获取全部字段

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
parent_id – 顶层目录的 id
fields –
需要获取的字段
- 如果为 str，直接获取这个字段的值（不返回字典）
- 如果为 True，获取所有字段，且包括（”ancestors”, “path”, “relpath”, “depth”），返回字典
- 如果为 False，获取所有字段，但除了（”ancestors”, “path”, “relpath”, “depth”），返回字典
- 否则，获取所指定的这组字段，返回字典
escape –
对文件名进行转义
- 如果为 None，则不处理；否则，这个函数用来对文件名中某些符号进行转义，例如 “/” 等
- 如果为 True，则使用 posixpatht.escape，会对文件名中 “/”，或单独出现的 “.” 和 “..” 用 “" 进行转义
- 如果为 False，则使用 posix_escape_name 函数对名字进行转义，会把文件名中的 “/” 转换为 “|”
- 如果为 Callable，则用你所提供的调用，以或者转义后的名字

Returns:

迭代器，产生一组信息的字典，大概包含如下字段：

(
    # NOTE: 这些具体由你的数据库 ``data`` 表确定
    "id", "parent_id", "name", "sha1", "size", "pickcode",
    "is_dir", "is_alive", "extra", "created_at", "updated_at",
    # NOTE: 这些是另外附加的字段
    "ancestors", "path", "relpath", "depth",
)

iter_dup_files() → Iterator[dict][source]¶

罗列所有重复文件

Parameters:: self – P115QueryDB 实例或者数据库连接或游标
Returns:: 迭代器，一组文件的信息

iter_existing_id(ids: Iterable[int], /, is_alive: bool = True) → Iterator[int][source]¶

筛选出一系列 id 中，在数据库中存在的

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
ids – 一系列的节点 id
is_alive – 是否存活

Returns:

一系列 id 中在数据库中存在的那些的迭代器（实际直接返回一个游标）

iter_id_to_parent_id(ids: Iterable[int], /, recursive: bool = False) → Iterator[tuple[int, int]][source]¶

找出一系列 id 所对应的父 id，返回 (id, parent_id) 的 2 元组

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
ids – 一系列的节点 id
recursive – 是否递归，如果为 True，则还会处理它们的祖先节点

Returns:

迭代器，产生 (id, parent_id) 的 2 元组（实际直接返回一个游标）

iter_id_to_path(path: str | Sequence[str] = '', ensure_file: None | bool = None, parent_id: int = 0, is_alive: bool = True) → Iterator[int][source]¶

查询匹配某个路径的文件或目录的信息字典

Note

同一个路径可以有多条对应的数据

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
path – 路径
ensure_file –
是否文件
- 如果为 True，必须是文件
- 如果为 False，必须是目录
- 如果为 None，可以是文件或目录
parent_id – 顶层目录的 id

Returns:

迭代器，产生一组匹配指定路径的（文件或目录）节点的 id

iter_parent_id(ids: Iterable[int], /) → Iterator[int][source]¶

找出一系列 id 所对应的父 id

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
ids – 一系列的节点 id

Returns:

它们的父 id 的迭代器（实际直接返回一个游标）

path_to_id(path: str | Sequence[str] = '', ensure_file: None | bool = None, parent_id: int = 0, is_alive: bool = True) → int[source]¶

查询匹配某个路径的文件或目录的信息字典，只返回找到的第 1 个

Parameters:

self – P115QueryDB 实例或者数据库连接或游标
path – 路径
ensure_file –
是否文件
- 如果为 True，必须是文件
- 如果为 False，必须是目录
- 如果为 None，可以是文件或目录
parent_id – 顶层目录的 id

Returns:

找到的第 1 个匹配的节点 id

select_na_ids() → set[int][source]¶

找出所有的失效节点和悬空节点的 id

Note

悬空节点，就是此节点有一个祖先节点的 parant_id，不为 0 且不在 data 表中

Parameters:: self – P115QueryDB 实例或者数据库连接或游标
Returns:: 一组悬空节点的 id 的集合

对某个目录执行一次拉取，以更新 SQLite 数据

Parameters:

client – 115 网盘客户端对象
dbfile – 数据库文件路径，如果为 None，则自动确定
cid – 目录的 id 或 pickcode
recursive – 如果为 True，则拉取所有以之为祖先（先驱）节点的节点信息；否则，拉取所有以之为父（前驱）节点的节点信息
max_workers – 最大并发数，如果为 None 或 < 0 则自动确定，如果为 0 则单工作者惰性执行
max_files – 估计最大存在的文件数，<= 0 时则无限
max_dirs – 估计最大存在的目录数，<= 0 时则无限
app – 使用指定 app（设备）的接口
async – 是否异步
request_kwargs – 其它请求参数

Returns:

返回总共影响到数据行数，即所有 DML SQL 执行后，游标的 .rowcount 累加

p115client.tool.updatedb.updatedb_history_iter(client: str | PathLike | P115Client, dbfile: None | str | PathLike | Connection | Cursor = None, from_id: int = -1, from_time: float = 0, cooldown: float = 0.2, app: str = 'android', *, async_: Literal[False] = False, **request_kwargs) → Iterator[list[dict]][source]¶

p115client.tool.updatedb.updatedb_history_iter(client: str | PathLike | P115Client, dbfile: None | str | PathLike | Connection | Cursor = None, from_id: int = -1, from_time: float = 0, cooldown: float = 0.2, app: str = 'android', *, async_: Literal[True], **request_kwargs) → AsyncIterator[list[dict]]

持续采集 115 历史记录，以更新 SQLite 数据库

Note

当 from_id < 0 时，会从数据库获取最大 id 作为 from_id，获取不到时设为 0。当 from_id != 0 时，如果 from_time 为 0，则自动重设为 -1。

Parameters:

client – 115 网盘客户端对象
dbfile – 数据库文件路径，如果为 None，则自动确定
from_id – 开始的事件 id （不含），若 < 0 则是从数据库获取最大 id
from_time – 开始时间（含），若为 0 则从当前时间开始，若 < 0 则从最早开始
cooldown – 冷却时间，大于 0 时，两次接口调用之间至少间隔这么多秒
app – 使用指定 app（设备）的接口
async – 是否异步
request_kwargs – 其它请求参数

Returns:

迭代器，每次产生一批事件（从当前到上次截止）

from time import sleep
from p115client import P115Client
from p115client.tool import updatedb_history_iter

client = P115Client.from_path()

for event_list in updatedb_history_iter(client):
    if event_list:
        print("采集到历史记录列表:", event_list)
    else:
        sleep(1)

p115client.tool.updatedb.updatedb_initdb(con: Connection | Cursor, /) → Cursor[source]¶: 初始化数据库，然后返回游标

p115client.tool.updatedb.updatedb_life_iter(client: str | PathLike | P115Client, dbfile: None | str | PathLike | Connection | Cursor = None, from_id: int = -1, from_time: float = 0, cooldown: float = 0.2, app: str = 'android', *, async_: Literal[False] = False, **request_kwargs) → Iterator[list[dict]][source]¶

p115client.tool.updatedb.updatedb_life_iter(client: str | PathLike | P115Client, dbfile: None | str | PathLike | Connection | Cursor = None, from_id: int = -1, from_time: float = 0, cooldown: float = 0.2, app: str = 'android', *, async_: Literal[True], **request_kwargs) → AsyncIterator[list[dict]]

持续采集 115 生活日志，以更新 SQLite 数据库

Note

当 from_id < 0 时，会从数据库获取最大 id 作为 from_id，获取不到时设为 0。当 from_id != 0 时，如果 from_time 为 0，则自动重设为 -1。

Parameters:

client – 115 网盘客户端对象
dbfile – 数据库文件路径，如果为 None，则自动确定
from_id – 开始的事件 id （不含），若 < 0 则是从数据库获取最大 id
from_time – 开始时间（含），若为 0 则从当前时间开始，若 < 0 则从最早开始
cooldown – 冷却时间，大于 0 时，两次接口调用之间至少间隔这么多秒
app – 使用指定 app（设备）的接口
async – 是否异步
request_kwargs – 其它请求参数

Returns:

迭代器，每次产生一批事件（从当前到上次截止）

from time import sleep
from p115client import P115Client
from p115client.tool import updatedb_life_iter

client = P115Client.from_path()

for event_list in updatedb_life_iter(client):
    if event_list:
        print("采集到操作事件列表:", event_list)
    else:
        sleep(1)

updatedb¶

`updatedb`¶