wtfutil

wtfutil is a Python utility library for everyday scripting and automation. It packs the most commonly needed "wheels" into one package: enhanced HTTP sessions, file I/O, encoding & crypto, SQLite/MySQL, process control, multi-channel push notifications, translation, and random images — organized into focused public submodules.

Author: vicrack  |  中文文档: README_zh.md  |  Full API reference: docs/

Installation

pip install wtfutil

Requires Python 3.10+.

⚠️ 1.3.0 Breaking Migration: Import from Submodules

[!IMPORTANT] wtfutil 1.3.0 removes package-root symbol re-exports. Import functions, classes, constants, and configuration objects from the public submodule that owns them.

Replace legacy package attributes such as wtfutil.read_lines(...), wtfutil.get_resource(...), and wtfutil.requests_session(...) with attributes on their owning modules:

import wtfutil.fileutil as fileutil
import wtfutil.httputil as httputil
import wtfutil.util as util

lines = fileutil.read_lines(util.get_resource("urls.txt"), unique=True)
session = httputil.requests_session(timeout=30)

Importing a physical public submodule (for example, from wtfutil import procutil) remains valid Python, but direct imports from wtfutil.<module> make ownership and migration clearer.

For example, from wtfutil.memshellutil import MemShellParty, MemShellPartyError is the canonical SDK import. wtfutil.memshell remains the CLI implementation module rather than a second SDK owner.

Feature Highlights

Quick Start

from wtfutil.fileutil import read_lines, write_json
from wtfutil.httputil import requests_session
from wtfutil.notifyutil import send
from wtfutil.util import get_resource

# HTTP session with proxy (int = 127.0.0.1:<port>)
req = requests_session(proxies=10809, timeout=30)
r = req.get("https://httpbin.org/ip")
print(r.json())

# Read a resource file (looks in cwd -> resource/ -> ~/)
resource_path = get_resource("urls.txt")
if resource_path is None:
    raise FileNotFoundError("urls.txt")
lines = read_lines(resource_path, unique=True)

# Write JSON
write_json("out.json", {"status": "ok", "count": len(lines)})

# Push to all configured channels at once
send("Job done", f"Processed {len(lines)} items")

HTTP — httputil

from wtfutil.httputil import requests_session
from urllib3 import Retry

# Minimal (random UA, TLS certificate verification disabled)
req = requests_session()

# Explicitly enable certificate verification when required
verified_req = requests_session(verify=True)

# Proxy + timeout
req = requests_session(proxies=10809, timeout=30)

# Custom retry + large connection pool (high-concurrency scraping)
req = requests_session(
    timeout=30,
    max_retries=Retry(total=5, backoff_factor=0.5, allowed_methods=["GET"]),
    pool_connections=100,
    pool_maxsize=100,
)

# Fixed base URL — use relative paths afterwards
req = requests_session(base_url="https://open.feishu.cn/open-apis", timeout=30)
r = req.get("/authen/v1/user_info")

# Rate limit: max 5 requests/sec
req = requests_session(rate_limit=5)

# Disk-based HTTP cache
req = requests_session(use_cache={"cache_name": "./data/http_cache"})

# Random X-Forwarded-For
req = requests_session(fake_ip=True)

# Debug mode: prints full request/response
req = requests_session(debug=True)

Send raw HTTP messages:

from wtfutil.httputil import httpraw

raw = """POST /api/login HTTP/1.1
Host: example.com
Content-Type: application/json

{"user":"admin","pass":"123"}
"""
resp = httpraw(raw, ssl=True, timeout=10)

URL / IP utilities:

from wtfutil.httputil import get_maindomain, is_port_in_use, is_private_ip, url2ip

is_private_ip("192.168.1.1")       # True
get_maindomain("sub.example.com")  # "example.com"
url2ip("example.com")              # "93.184.216.34"
is_port_in_use(8080)               # False

Files — fileutil

from wtfutil.fileutil import (
    file_md5,
    read_json,
    read_lines,
    read_text,
    write_json,
    write_lines,
    write_text,
)
from wtfutil.util import get_resource

# Read lines, skip blanks, preserve-order dedup
lines = read_lines("targets.txt", unique=True)

# Graceful read — returns [] instead of raising if file is missing
lines = read_lines("state.txt", not_exists_ok=True)

# Read JSON, returns {} if file is missing
config = read_json("config.json", not_exists_ok=True)

# Write JSON (ensure_ascii=False, indent=4)
write_json("result.json", {"items": lines, "total": len(lines)})

# Write lines
write_lines("output.txt", ["line1", "line2", "line3"])

# File MD5
print(file_md5("app.zip"))

# Resource file lookup: cwd → resource/ → ~/
path = get_resource("blacklist.txt")
if path is None:
    raise FileNotFoundError("blacklist.txt")
blacklist = read_lines(path, unique=True)

Strings & Crypto — strutil

from wtfutil.strutil import (
    str_md5, str_sha256,
    base64encode, base64decode,
    url_encode, url_decode,
    rsa_encrypt, rsa_decrypt,
    rand_base, get_middle_text,
)

# Hashing
str_md5("hello")                         # "5d41402abc4b2a76b9719d911017c592"
str_sha256(b"data")

# Base64
base64encode(b"hello world")             # "aGVsbG8gd29ybGQ="
base64decode("aGVsbG8gd29ybGQ=")         # "hello world"

# URL encoding
url_encode("a=1&b=hello world")          # "a%3D1%26b%3Dhello%20world"

# Random token (default charset: lowercase letters + digits, a-z0-9; pass letters= to customize)
token = rand_base(32)

# Extract text between two markers (useful for HTML scraping)
value = get_middle_text(html, 'name="token" value="', '"')

# RSA encryption (auto-segments long data)
encrypted = rsa_encrypt(b"secret data", public_key_pem)
plaintext = rsa_decrypt(encrypted, private_key_pem)

Database — sqlutil

from wtfutil.sqlutil import MYSQL, SQLite, next_id

# SQLite
db = SQLite("data.db")

db.execute("""
    CREATE TABLE IF NOT EXISTS items (
        id TEXT PRIMARY KEY,
        url TEXT,
        status INTEGER DEFAULT 0
    )
""")

# Single insert
db.insert("items", {"id": next_id(), "url": "https://a.com", "status": 0})

# Bulk insert
rows = [{"id": next_id(), "url": u, "status": 0} for u in url_list]
db.insert_many("items", rows)

# Query
row = db.select_one("items", where_clause={"url": "https://a.com"})
rows = db.select("items", where_clause={"status": 0})

# Update / delete
db.update("items", {"status": 1}, {"url": "https://a.com"})
db.delete("items", {"status": 0})

# MySQL — same API
db = MYSQL(host="127.0.0.1", user="root", password="pass", database="mydb")
db.insert_or_replace("items", {"id": "xxx", "url": "https://b.com"})

Notifications — notifyutil

Configure channels before importing notifyutil (via wtfconfig.ini or environment variables), then your code stays unchanged regardless of how many channels you add:

from wtfutil.notifyutil import send

# wtfconfig.ini
# [notify]
# FEISHU_KEY = your_webhook_key
# TG_BOT_TOKEN = 123456:xxx
# TG_USER_ID = 88888888
# BARK_PUSH = https://api.day.app/your_key

# Fan out to all configured channels concurrently
send("Scraper error", "Target site returned 403, pausing for 5 minutes")

Changing values in push_config at runtime does not rebuild the enabled-channel list, so use the ini file or environment variables to enable new channels.

Call a single channel directly:

from wtfutil.notifyutil import feishu_bot, telegram_bot

feishu_bot("Alert", "Disk usage exceeded 90%")
telegram_bot("Alert", "Disk usage exceeded 90%")

Single Instance — singleinstance

Prevent a script or scheduled job from running more than once at a time:

from wtfutil.singleinstance import SingleInstance, SingleInstanceException, single_instance

# Context manager
try:
    with SingleInstance(flavor_id="crawler_job"):
        run_crawler()
except SingleInstanceException:
    print("Already running, skipping")

# Decorator
@single_instance(flavor_id="data_sync")
def sync_data():
    ...

flavor_id lets multiple independent lock modes coexist in the same script.
Details: docs/en/singleinstance.md

Process Control — procutil

Find/kill is cross-platform; suspend/resume is Windows-only.

from wtfutil.procutil import (
    find_python_processes_by_cmdline,
    find_python_processes_by_script,
    kill_python_processes_by_script,
    resume_process_by_pid,
    suspend_process_by_pid,
)

# Find Python processes by script path
procs = find_python_processes_by_script("worker.py")

# Find by command-line substring
procs = find_python_processes_by_cmdline("celery worker")

# Suspend / resume (Windows only) / kill by script (cross-platform)
suspend_process_by_pid(pid)
resume_process_by_pid(pid)
kill_python_processes_by_script("worker.py")

pykill CLI (installed with the package):

pykill                       # list all Python processes, interactive multi-select kill
pykill worker.py             # kill by script path
pykill worker.py -l          # list only, no kill
pykill -c "celery worker"    # match by command-line substring

Details: docs/en/pykill.md

memshell CLI (MemShellParty memory-shell generator):

memshell config
memshell generate --shell-tool Behinder --shell-type Listener -o payload.txt
memshell probe -o payload.txt
memshell install-skill --project   # install to ./.agents/skills/memshell

Details: docs/en/memshellutil.md

Misc Utilities — util

from wtfutil.util import UniqueQueue, cut_list, group_data, measure_time

# Dedup queue: duplicate dicts are silently dropped (great for multi-thread crawlers)
q = UniqueQueue()
q.put({"url": "https://a.com"})
q.put({"url": "https://a.com"})  # ignored
print(q.qsize())  # 1

# Timing decorator
@measure_time
def heavy_task():
    ...

# Slice a list into fixed-size batches (bulk DB writes, batch API calls)
for batch in cut_list(url_list, 50):
    process_batch(batch)

# Group rows by a field
groups = group_data(rows, group_by="status")  # {0: [...], 1: [...]}

Configuration

Loaded via wtfutil.configutil (mtime hot-reload on ensure_section). Search order — env vars always win:

current working directory → resource/wtfconfig.ini → ~/wtfconfig.ini

[notify]
CONSOLE = true
FEISHU_KEY =
FEISHU_SECRET =
DD_BOT_TOKEN =
DD_BOT_SECRET =
TG_BOT_TOKEN =
TG_USER_ID =
BARK_PUSH =
SMTP_SERVER =
SMTP_EMAIL =
SMTP_PASSWORD =
WEBHOOK_URL =

[img]
APIHZ_IMG_ID =
APIHZ_IMG_KEY =

[memshell]
# BASE_URL = https://party.mem.mk

Full key list: wtfconfig.ini.example; API details: configutil, notifyutil, imgutil, memshellutil.

Module Overview

Public SDK APIs live in the physical submodules listed above. Import each symbol from its owning wtfutil.<module> path.

Contributing

Issues and pull requests are welcome on GitHub.

When adding or changing public APIs, update the owning module's API documentation under docs/en/ and docs/zh/. See AGENTS.md for agent-oriented project notes.