zytools-fs

zytools-fs is a small collection of Python utilities for personal crawling, FTP transfer, task heartbeats, persistent URL de-duplication, and Bilibili video downloads.

The package is intentionally lightweight: every helper can be imported and used directly in scripts without a framework.

Features

• FTP and SFTP recursive download/upload with retry, progress, skip-by-size, and concurrent transfer support.
• Bilibili video downloader based on requests.
• LMDB-backed URL filter for large crawl de-duplication.
• Article page detector and simple same-domain crawler.
• Task heartbeat helper for reporting script status to a /tasks endpoint.
• zytools command for a quick installation check.

Installation

pip install zytools-fs

Python 3.9 or newer is required.

For Bilibili DASH video merging, install ffmpeg and make sure it is available in PATH. If ffmpeg is not found, audio and video streams are kept as separate files.

Quick Check

zytools

Expected output:

zytools installed successfully.

FTP and SFTP Transfer

FTPClient and SFTPClient share the same high-level API for recursive upload/download. Both clients support per-file progress logs, retries, skip-by-size, concurrent directory transfers, and final transfer summaries.

FTP

from zytools.utils import FTPClient

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

SFTP

from zytools.utils import SFTPClient

with SFTPClient(
    host="127.0.0.1",
    user="user",
    password="password",
    port=22,
    # key_filename="~/.ssh/id_rsa",
    workers=4,
) as sftp:
    sftp.download("/remote/path", "./downloads")
    sftp.upload("./reports", "/remote/reports")

Transfer Behavior

• Direct file paths transfer one file; directory paths are traversed recursively.
• Existing target files with the same size are skipped and counted as success.
• Uploads and downloads print per-file progress when show_progress=True.
• Each download() or upload() call ends with total, success, and error.
• workers>1 enables concurrent file transfers for directories; each worker uses its own connection.

Example summary:

upload summary: total=10 success=9 error=1
download summary: total=10 success=10 error=0

Useful options:

• port: FTP defaults to 21; SFTP defaults to 22.
• encoding: FTP filename encoding, default utf-8.
• passive: FTP passive mode, default True (FTP only).
• key_filename: SSH private key path for SFTP authentication.
• download_retries and upload_retries: retry count.
• retry_wait_seconds: wait time between retries.
• workers: concurrent workers for directory downloads and uploads, default 1.
• show_progress: print transfer progress through loguru.

Bilibili Video Download

from zytools.video import download_bili_video

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

print(ok)

Parameters:

• quality: "max" for the highest available stream, "min" for the lowest. The actual quality depends on Bilibili account permissions and returned DASH streams.
• page: "all", a single page such as "1", or a range/list such as "1,3-5".
• cookie: optional Bilibili cookies for videos that require login.
• proxies: optional proxies for Bilibili page/API requests; media stream downloads do not use it.
• force: when False, existing final video files are skipped; when True, they are overwritten.

Filename template fields:

• {Title}: video title.
• {BV}: BV id.
• {Date}: publish date in YYYYMMDD format.
• {Page}: page number.
• {Part}: same as page number.
• {Duration}: duration in seconds.
• {PartTitle}: page title.

Only download content that you own or are allowed to download.

URL Filter

UrlFilter stores compact MD5 fingerprints in LMDB. Unlike a Bloom filter, it does not intentionally produce false positives.

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("new url")
    else:
        print("seen before")

    print(len(url_filter))

Batch import and export:

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 {added} urls")

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

Article Detection

Use check_response to request one URL and classify it as an article, other HTML page, binary resource, or fetch error.

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"))

Return type values:

• article: article page with extracted title, date, author, and text.
• other: HTML page that does not look like an article.
• binary: image, PDF, JavaScript, CSS, video, archive, or other non-HTML file.
• fetch_error: request failed or returned a bad HTTP status.

Simple URL Crawler

UrlCrawler starts from one URL, follows links breadth-first, and yields article items. It can optionally use UrlFilter to avoid saving the same article URL across runs.

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()

Each yielded item has:

• title: extracted article title.
• creat_date: extracted article publish date.
• content: extracted article text.
• url: final article URL.
• get_date: crawl batch date.

Task Heartbeat

Use update_task for a single heartbeat request, or TaskUpdater when a script needs repeated updates with a minimum interval.

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)

The server is expected to accept POST /tasks with a JSON body containing name, machine_id, script_path, enabled, and timeout_seconds.

Development

Build the package locally:

python -m build

Check the distribution metadata:

python -m twine check dist/*

Publish to PyPI:

python -m twine upload dist/*

License

MIT