Metadata-Version: 2.4
Name: nutricare-data-packages
Version: 0.2.10
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
Requires-Dist: alibabacloud-oss-v2>=1.2.0
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 KEY_OBJECT_KEY, OssStorage, 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_type="报告",
)

# 查询元数据（原始 MongoDB 文档）
doc = storage.metadata_collection.find_one(
    {"storage_subdir": "reports", "relative_path": relative_path}
)

# 获取私有 OSS 预签名 URL（省略 expires 时 SDK 默认 15 分钟）
oss_key = f"{storage.storage_subdir}/{doc[KEY_OBJECT_KEY]}"
url = OssStorage().presigned_url(oss_key, expires=datetime.timedelta(hours=1))
```

**`write` 行为说明：**

| 项 | 说明 |
|----|------|
| `relative_path` | 相对于 `storage_subdir` 的逻辑路径；禁止 `..`、绝对路径及非法字符 |
| 内容来源 | `data`（字节）与 `file_path`（本地文件）二选一；均省略时按空内容处理 |
| 可选参数 | `title`、`doc_type`、`journal`、`author`、`publication_date`、`citation_count`、`source_url`、`authority_mark`、`impact_factor`、`file_type`、`status`（默认 `success`）、`metadata`、`content_type` |
| 幂等 | 相同 MD5 且 OSS 对象已存在时跳过上传，但仍会刷新元数据（如绑定当前 `task_id`），返回 `True` |
| 分区路径 | 新文件在 OSS 上使用 `_p/xx/yy/...` 分区以降低单目录对象数；已存在的未分区旧路径仍兼容 |
| `object_key` | 元数据中保存分区后的相对键；预签名 URL 需拼接 `{storage_subdir}/{object_key}` |

文献类元数据写入示例：

```python
storage.write(
    "papers/2024/review.pdf",
    pdf_bytes,
    title="某营养指南",
    doc_type="指南",
    journal="中华临床营养杂志",
    author="张三; 李四",
    publication_date="2024-06-01",
    citation_count=128,
    source_url="https://example.com/article/123",
    authority_mark=["CSTPCD", "SCI"],
    impact_factor=5.2,
    file_type="pdf",
)
```

从本地文件上传示例：

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

store = MetadataStore()
store.save(
    storage_subdir="reports",
    relative_path="daily/report.txt",
    title="日报",
    doc_type="报告",
    journal="Nutricare 内刊",
    author="张三",
    publication_date="2024-06-01",
    citation_count=0,
    source_url="https://example.com/report/1",
    authority_mark=["CSTPCD"],
    impact_factor=3.5,
    object_key="_p/ab/cd/daily/report.txt",
    size=5,
    md5="...",
    file_type="txt",
    status="success",
)

# 按路径查询（原始文档 / 序列化后）
raw = store.find_by_path("reports", "daily/report.txt")
item = store.get_one("reports", "daily/report.txt")  # 含 id 字段

# upsert 返回序列化文档；list_metadata 支持 ticket_id / task_id / status / storage_subdir 过滤
total, rows = store.list_metadata(ticket_id="ticket-001", skip=0, limit=20)
stats = store.count_by_ticket(storage_subdir="reports")
```

`ticket_id` 从环境变量读取，`task_id` 从执行记录生命周期上下文读取，无需每次 `save` 传入。`MetadataStore(..., auto_bind_ticket_task=False)` 可关闭自动绑定。序列化字段见 [元数据 `doc_metadata`](#元数据-doc_metadata)。

### 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 }
)
```

| 字段 | 说明 |
|------|------|
| `id` | MongoDB `_id` 字符串（`serialize_metadata` / `get_one` / `upsert` 返回） |
| `storage_subdir` | 存储子目录（与 `StorageBase.storage_subdir` 一致） |
| `relative_path` | 逻辑相对路径（未分区） |
| `object_key` | OSS 上相对于 `storage_subdir` 的物理键（含 `_p/...` 分区） |
| `title` | 展示标题 |
| `doc_type` | 文档类型（如指南、综述）；与 `file_type`（文件扩展名）不同 |
| `journal` | 所属期刊 |
| `author` | 作者 |
| `publication_date` | 出版日期（字符串，如 `2024-06-01` 或 `2024-06`） |
| `citation_count` | 被引用次数（整数） |
| `source_url` | 原文链接 |
| `authority_mark` | 权威标记（字符串数组，如 `["CSTPCD", "SCI"]`） |
| `impact_factor` | IF 值（影响因子，浮点数） |
| `size` / `md5` | 内容与校验 |
| `file_type` | 小写扩展名，见下表 |
| `ticket_id` | 来自 `NUTRICARE_TICKET_ID` |
| `task_id` | 来自执行记录 `id`（`ExecutionRecordLifecycle.start()`） |
| `status` | 业务状态，写入时默认 `success` |
| `metadata` | 扩展 JSON 对象 |
| `created_at` / `updated_at` | 创建 / 更新时间（UTC+8，ISO8601） |

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

字段常量可从 `nutricare_data_packages.doc_metadata` 导入，例如 `KEY_AUTHORITY_MARK`、`KEY_IMPACT_FACTOR`。

### 执行记录 `execution_records`

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

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

| `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
# 安装可编辑包与开发依赖（OSS SDK 已包含在主依赖中）
pip install -e ".[dev]"

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

包内主要导出见 `nutricare_data_packages.__all__`；子包 `doc_metadata`、`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/*
```

