dddtools 0.1.3

常用工具函数集合

dddtools

常用工具函数集合。支持 Python 3.8+。

---

目录

• 安装
• 快速开始
• 日志模块 (logging)
• 装饰器 (decorator)
• FTP操作 (ftp)
• 邮件发送 (mail)
• HTML模板 (mail.template)
• 缓存管理 (redis)
• 文件结构
• 常见问题

---

安装

PyPI 安装（稳定版）

pip install dddtools

本地安装（开发版）

cd /path/to/dddtools
pip install -e .

从源码安装

git clone https://github.com/yourname/dddtools.git
cd dddtools
pip install -e .

---

快速开始

# 导入所有功能
from dddtools import logger, timer, log_call, ftp, mail

# 1. 日志自动启动
logger.info("dddtools 已就绪")

# 2. 使用装饰器
@timer
@log_call()
def fetch_data():
    return {"data": [1, 2, 3]}

result = fetch_data()

# 3. FTP操作
with ftp.connect("ftp.example.com", username="user", password="pass") as conn:
    files = conn.list_files("/")
    conn.download_file("/pub/data.csv", "./data.csv")

# 4. 发送邮件
mail.send_simple(
    to_addrs=["[e-mail]"],
    subject="系统通知",
    content="任务已完成",
    username="[e-mail]",
    password="your授权码"
)

---

日志模块 (logging)

自动创建logs目录，按日期存储，自动清理7天前日志。

默认日志器

from dddtools import logger

logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告")
logger.error("错误")
logger.critical("严重错误")

自定义日志器

from dddtools import get_logger

# 自定义名称、目录、保留天数
app_logger = get_logger(
    name="MyApp",           # 日志器名称
    log_dir="/var/log/myapp",  # 日志目录
    days=30                  # 保留30天
)

app_logger.info("自定义日志器")

日志格式

2026-02-07 14:03 | INFO     | dddtools     | dddtools 已就绪
2026-02-07 14:03 | INFO     | MyApp      | 自定义日志器

• 自动在运行目录创建 logs 文件夹
• 按日期切割：2026-02-07.log
• 同时输出到控制台和文件
• 默认保留7天

---

装饰器 (decorator)

timer - 耗时统计

统计函数执行时间（毫秒）。

from dddtools import timer

@timer
def slow_function():
    time.sleep(2)
    return "完成"

result = slow_function()
# 输出: [slow_function] 耗时: 2002.35ms

log_call - 调用日志

记录函数调用详情。

from dddtools import log_call

# 基本用法（使用函数名作为日志名）
@log_call()
def add(a, b):
    return a + b

# 自定义日志名
@log_call(name="计算器")
def multiply(x, y):
    return x * y

# 自定义参数截断长度
@log_call(arg_limit=500, return_limit=800)
def process(data):
    return {"result": data}

# 带关键字参数
@log_call()
def create_user(name, age, *, email=None):
    return {"name": name, "age": age, "email": email}

# 调用示例
add(1, 2)
# 输出:
# [add] 传入: (a=1, b=2)
# [add] 返回: 3 | 耗时: 0.05ms

create_user("张三", 25, email="[e-mail]")
# 输出:
# [create_user] 传入: (name='张三', age=25, email='[e-mail]')
# [create_user] 返回: {'name': '张三', 'age': 25, 'email': '[e-mail]'} | 耗时: 0.12ms

log_call 参数

参数	类型	默认值	说明
name	str	函数名	日志名称
arg_limit	int	1000	参数字符串截断长度
return_limit	int	1000	返回值字符串截断长度

---

FTP操作 (ftp)

连接管理

from dddtools import ftp

# 基本连接
with ftp.connect("ftp.example.com") as conn:
    pass

# 带认证
with ftp.connect(
    host="ftp.example.com",
    port=21,
    username="your_user",
    password="your_password",
    timeout=30
) as conn:
    pass

# 匿名连接
with ftp.connect("ftp.example.com", username="anonymous", password="") as conn:
    pass

获取文件列表

from dddtools import ftp

with ftp.connect("ftp.example.com", username="user") as conn:
    # 获取当前目录列表
    files = conn.list_files("/")
    for f in files:
        print(f"{'[DIR]' if f.is_dir else 'FILE']} {f.name} ({f.size} bytes)")
    
    # 递归获取所有文件
    all_files = conn.list_all("/", recursive=True)
    
    # FTPFile 属性
    # - name: 文件名/完整路径
    # - size: 文件大小（字节）
    # - is_dir: 是否为文件夹
    # - modify_time: 修改时间

下载文件

from dddtools import ftp

with ftp.connect("ftp.example.com", username="user") as conn:
    # 下载单个文件
    conn.download_file(
        remote_path="/pub/data/report.csv",
        local_path="./downloads/report.csv",
        overwrite=True  # 是否覆盖
    )
    
    # 下载整个文件夹
    conn.download_folder(
        remote_folder="/pub/backups",
        local_folder="./backups",
        overwrite=True
    )

上传文件

from dddtools import ftp

with ftp.connect("ftp.example.com", username="user") as conn:
    # 上传单个文件
    conn.upload_file(
        local_path="./data.csv",
        remote_path="/pub/uploads/data.csv",
        overwrite=True
    )
    
    # 上传整个文件夹
    conn.upload_folder(
        local_folder="./project",
        remote_path="/pub/projects/project",
        overwrite=True
    )

删除操作

from dddtools import ftp

with ftp.connect("ftp.example.com", username="user") as conn:
    # 删除文件
    conn.delete_file("/pub/old_file.txt")
    
    # 删除空文件夹
    conn.delete_folder("/pub/empty_folder")
    
    # 递归删除（文件夹及其内容）
    conn.delete_recursive("/pub/to_delete")

FTP 完整示例

from dddtools import ftp, logger

def backup_website():
    """备份网站文件"""
    logger.info("开始备份")
    
    with ftp.connect("ftp.yoursite.com", username="admin", password="pass") as conn:
        # 获取文件列表
        files = conn.list_all("/public_html", recursive=True)
        logger.info(f"共找到 {len(files)} 个文件")
        
        # 下载
        conn.download_folder(
            remote_folder="/public_html",
            local_folder="./backup"
        )
    
    logger.info("备份完成")

backup_website()

---

邮件发送 (mail)

快速发送

from dddtools import mail

# 发送普通文本邮件
mail.send_simple(
    to_addrs=["[e-mail]", "[e-mail]"],
    subject="系统通知",
    content="您的订单已发货，请注意查收。",
    username="[e-mail]",
    password="your授权码"
)

# 发送HTML邮件
mail.send_html(
    to_addrs=["[e-mail]"],
    subject="欢迎注册",
    html_content="<h1>欢迎使用 dddtools</h1><p>感谢您的注册</p>",
    username="[e-mail]",
    password="your授权码"
)

# 发送带附件的邮件
mail.send_with_attachment(
    to_addrs=["[e-mail]"],
    subject="月度报告",
    content="请查收附件中的月度报告。",
    attachments=[
        "/path/to/report.pdf",
        "/path/to/summary.xlsx"
    ],
    username="[e-mail]",
    password="your授权码"
)

使用 MailClient

from dddtools import mail

# 创建客户端（重复发送时更高效）
client = mail.create_client(
    username="[e-mail]",
    password="your授权码",
    smtp_host="smtp.qq.com",
    smtp_port=465
)

# 多次发送
client.send(
    to_addrs=["[e-mail]"],
    subject="通知1",
    content="内容1"
)

client.send(
    to_addrs=["[e-mail]"],
    subject="通知2",
    content="内容2",
    is_html=True,
    attachments=["file.pdf"]
)

参数说明

参数	说明
to_addrs	收件人列表，支持多个
subject	邮件主题
content	邮件内容（普通或HTML）
is_html	是否为HTML格式，默认False
attachments	附件路径列表
username	发件人邮箱
password	SMTP授权码（不是密码）
smtp_host	SMTP服务器，默认 smtp.qq.com
smtp_port	SMTP端口，默认 465

QQ邮箱授权码获取

1. 登录 QQ 邮箱
2. 进入「设置」→「账户」
3. 找到「POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务」
4. 开启「POP3/SMTP服务」
5. 点击「生成授权码」

---

HTML模板 (mail.template)

8种内置模板

模板名	用途	变量
simple	基础模板	content, now
notification	通知/公告	title, content, button_url, button_text, highlights
verification	验证码	title, description, code
welcome	欢迎邮件	username, site_name, message, feature1, feature2, feature3
data_table	表格数据	title, description, table_html, summary
order	订单通知	username, order_no, order_time, status, tracking_no, total_amount, order_details
password_reset	密码重置	username, reset_url, expire_time
report	周报/月报	report_title, period, metrics_html, highlights_html, notes

基础用法

from dddtools.mail.template import render

# 渲染模板
html = render("simple", content="这是一封测试邮件")
html = render("notification", title="重要通知", content="您的账户已更新")

验证码模板

from dddtools.mail.template import render, mail

html = render(
    "verification",
    title="验证码",
    description="您的验证码如下，请于5分钟内完成验证：",
    code="852741"
)

mail.send_html(
    to_addrs=["[e-mail]"],
    subject="验证码",
    html_content=html,
    username="[e-mail]",
    password="授权码"
)

订单通知模板

from dddtools.mail.template import render, mail

html = render(
    "order",
    username="张三",
    order_no="DD20260207001",
    order_time="2026-02-07 14:30:00",
    status="已发货",
    tracking_no="SF1234567890",
    total_amount="299.00"
)

mail.send_html(
    to_addrs=["[e-mail]"],
    subject="订单已发货",
    html_content=html,
    username="[e-mail]",
    password="授权码"
)

数据表格模板

from dddtools.mail.template import render, mail, make_table_html

# 生成表格HTML
table = make_table_html(
    headers=["姓名", "数学", "语文", "英语"],
    rows=[
        ["张三", 95, 88, 92],
        ["李四", 85, 90, 87],
        ["王五", 92, 85, 94]
    ]
)

html = render(
    "data_table",
    title="期末考试成绩",
    description="本次考试共3人参加",
    table_html=table,
    summary="平均分: 90.2"
)

指标卡片模板

from dddtools.mail.template import render, mail, make_metrics_html

# 生成指标HTML
metrics = make_metrics_html({
    "总访问量": {"value": "12,345", "change": "+15%", "type": "up"},
    "新增用户": {"value": "528", "change": "+8%", "type": "up"},
    "转化率": {"value": "3.2%", "change": "-0.5%", "type": "down"},
    "流失率": {"value": "5.1%", "change": "0%", "type": "neutral"}
})

html = render(
    "report",
    report_title="周报",
    period="2026-02-01 ~ 2026-02-07",
    metrics_html=metrics
)

辅助函数

from dddtools.mail.template import make_table_html, make_metrics_html, make_highlights_html

# 表格
make_table_html(
    headers=["列1", "列2", "列3"],
    rows=[["a", "b", "c"], ["d", "e", "f"]]
)

# 指标（up=绿色，down=红色，neutral=黑色）
make_metrics_html({
    "指标名": {"value": "数值", "change": "+10%", "type": "up"}
})

# 亮点列表
make_highlights_html(["亮点1", "亮点2", "亮点3"])

注册自定义模板

from dddtools.mail.template import register

# 注册模板
register("newsletter", """
<!DOCTYPE html>
<html>
<body>
    <h1>{{title}}</h1>
    <div>{{content}}</div>
    <footer>unsubscribe at {{unsubscribe_url}}</footer>
</body>
</html>
""")

# 使用
html = render("newsletter", title="新闻简报", content="...", unsubscribe_url="...")

---

缓存管理 (redis)

连接管理

from dddtools import connect, RedisClient

# 方式一：使用 connect 函数
redis = connect(host="localhost", port=6379, db=0, password="your_password")

# 方式二：使用连接字符串
redis = RedisClient(host="redis.example.com", port=6379, password="pass")

# 测试连接
redis.ping()  # True/False

# 使用原生 Redis 客户端
redis.client.set("key", "value")

基础操作

from dddtools import redis

r = redis.connect(host="localhost", password="your_password")

# 字符串操作
r.set("name", "张三", ex=3600)  # 1小时过期
value = r.get("name")

# 检查存在
r.exists("name")  # 返回数量

# 设置过期
r.expire("name", 1800)  # 30秒
print(r.ttl("name"))  # 查看剩余时间

# 删除
r.delete("name")

Hash 操作

r = redis.connect()

# 设置 Hash
r.hset("user:1", "name", "张三")
r.hset("user:1", "age", "25")

# 获取字段
r.hget("user:1", "name")  # "张三"

# 获取全部
r.hgetall("user:1")  # {'name': '张三', 'age': '25'}

# 删除字段
r.hdel("user:1", "age")

List 操作

r = redis.connect()

# 插入
r.lpush("queue", "task1", "task2")
r.rpush("queue", "task3")

# 获取
r.lrange("queue", 0, -1)  # ['task2', 'task1', 'task3']
r.lpop("queue")  # 'task2'
r.llen("queue")  # 长度

缓存装饰器

from dddtools import cache

# 简单缓存
@cache(key="user:{user_id}", expire=600)
def get_user(user_id: int):
    # 首次调用：执行函数并缓存结果
    return db.query_user(user_id)
    # 后续调用：直接从 Redis 返回缓存

# 动态键
@cache(key="{id}", expire=300)
def get_data(id: int):
    return api.fetch(id)

# 清除缓存
from dddtools import clear_cache, clear_prefix

clear_cache(key="user:123")  # 清除单个
clear_prefix(prefix="cache")  # 清除前缀（谨慎）

缓存装饰器参数

参数	类型	默认值	说明
key	str	空	缓存键，支持 {func_name}, {args[i]}, {kwargs[key]}
expire	int	300	过期时间（秒）
prefix	str	"cache"	缓存前缀

---

加密模块 (encryption)

AES加密

from dddtools import AESEncrypt, encrypt_aes, decrypt_aes

# 方式一：使用类
aes = AESEncrypt(key="16位密钥字符串")
encrypted = aes.encrypt("Hello World")
decrypted = aes.decrypt(encrypted)

# 方式二：快速函数
encrypted = encrypt_aes("敏感数据", "密钥")
decrypted = decrypt_aes(encrypted, "密钥")

RSA加密

from dddtools import RSAEncrypt, generate_rsa_keys, encrypt_rsa, decrypt_rsa

# 生成密钥对
private_key, public_key = generate_rsa_keys(key_size=2048)

# 加密/解密
rsa = RSAEncrypt(private_key=private_key, public_key=public_key)
encrypted = rsa.encrypt("秘密消息")
decrypted = rsa.decrypt(encrypted)

# 快速函数
encrypted = encrypt_rsa("消息", public_key)
decrypted = decrypt_rsa(encrypted, private_key)

RSA签名/验签

from dddtools import sign_data, verify_signature

# 签名
private_key = "-----BEGIN RSA PRIVATE KEY-----..."
signature = sign_data("要签名的数据", private_key, hash_method="sha256")

# 验签
public_key = "-----BEGIN PUBLIC KEY-----..."
is_valid = verify_signature("要验证的数据", signature, public_key)
# 返回 True/False

参数说明

功能	参数	说明
generate_rsa_keys	key_size	密钥长度（1024/2048/4096）
AESEncrypt	key	16/24/32字节（对应AES-128/192/256）
sign_data	hash_method	哈希算法（md5/sha1/sha256）

---

MongoDB数据库 (mongodb)

连接管理

from dddtools import MongoDB, mongo_connect

# 方式一：使用 connect 函数
db = mongo_connect("mongodb://localhost:27017", db_name="myapp")

# 方式二：使用类
mongo = MongoDB("mongodb://localhost:27017")
mongo.set_db("myapp")

# 测试连接
print(mongo.ping())  # True/False

mongo.close()

集合操作

from dddtools import mongo_connect, MongoCollection

db = mongo_connect(db_name="testdb")
users = db.get_collection("users")  # 返回 MongoCollection 对象

# 插入
id = users.insert_one({"name": "张三", "age": 25})
ids = users.insert_many([
    {"name": "李四", "age": 30},
    {"name": "王五", "age": 28}
])

# 查询
user = users.find_one({"name": "张三"})
user = users.find_by_id("507f1f77bcf86cd799439011")  # 根据ID查询

all_users = users.find_all()  # 查询所有
all_users = users.find_all({"age": {"$gt": 25}})  # 条件查询
all_users = users.find_all(sort=[("age", -1)], limit=10)  # 排序和限制

count = users.count({"name": "张三"})  # 统计数量
exists = users.exists({"name": "李四"})  # 判断是否存在

# 更新
users.update_one({"name": "张三"}, {"age": 26})  # 更新单个
users.update_by_id("507f1f77bcf86cd799439011", {"age": 27})  # 根据ID更新
users.increment({"name": "李四"}, "age", 1)  # 字段自增

# 删除
users.delete_one({"name": "王五"})  # 删除单个
users.delete_by_id("507f1f77bcf86cd799439011")  # 根据ID删除
users.delete_many({"status": "inactive"})  # 删除多个

# 聚合
pipeline = [
    {"$match": {"age": {"$gte": 25}}},
    {"$group": {"_id": None, "avg_age": {"$avg": "$age"}, "count": {"$sum": 1}}}
]
result = users.aggregate(pipeline)

# 索引
users.create_index([("name", 1)])  # 普通索引
users.create_unique_index("email")  # 唯一索引

MongoCollection 方法

方法	说明
insert_one(doc)	插入单个文档，返回ID
insert_many(docs)	插入多个文档，返回ID列表
find_one(query)	查询单个文档
find_by_id(id)	根据ID查询
find_all(query, sort, limit)	查询多个文档
count(query)	统计数量
exists(query)	判断是否存在
update_one(query, update)	更新单个
update_by_id(id, update)	根据ID更新
update_many(query, update)	更新多个
increment(query, field, amount)	字段自增
delete_one(query)	删除单个
delete_by_id(id)	根据ID删除
delete_many(query)	删除多个
aggregate(pipeline)	聚合查询
create_index(keys)	创建索引
create_unique_index(field)	创建唯一索引

---

文件结构

dddtools/
├── pyproject.toml          # Python项目配置
├── README.md               # 本文档
├── .gitignore              # Git忽略配置
└── src/dddtools/
    ├── __init__.py         # 统一导出所有功能
    ├── logging/            # 日志模块
    │   └── __init__.py     # 自动日志、get_logger
    ├── decorator/          # 装饰器
    │   ├── __init__.py     # timer, log_call
    │   ├── timer.py        # 耗时统计
    │   └── log_call.py     # 调用日志
    ├── ftp/               # FTP操作
    │   ├── __init__.py     # 导出所有FTP函数
    │   ├── connection.py   # 连接管理
    │   ├── list.py         # 文件列表
    │   ├── download.py     # 下载
    │   ├── upload.py       # 上传
    │   └── delete.py       # 删除
    ├── mail/              # 邮件发送
    │   ├── __init__.py     # send_simple, send_html, send_with_attachment
    │   └── template/      # HTML模板
    │       └── __init__.py # 8种模板 + 辅助函数
    ├── redis/             # 缓存管理
    │   ├── __init__.py     # RedisClient, connect, cache, clear_cache
    │   └── decorator.py    # 缓存装饰器
    ├── encryption/         # 加密模块
    │   ├── __init__.py     # RSAEncrypt, AESEncrypt
    │   ├── rsa.py          # RSA加解密、签名验签
    │   └── aes.py          # AES加解密
    ├── mongodb/            # MongoDB数据库
    │   ├── __init__.py     # MongoDB, connect, MongoCollection
    │   └── collection.py   # 集合操作
    ├── file/              # (预留) 文件操作
    ├── string/            # (预留) 字符串处理
    ├── system/            # (预留) 系统工具
    └── datetime/          # (预留) 日期时间

---

常见问题

Q: 日志文件在哪里？

默认在运行目录下创建 logs 文件夹。可通过 get_logger(log_dir="/path") 自定义。

Q: 如何更改日志保留天数？

logger = get_logger(days=30)  # 保留30天

Q: log_call 如何自定义日志名？

@log_call(name="MyFunction")
def my_func(): ...

Q: FTP 连接失败？

• 检查用户名密码是否正确
• 检查服务器地址和端口
• 确保防火墙开放 FTP 端口（21）
• 尝试使用主动模式：ftp.connect(...)._ftp.set_pasv(False)

Q: QQ 邮箱发送失败？

• 确保已开启 POP3/SMTP 服务
• 使用授权码而非登录密码
• 确认 SMTP 地址和端口（smtp.qq.com:465）

Q: 如何调试 log_call？

设置环境变量或查看控制台输出，它会显示参数和返回值。

Q: 支持异步吗？

当前版本为同步实现。如需异步支持，可使用 asyncio.to_thread 包装。

Q: Redis 连接失败？

• 检查 Redis 服务是否启动
• 确认 host、port、password 是否正确
• 检查防火墙是否开放 6379 端口

Q: 缓存装饰器如何动态生成键？

@cache(key="{user_id}:{page}", expire=300)
def get_user_posts(user_id: int, page: int):
    return db.query_posts(user_id, page)

Q: 如何查看当前缓存？

直接使用 redis.client.keys("cache:*") 或 scan_iter 遍历。

---

License

MIT License