Metadata-Version: 2.4
Name: zytools-fs
Version: 0.0.10
Summary: Personal Python utilities for FTP and video 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: lmdb
Requires-Dist: loguru
Requires-Dist: lxml
Requires-Dist: requests
Requires-Dist: trafilatura
Dynamic: license-file

# 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 recursive download and upload with retry 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

```bash
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

```bash
zytools
```

Expected output:

```text
zytools installed successfully.
```

## FTP Transfer

```python
from zytools.utils import FTPDownloader

with FTPDownloader(
    host="127.0.0.1",
    user="user",
    password="password",
) as ftp:
    ftp.download("/remote/path", "./downloads")
    ftp.upload("./reports", "/remote/reports")
```

Useful options:

- `port`: FTP port, default `21`.
- `encoding`: FTP filename encoding, default `utf-8`.
- `passive`: whether to use passive mode, default `True`.
- `download_retries` and `upload_retries`: retry count.
- `show_progress`: print transfer progress through `loguru`.

## Bilibili Video Download

```python
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",
    },
)

print(ok)
```

Parameters:

- `quality`: `"max"` for the highest available stream, `"min"` for the lowest.
- `page`: `"all"`, a single page such as `"1"`, or a range/list such as
  `"1,3-5"`.
- `force`: overwrite existing output files when set to `True`.
- `cookie`: optional Bilibili cookies for videos that require login.

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.

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

    print(len(url_filter))
```

Batch import and export:

```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 {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.

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

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.

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

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.

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

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:

```bash
python -m build
```

Check the distribution metadata:

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

Publish to PyPI:

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

## License

MIT
