Metadata-Version: 2.4
Name: nutricare-data-packages
Version: 0.2.1
Summary: Nutricare data access package
Author: Nutricare Team
Project-URL: Homepage, https://example.com
Project-URL: Repository, https://example.com/nutricare-data-packages
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: pymongo>=4.0.0
Provides-Extra: oss
Requires-Dist: alibabacloud-oss-v2>=1.2.0; extra == "oss"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"

# 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 集成](#典型-worker-集成)
- [本地开发与测试](#本地开发与测试)
- [构建与发布](#构建与发布)

## 安装

```bash
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：

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

从本地文件上传示例：

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

### OssStorage：直接使用 OSS

不经过元数据表、自行管理对象键时使用：

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

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

```python
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",      # 可选
        )
```

### 心跳守护进程

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

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

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

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

```python
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)`。建议索引：

```javascript
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 集成

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

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

## 本地开发与测试

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

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

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

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

## 构建与发布

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

