nutricare-data-packages 0.2.2

Nutricare data access package

nutricare-data-packages

Nutricare 的数据访问基础包（Python ≥ 3.9），面向采集 Worker 与数据脚本，统一封装：

模块	说明
StorageBase	统一写入入口：校验路径 → MD5 幂等 → 上传 OSS → 写入 MongoDB 元数据
OssStorage	阿里云 OSS（SDK V2）上传、存在性检查与预签名 URL
MetadataStore	doc_metadata 集合的 upsert / 按路径查询
ExecutionRecordStore	execution_records 工单执行记录的增删改查与统计
ExecutionRecordLifecycle	进程启动自动建记录、退出自动更新状态（可选 atexit / 信号钩子）
MongoRepositoryBase	MongoDB 集合访问基类（URL 规范化、CRUD 封装）
HeartbeatDaemon	Worker 向主程序周期性上报在线心跳

目录

• 安装
• 环境变量
• 快速使用
• 数据表结构
• 典型 Worker 集成
• 本地开发与测试
• 构建与发布

安装

pip install nutricare-data-packages

环境变量

布尔类变量（如 NUTRICARE_EXECUTION_RECORD_ENABLED）支持：true / false、1 / 0、yes / no、on / off（不区分大小写）。

MongoDB（元数据与执行记录共用）

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

业务上下文

写入元数据、心跳上报、执行记录生命周期会读取工单 ID（get_ticket_id_from_env()）：

变量	必需	说明
NUTRICARE_TICKET_ID	是*	工单 ID

* 使用 MetadataStore / StorageBase.write、心跳或 ExecutionRecordLifecycle 时必需。任务 ID（task_id）由 ExecutionRecordLifecycle.start() 创建的执行记录 id 提供，不再通过环境变量配置。

执行记录生命周期（可选）

变量	必需	说明
NUTRICARE_EXECUTION_RECORD_ENABLED	否	是否自动维护执行记录，默认 true

退出状态固定不可配置：正常 resolved、中断 closed、异常 failed。

需已配置 MongoDB，且设置 NUTRICARE_TICKET_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，最小 1
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["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（省略 expires 时 SDK 默认 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（默认 success）、metadata、content_type
幂等	相同 MD5 且 OSS 对象已存在时跳过上传，返回 True
分区路径	新文件在 OSS 上使用 _p/xx/yy/... 分区以降低单目录对象数；已存在的未分区旧路径仍兼容
object_key	元数据中保存分区后的相对键；get_private_access_url 会据此解析实际 OSS 键

从本地文件上传示例：

storage.write("exports/report.pdf", file_path="/tmp/report.pdf", title="导出报告")

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

可通过 load_oss_config(...) 显式传入配置，否则从环境变量加载。

MetadataStore：仅操作元数据

不继承 StorageBase、只需读写 doc_metadata 时：

from nutricare_data_packages import MetadataStore, make_filter

store = MetadataStore()
store.save(
    storage_subdir="reports",
    relative_path="daily/report.txt",
    title="日报",
    object_key="_p/ab/cd/daily/report.txt",
    size=5,
    md5="...",
    file_type="txt",
    status="success",
)
doc = store.find_by_path("reports", "daily/report.txt")

ticket_id 从环境变量读取，task_id 从执行记录生命周期上下文读取，无需每次 save 传入。

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()   # 后台线程周期性 POST online
# ... 执行业务 ...
daemon.stop()    # 停止并在退出前上报 offline

也可使用 HeartbeatService().beat() 手动上报，或 run_blocking() 在当前线程阻塞循环。

心跳请求体（JSON POST）： ticket_id、task_id、worker_id、status（online / offline）、hostname、pid、heartbeat_at（ISO8601，UTC+8）、可选 metadata。ticket_id 来自 NUTRICARE_TICKET_ID，task_id 来自执行记录生命周期。

ExecutionRecordStore：执行记录 CRUD

from nutricare_data_packages import ExecutionRecordStore

store = ExecutionRecordStore()

record = store.create(ticket_id="674a...")
# record["id"] 即任务 ID（task_id）

page = store.list_records(ticket_id="674a...", status="in_progress", skip=0, limit=20)
# 返回 {"pageNum", "pageSize", "total", "rows"}，与控制台 TaskView 分页字段一致

stats = store.count_statistics(ticket_id="674a...")

store.mark_resolved(record["id"], message="完成")
store.mark_failed(record["id"], message="超时")
store.append_message(record["id"], "cleanup done")
store.stop(record["id"], message="用户手动停止")  # 仅 in_progress → closed
store.delete_by_id(record["id"])

也可从子模块导入：from nutricare_data_packages.execution_record import ExecutionRecordStore。

ExecutionRecordLifecycle：启动建记录、退出改状态

采集 Worker 在设置好 MongoDB 与 NUTRICARE_TICKET_ID 后：

from nutricare_data_packages import ExecutionRecordLifecycle

lifecycle = ExecutionRecordLifecycle(register_shutdown_hooks=True)
lifecycle.start()
try:
    run_job()
    lifecycle.stop_success()
except KeyboardInterrupt:
    lifecycle.stop_interrupted(message="KeyboardInterrupt")
    raise
except Exception as exc:
    lifecycle.stop_failed(message=str(exc))
    raise

或使用上下文管理器（按异常类型自动收尾）：

from nutricare_data_packages import execution_record_lifecycle

with execution_record_lifecycle(register_shutdown_hooks=True) as lifecycle:
    run_job()

register_shutdown_hooks=True 时注册 atexit 与 SIGINT / SIGTERM，避免未显式 stop() 时记录长期停留在 in_progress。stop(..., skip_if_not_in_progress=True)（默认）时，若控制台已将记录标为终态，则不再覆盖。

关闭生命周期：NUTRICARE_EXECUTION_RECORD_ENABLED=false 或代码中 ExecutionRecordLifecycle(config=load_execution_record_lifecycle_config(enabled=False))。

元数据 doc_metadata

唯一业务键：(storage_subdir, relative_path)。建议索引：

db.doc_metadata.createIndex(
  { storage_subdir: 1, relative_path: 1 },
  { unique: true }
)

字段	说明
storage_subdir	存储子目录（与 StorageBase.storage_subdir 一致）
relative_path	逻辑相对路径（未分区）
object_key	OSS 上相对于 storage_subdir 的物理键（含 _p/... 分区）
title	展示标题
size / md5	内容与校验
file_type	小写扩展名，见下表
ticket_id	来自 NUTRICARE_TICKET_ID
task_id	来自执行记录 id（ExecutionRecordLifecycle.start()）
status	业务状态，写入时默认 success
metadata	扩展 JSON 对象
created_at / updated_at	创建 / 更新时间

支持的 file_type：pdf、doc、docx、md、txt、xls、xlsx、ppt、pptx；未显式传入时从路径扩展名推断，不在列表中则为 null。

执行记录 execution_records

建议索引：(ticket_id, status, started_at)。

字段	说明
id	任务 ID（MongoDB _id 字符串，创建记录时生成）
ticket_id	工单 ID
status	见下表
started_at / finished_at	开始 / 结束时间
message	执行日志或备注（字符串）
created_at / updated_at	创建 / 更新时间

status	含义
in_progress	运行中
resolved	已完成（生命周期默认成功态）
closed	已关闭（中断或手动停止）
failed	失败

典型 Worker 集成

采集进程启动时建议同时开启执行记录与心跳（按环境变量控制开关）：

import os

from nutricare_data_packages import (
    HeartbeatDaemon,
    execution_record_lifecycle,
    StorageBase,
)

os.environ.setdefault("NUTRICARE_TICKET_ID", "...")
# MongoDB、OSS、NUTRICARE_HEARTBEAT_URL 等由部署环境注入


class ArtifactStorage(StorageBase):
    @property
    def storage_subdir(self) -> str:
        return "artifacts"


def main() -> None:
    heartbeat = HeartbeatDaemon()
    heartbeat.start()
    try:
        with execution_record_lifecycle(register_shutdown_hooks=True):
            storage = ArtifactStorage()
            storage.write("output/result.json", b"{}", title="结果")
            # ... 采集逻辑 ...
    finally:
        heartbeat.stop()


if __name__ == "__main__":
    main()

本地开发与测试

# 安装可编辑包与开发依赖
pip install -e ".[oss,dev]"

# 运行示例脚本（需配置真实 MongoDB / OSS 环境变量）
python test/run_test.py

# 单元测试（生命周期配置等，不依赖外部服务）
pytest test/

包内主要导出见 nutricare_data_packages.__all__；子包 execution_record、heartbeat、oss 亦可按需单独引用。

构建与发布

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