dddtools 0.1.3
常用工具函数集合
License: MIT
Requires Python: >=3.8
Keywords: utils, tools

# dddtools

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

---

## 目录

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

---

## 安装

### PyPI 安装（稳定版）

```bash
pip install dddtools
```

### 本地安装（开发版）

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

### 从源码安装

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

---

## 快速开始

```python
# 导入所有功能
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=["admin@example.com"],
    subject="系统通知",
    content="任务已完成",
    username="your@qq.com",
    password="your授权码"
)
```

---

## 日志模块 (logging)

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

### 默认日志器

```python
from dddtools import logger

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

### 自定义日志器

```python
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 - 耗时统计

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

```python
from dddtools import timer

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

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

### log_call - 调用日志

记录函数调用详情。

```python
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="zhangsan@example.com")
# 输出:
# [create_user] 传入: (name='张三', age=25, email='zhangsan@example.com')
# [create_user] 返回: {'name': '张三', 'age': 25, 'email': 'zhangsan@example.com'} | 耗时: 0.12ms
```

#### log_call 参数

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

---

## FTP操作 (ftp)

### 连接管理

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

### 获取文件列表

```python
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: 修改时间
```

### 下载文件

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

### 上传文件

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

### 删除操作

```python
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 完整示例

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

### 快速发送

```python
from dddtools import mail

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

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

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

### 使用 MailClient

```python
from dddtools import mail

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

# 多次发送
client.send(
    to_addrs=["user1@example.com"],
    subject="通知1",
    content="内容1"
)

client.send(
    to_addrs=["user2@example.com"],
    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` |

### 基础用法

```python
from dddtools.mail.template import render

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

### 验证码模板

```python
from dddtools.mail.template import render, mail

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

mail.send_html(
    to_addrs=["user@example.com"],
    subject="验证码",
    html_content=html,
    username="123456@qq.com",
    password="授权码"
)
```

### 订单通知模板

```python
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=["user@example.com"],
    subject="订单已发货",
    html_content=html,
    username="123456@qq.com",
    password="授权码"
)
```

### 数据表格模板

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

### 指标卡片模板

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

### 辅助函数

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

### 注册自定义模板

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

### 连接管理

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

### 基础操作

```python
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 操作

```python
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 操作

```python
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")  # 长度
```

### 缓存装饰器

```python
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加密

```python
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加密

```python
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签名/验签

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

### 连接管理

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

### 集合操作

```python
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: 如何更改日志保留天数？

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

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

```python
@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: 缓存装饰器如何动态生成键？

```python
@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