nutricare-data-packages 0.2.0

Nutricare data access package

nutricare-data-packages

Nutricare 的数据访问基础包，提供：

• StorageBase：统一写入入口，上传阿里云 OSS 并写入 MongoDB 元数据
• OssStorage：OSS 上传、存在性检查与预签名 URL
• MetadataStore：doc_metadata 集合的增删改查封装
• MongoRepositoryBase：MongoDB 集合访问基类
• HeartbeatDaemon：Worker 心跳上报守护进程

安装

pip install nutricare-data-packages

使用 OSS 功能时需额外安装可选依赖：

pip install nutricare-data-packages[oss]

环境变量

MongoDB（元数据存储）

变量	必需	说明
mongodb_url 或 MONGODB_URL	是	MongoDB 连接地址，支持 127.0.0.1:27017、mongodb:127.0.0.1:27017 或完整 mongodb://...
MONGODB_DB	是	数据库名
MONGODB_USERNAME	否	用户名
MONGODB_PASSWORD	否	密码
MONGODB_AUTH_SOURCE	否	认证库

业务上下文

变量	必需	说明
NUTRICARE_TICKET_ID	是	工单 ID，写入元数据
NUTRICARE_TASK_ID	是	任务 ID，写入元数据

阿里云 OSS

变量	必需	说明
OSS_ACCESS_KEY_ID	是	AccessKey ID
OSS_ACCESS_KEY_SECRET	是	AccessKey Secret
OSS_REGION	是	地域，如 cn-hangzhou
OSS_BUCKET_NAME	是	存储桶名称
OSS_ENDPOINT	否	Endpoint 或已绑定桶的自定义域名
OSS_USE_CNAME	否	自定义域名访问时设为 true；未设置且 Endpoint 非 *.aliyuncs.com 时自动启用

心跳（可选）

变量	必需	说明
NUTRICARE_HEARTBEAT_ENABLED	否	是否开启，默认 true
NUTRICARE_HEARTBEAT_URL	开启时	主程序心跳接收地址，如 http://host:8080/api/heartbeat
NUTRICARE_HEARTBEAT_INTERVAL	否	上报间隔（秒），默认 30
NUTRICARE_HEARTBEAT_TIMEOUT	否	单次 HTTP 超时（秒），默认 10
NUTRICARE_WORKER_ID	否	Worker 标识；未设置时使用 hostname:pid

快速使用

StorageBase：上传 OSS 并写入元数据

子类只需实现 storage_subdir（作为 OSS 对象键前缀），无需自行处理文件 I/O：

import datetime
import os

from nutricare_data_packages import StorageBase

# 运行前设置必需环境变量（见上文）
os.environ["mongodb_url"] = "127.0.0.1:27017"
os.environ["MONGODB_DB"] = "nutricare"
os.environ["NUTRICARE_TICKET_ID"] = "ticket-001"
os.environ["NUTRICARE_TASK_ID"] = "task-001"
os.environ["OSS_ACCESS_KEY_ID"] = "your-key-id"
os.environ["OSS_ACCESS_KEY_SECRET"] = "your-key-secret"
os.environ["OSS_REGION"] = "cn-hangzhou"
os.environ["OSS_BUCKET_NAME"] = "your-bucket"


class ReportStorage(StorageBase):
    @property
    def storage_subdir(self) -> str:
        return "reports"


storage = ReportStorage()
relative_path = "daily/report.txt"
content = b"hello"

# 统一写入：路径校验 → MD5 幂等 → 上传 OSS → 写入 doc_metadata
ok = storage.write(relative_path, content, title="日报")

# 查询元数据
doc = storage.metadata_collection.find_one(
    {"storage_subdir": "reports", "relative_path": relative_path}
)

# 获取私有 OSS 预签名 URL（默认 15 分钟有效）
url = storage.get_private_access_url(
    relative_path,
    expires=datetime.timedelta(hours=1),
)

write 行为说明：

• relative_path 为相对于 storage_subdir 的逻辑路径，禁止 ..、绝对路径及非法字符
• 支持 data（字节）或 file_path（本地文件）作为内容来源，二选一
• 可选参数：title、file_type、status、metadata、content_type
• 相同内容（MD5 一致且 OSS 对象已存在）时跳过上传，直接返回 True
• 新文件在 OSS 上使用分区路径（_p/xx/yy/...）以降低单目录文件数；已存在的旧路径仍兼容

OssStorage：直接使用 OSS

from nutricare_data_packages import OssStorage

oss = OssStorage()
oss.upload("reports/daily/report.pdf", pdf_bytes, content_type="application/pdf")
oss.exists("reports/daily/report.pdf")
url = oss.presigned_url("reports/daily/report.pdf")

MongoRepositoryBase：自定义 MongoDB 仓储

from nutricare_data_packages import MongoRepositoryBase


class UserRepo(MongoRepositoryBase):
    def __init__(self):
        super().__init__(
            mongodb_url="127.0.0.1:27017",
            db_name="nutricare",
            collection_name="users",
            auth_source="admin",  # 可选
            username="user",      # 可选
            password="pass",      # 可选
        )

心跳守护进程

from nutricare_data_packages import HeartbeatDaemon

daemon = HeartbeatDaemon()
daemon.start()   # 后台线程周期性上报
# ...
daemon.stop()    # 停止并在退出前标记 offline

也可使用 HeartbeatService().beat() 手动上报，或 run_blocking() 阻塞运行。

元数据表结构

集合名：doc_metadata。唯一键为 (storage_subdir, relative_path)，建议建立唯一索引。

主要字段：storage_subdir、relative_path、title、object_key、size、md5、file_type、ticket_id、task_id、status、metadata、created_at、updated_at。

支持的 file_type：pdf、doc、docx、md、txt、xls、xlsx、ppt、pptx；未显式传入时从路径扩展名推断。

本地构建与发布

python -m pip install --upgrade build twine
python -m build
python -m twine check dist/*
python -m twine upload dist/*