Metadata-Version: 2.4
Name: zytools-fs
Version: 0.0.20
Summary: Personal Python utilities for FTP, video, and audio downloads
Author: zytools
License-Expression: MIT
Keywords: zytools,ftp,video,download
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE.txt
Requires-Dist: cryptography
Requires-Dist: lmdb
Requires-Dist: loguru
Requires-Dist: lxml
Requires-Dist: paramiko
Requires-Dist: requests
Requires-Dist: tqdm
Requires-Dist: trafilatura
Dynamic: license-file

# zytools-fs

`zytools-fs` 是一个轻量的 Python 工具集合，包含 FTP/SFTP 文件传输、任务心跳、URL 持久化去重、文章抓取、哔哩哔哩视频下载和猫耳 FM 音频下载等功能。各模块均可直接在脚本中导入使用，不依赖额外框架。

## 功能

- FTP 和 SFTP 文件/目录递归上传下载，支持重试、进度显示、按文件大小跳过和并发传输。
- 基于 `requests` 的哔哩哔哩视频下载器。
- 支持 DRM 解密及格式转换的猫耳 FM 音频下载器。
- 基于 LMDB 的大规模 URL 持久化去重。
- 文章页面识别和简单的同域名爬虫。
- 向 `/tasks` 接口上报脚本状态的任务心跳工具。
- 通过 Clash/Mihomo 外部控制器查询代理组、检测节点延迟和切换代理节点。
- 用于快速检查安装结果的 `zytools` 命令。

## 安装

```bash
pip install zytools-fs
```

要求 Python 3.9 或更高版本。

哔哩哔哩 DASH 音视频合并和猫耳音频格式转换依赖 `ffmpeg`，请先安装并确保命令已加入 `PATH`。未安装 `ffmpeg` 时，猫耳下载器无法生成最终音频文件。

安装后可运行以下命令检查：

```bash
zytools
```

预期输出：

```text
zytools installed successfully.
```

## FTP 和 SFTP 文件传输

`FTPClient` 和 `SFTPClient` 使用相同的上传下载接口。两者均支持单文件传输、目录递归传输、失败重试、传输进度、按大小跳过已有文件、并发传输和结果汇总。

### FTP 示例

```python
from zytools.utils import FTPClient

with FTPClient(
    host="127.0.0.1",
    user="user",
    password="password",
    port=21,
    passive=True,
    workers=4,
) as ftp:
    download_result = ftp.download("/remote/path", "./downloads")
    upload_result = ftp.upload("./reports", "/remote/reports")
```

### SFTP 示例

```python
from zytools.utils import SFTPClient

with SFTPClient(
    host="127.0.0.1",
    user="user",
    password="password",
    port=22,
    # key_filename="~/.ssh/id_rsa",
    # allow_unknown_host=True,  # 仅建议在可信的私有主机上使用
    workers=4,
) as sftp:
    download_result = sftp.download("/remote/path", "./downloads")
    upload_result = sftp.upload("./reports", "/remote/reports")
```

### 传输规则

- 文件路径只传输一个文件，目录路径会递归处理全部子目录和文件。
- 目标文件已存在且大小一致时会跳过，并计为成功。
- `show_progress=True` 时显示固定位置的进度条和完成日志；并发模式最多复用 `workers` 行进度显示。
- `download()` 和 `upload()` 均返回包含 `total`、`success`、`error` 的字典。
- 目录传输时设置 `workers>1` 可启用并发，每个工作线程使用独立连接。

返回值示例：

```python
{"total": 10, "success": 9, "error": 1}
```

常用参数：

- `port`：FTP 默认端口为 `21`，SFTP 默认端口为 `22`。
- `encoding`：FTP 文件名编码，默认为 `utf-8`。
- `passive`：是否使用 FTP 被动模式，默认为 `True`，仅 FTP 可用。
- `key_filename`：SFTP 使用的 SSH 私钥路径，支持 `~`。
- `allow_unknown_host`：是否允许未知的 SFTP 主机密钥，默认为 `False`。
- `download_retries`、`upload_retries`：下载和上传的尝试次数。
- `retry_wait_seconds`：两次重试之间的等待秒数。
- `workers`：目录上传下载的并发数，默认为 `1`。
- `show_progress`：是否显示进度条和完成日志。

## 哔哩哔哩视频下载

```python
from zytools.video import download_bili_video

ok = download_bili_video(
    "https://www.bilibili.com/video/BVxxxx?p=8",
    output_dir="./downloads",
    quality="max",
    filename="Bilibili_{BV}_{Date}_{Page}_{PartTitle}",
    cookie={
        "SESSDATA": "your_sessdata",
        "bili_jct": "your_bili_jct",
    },
    proxies={
        "http": "http://127.0.0.1:7890",
        "https": "http://127.0.0.1:7890",
    },
)

print(ok)
```

参数说明：

- `quality`：`"max"` 选择可用的最高画质，`"min"` 选择最低画质。实际画质取决于账号权限和接口返回的 DASH 流。
- `page`：可选的分 P 设置。省略时，URL 中包含 `p=8` 就只下载 P8；URL 中没有 `p` 则下载全部分 P。也可显式传入 `"all"`、`"1"` 或 `"1,3-5"`。
- `cookie`：可选的哔哩哔哩 Cookie，用于需要登录权限的视频。
- `proxies`：页面和 API 请求使用的 `requests` 格式代理字典；媒体流下载不使用该代理。
- `force`：为 `False` 时跳过已有的最终视频文件，为 `True` 时覆盖。

文件名模板支持以下字段：

- `{Title}`：视频标题。
- `{BV}`：BV 号。
- `{Date}`：`YYYYMMDD` 格式的发布日期。
- `{Page}`、`{Part}`：分 P 序号。
- `{Duration}`：时长，单位为秒。
- `{PartTitle}`：分 P 标题。

请仅下载自己拥有或已获得授权的内容。

## 猫耳 FM 音频下载

```python
from zytools.video import download_maoer_video

output_file = download_maoer_video(
    sound_id=13073155,
    filepath="./downloads/1.wav",
    proxies={
        "http": "http://127.0.0.1:7890",
        "https": "http://127.0.0.1:7890",
    },
)

print(output_file)
```

`filepath` 必须是完整的目标文件路径。扩展名决定输出格式，可使用 `.wav`、`.m4a`、`.mp3` 或 `.flac`。程序会自动创建父目录，不会使用猫耳页面标题作为文件名。

代理只用于页面、播放列表和 DRM API 请求；音频分片下载会绕过传入的代理及环境代理。函数成功后返回最终音频文件的绝对路径。

请仅下载自己拥有或已获得授权的内容。

## URL 去重

`UrlFilter` 在 LMDB 中保存 URL 的 MD5 指纹。它不是布隆过滤器，不会主动引入假阳性。

```python
from zytools.utils import UrlFilter

with UrlFilter(file_path="url_seen.lmdb") as url_filter:
    url = "https://example.com/video?id=1"

    if url_filter.add(url):
        print("首次出现")
    else:
        print("已经存在")

    print(len(url_filter))
```

批量添加、导出和导入：

```python
from zytools.utils import UrlFilter

with UrlFilter("url_seen.lmdb") as url_filter:
    added = url_filter.add_many([
        "https://example.com/a",
        "https://example.com/b",
    ])
    url_filter.to_csv("url_seen.csv")

print(f"新增 {added} 个 URL")

UrlFilter.to_lmdb("url_seen.csv", file_path="url_seen_copy.lmdb")
```

## 文章页面识别

使用 `check_response` 请求一个 URL，并将结果分类为文章、其他 HTML 页面、二进制资源或请求失败。

```python
from zytools.artice import check_response

result = check_response("https://example.com/news/1.html")

if result["type"] == "article":
    print(result["title"])
    print(result["date"])
    print(result["text"][:300])
else:
    print(result["type"], result.get("reason"))
```

`type` 可能为：

- `article`：文章页面，包含 `title`、`date`、`author` 和 `text`。
- `other`：不像文章的普通 HTML 页面。
- `binary`：图片、PDF、JavaScript、CSS、视频、压缩包等非 HTML 资源。
- `fetch_error`：请求失败或 HTTP 状态异常。

> 注意：当前公开模块名为 `zytools.artice`，请按上述拼写导入。

## 简单 URL 爬虫

`UrlCrawler` 从一个 URL 开始广度优先遍历链接，并逐条返回识别到的文章。可配合 `UrlFilter` 避免跨多次运行重复保存文章 URL。

```python
from zytools.artice import UrlCrawler
from zytools.utils import UrlFilter

with UrlFilter("article_urls.lmdb") as url_filter:
    crawler = UrlCrawler(
        start_url="https://example.com/",
        max_saved_urls=20,
        same_domain=True,
        max_depth=5,
        url_fp=url_filter,
    )

    for item in crawler.crawl():
        print(item["title"], item["url"])

    crawler.save_url_filter()
```

每条结果包含：

- `title`：文章标题。
- `creat_date`：文章发布日期（字段名保持现有接口拼写）。
- `content`：文章正文。
- `url`：最终文章 URL。
- `get_date`：当前抓取批次日期。

## 任务心跳

`update_task` 用于发送一次心跳；需要限制连续上报频率时可使用 `TaskUpdater`。

```python
from zytools.utils import TaskUpdater, update_task

result = update_task(
    name="daily job",
    machine_id="machine-1",
    script_path="/path/to/script.py",
    server="http://127.0.0.1:8001",
)

print(result)

task = TaskUpdater(
    name="daily job",
    machine_id="machine-1",
    script_path="/path/to/script.py",
    server="http://127.0.0.1:8001",
    min_interval=60,
)

task.update()
task.update(force=True)
```

服务端需要接收 `POST /tasks` 请求，请求体为 JSON，并包含 `name`、`machine_id`、`script_path`、`enabled` 和 `timeout_seconds` 字段。

## Clash 代理切换

`ClashVerge` 通过 Clash/Mihomo 外部控制器 API 查看代理组、检测节点延迟，并切换到指定或随机可用节点。控制器需要开启外部连接（默认地址为 `http://127.0.0.1:9090`）。

主要功能：

- `get_group()`：获取全部代理组名称。
- `get_group_nodes()`：递归展开嵌套代理组，返回去重后的实际节点名称。
- `get_activate()`：查看指定代理组当前选中的节点。
- `check_group_delay()`：批量检测代理组成员延迟，超时节点的延迟记为 `0`。
- `check_node_delay()`：检测单个节点的延迟。
- `switch_node()`：将 Selector 类型的代理组切换到指定节点。
- `switch_random_node()`：随机检测候选节点，并切换到第一个延迟检查通过的节点。
- `del_node()`：暂时排除不可用节点，到达 `node_delay_reset` 设置的时间后自动恢复候选资格。

请求失败、代理组不存在、节点不属于代理组或代理组不支持手动切换时，会抛出 `ClashAPIError`。

```python
from zytools.utils import ClashAPIError, ClashVerge

clash = ClashVerge(
    url="http://127.0.0.1:9090",
    secret="your-secret",
    default_group="主代理",
)

try:
    print(clash.get_group())
    print(clash.get_group_nodes())
    print(clash.get_activate())

    # 手动切换到指定节点
    clash.switch_node(proxy_name="香港节点")

    # 测试延迟并切换到随机可用节点
    result = clash.switch_random_node()
    print(result)  # {"group": "主代理", "node": "香港节点", "delay": 123}
except ClashAPIError as exc:
    print(f"Clash 操作失败：{exc}")
```

`switch_node` 也支持传入 `group_name` 和 `proxy_name`，`switch_random_node` 会自动跳过检测失败的节点，并在延迟一段时间后重新允许尝试。不要在不可信网络中暴露 Clash 控制器端口。

## 版本说明

- `0.0.19`：统一哔哩哔哩和猫耳的代理行为。页面与 API 请求使用传入的 `proxies`，媒体文件下载使用独立直连会话并忽略环境代理。
- `0.0.18`：未传 `page` 时自动识别哔哩哔哩 URL 的 `p` 参数。
- `0.0.17`：猫耳 `filepath` 改为完整目标文件名，扩展名用于选择输出格式。
- `0.0.16`：增加猫耳 FM 下载，并修复哔哩哔哩多分段下载、猫耳字节范围解析、URL 去重、HTTPS 校验、FTP/SFTP 文件完整性检查和任务接口非 JSON 响应处理等问题。

完整变更记录见源码包中的 `CHANGELOG.md`。

## 开发与发布

构建安装包：

```bash
python -m build
```

检查发布包元数据：

```bash
python -m twine check dist/*
```

发布到 PyPI：

```bash
python -m twine upload dist/*
```

## 许可证

MIT
