样本入库信息短信通知 — 实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 克隆样本接收信息的完整链路（前端页面 + 后端 API + 独立通知表），实现样本入库信息的短信通知管理。

Architecture: 新建独立表 biz_storage_entry_sms_notification，关联 biz_storage_release_review；后端 CRUD 克隆 sample_sms_notification.py 的联查+懒创建+短信生成模式；前端克隆 sample-reception-info/index.vue 页面。

Tech Stack: FastAPI + SQLAlchemy + Pydantic (后端) / Vue 3 + Element Plus + TypeScript (前端)

Design Spec: docs/superpowers/specs/2025-05-23-storage-entry-sms-notification-design.md

File Structure

新建文件

文件	职责
`docs/sql/biz_storage_entry_sms_notification.sql`	建表 DDL
`backend/app/crud/storage_entry_sms_notification.py`	CRUD 层：联查、懒创建、短信生成
`backend/app/api/storage_entry_sms_notification.py`	API 路由：3 个端点
`web-admin/src/api/storage-entry-sms-notification.ts`	前端 API 封装
`web-admin/src/views/cell-storage/storage-entry-info/index.vue`	前端页面

修改文件

文件	修改内容
`backend/app/models/models.py`	追加 `BizStorageEntrySmsNotification` 模型类
`backend/app/api/__init__.py`	导入并注册新路由
`backend/app/main.py`	include_router 挂载新路由
`web-admin/src/router/modules/sqxp.ts`	CellStorage 下添加路由条目

Task 1: 建表 DDL

Files:

Create: docs/sql/biz_storage_entry_sms_notification.sql
[ ] Step 1: 创建 SQL 文件

-- 样本入库短信通知记录表
CREATE TABLE biz_storage_entry_sms_notification (
  id                      VARCHAR(64)  NOT NULL COMMENT 'UUID',
  storage_review_id       VARCHAR(64)  NOT NULL COMMENT '关联 biz_storage_release_review.id',
  protocol_id             VARCHAR(64)  DEFAULT NULL COMMENT '关联采集协议单ID（冗余）',
  sms_content             VARCHAR(500) DEFAULT NULL COMMENT '生成的短信内容',
  sms_notified            TINYINT(1)   DEFAULT 0 COMMENT '0=未通知 1=已通知',
  notified_by             VARCHAR(64)  DEFAULT NULL COMMENT '通知操作人ID',
  notified_time           DATETIME     DEFAULT NULL COMMENT '通知时间',
  -- 公共字段
  tenant_no               VARCHAR(32)  DEFAULT NULL,
  order_num               INT          DEFAULT 0,
  remark                  VARCHAR(500) DEFAULT NULL,
  create_by               VARCHAR(64)  DEFAULT NULL,
  create_dept_id          VARCHAR(64)  DEFAULT NULL,
  create_time             DATETIME     DEFAULT CURRENT_TIMESTAMP,
  update_by               VARCHAR(64)  DEFAULT NULL,
  update_time             DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  del_flag                CHAR(1)      DEFAULT '0',
  PRIMARY KEY (id),
  KEY idx_storage_review_id (storage_review_id),
  KEY idx_protocol_id (protocol_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='样本入库短信通知记录';

Step 2: 在数据库执行建表

Run: source venv/bin/activate && python -c "from app.core.database import engine; from sqlalchemy import text; import os; sql=open('docs/sql/biz_storage_entry_sms_notification.sql').read(); conn=engine.connect(); conn.execute(text(sql)); conn.commit(); print('OK')"

或直接在 MySQL 客户端执行该 SQL 文件。

Task 2: ORM 模型

Files:

Modify: backend/app/models/models.py — 在 BizSampleSmsNotification 类之后追加
[ ] Step 1: 追加模型类

在 BizSampleSmsNotification 类（约第 1117 行 del_flag 之后）追加：



class BizStorageEntrySmsNotification(Base):
    """样本入库短信通知 biz_storage_entry_sms_notification"""

    __tablename__ = "biz_storage_entry_sms_notification"

    id = Column(String(64), primary_key=True)
    storage_review_id = Column(String(64), nullable=False, comment="关联 biz_storage_release_review.id")
    protocol_id = Column(String(64), comment="关联采集协议单ID")
    sms_content = Column(String(500), comment="生成的短信内容")
    sms_notified = Column(SmallInteger, default=0, comment="0=未通知 1=已通知")
    notified_by = Column(String(64), comment="通知操作人ID")
    notified_time = Column(DateTime, comment="通知时间")
    tenant_no = Column(String(32))
    order_num = Column(Integer, default=0)
    remark = Column(String(500))
    create_by = Column(String(64))
    create_dept_id = Column(String(64))
    create_time = Column(DateTime, default=datetime.now)
    update_by = Column(String(64))
    update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    del_flag = Column(String(1), default="0")

Step 2: 验证 import 无报错

Run: cd backend && source venv/bin/activate && python -c "from app.models.models import BizStorageEntrySmsNotification; print('Model OK:', BizStorageEntrySmsNotification.__tablename__)"

Expected: Model OK: biz_storage_entry_sms_notification

Task 3: 后端 CRUD 层

Files:

Create: backend/app/crud/storage_entry_sms_notification.py

参考文件: backend/app/crud/sample_sms_notification.py（完整克隆并调整）

Step 1: 创建 CRUD 文件

关键差异点（相对 sample_sms_notification.py）：

主表从 BizSampleReception → BizStorageReleaseReview
通知表从 BizSampleSmsNotification → BizStorageEntrySmsNotification
外键字段从 sample_reception_id → storage_review_id
过滤条件：reception_source='cell_storage'，(storage_status='yrk' OR review_conclusion='btg')
排序：BizStorageReleaseReview.create_time.desc()
短信模板：入库/拒绝入库两种文案
_row_to_dict 字段调整：storageReviewId、storageStatus、storageStatusLabel、storageTime

"""样本入库短信通知 CRUD"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizStorageEntrySmsNotification,
    BizStorageReleaseReview,
)

_UNICODE_CI = "utf8mb4_unicode_ci"


def _unicode_ci(col):
    return col.collate(_UNICODE_CI)


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _storage_status_label(row: BizStorageReleaseReview) -> str:
    """入库状态标签：已入库 or 拒绝入库"""
    if row.storage_status == "yrk":
        return "已入库"
    if row.review_conclusion == "btg":
        return "拒绝入库"
    return "待入库"


def _row_to_dict(
    *,
    notification_id: str = "",
    storage_review_id: str = "",
    protocol_id: str = "",
    sms_notified: int = 0,
    sms_content: str = "",
    protocol_no: str = "",
    mother_name: str = "",
    newborn_name: str = "",
    sample_code: str = "",
    sample_type_name: str = "",
    storage_time: str = "",
    storage_status_label: str = "",
    collection_protocol_id: str = "",
) -> Dict[str, Any]:
    return {
        "id": notification_id,
        "storageReviewId": storage_review_id,
        "protocolId": protocol_id,
        "protocolNo": protocol_no,
        "motherName": mother_name,
        "newbornName": newborn_name,
        "sampleCode": sample_code,
        "sampleTypeName": sample_type_name,
        "storageTime": storage_time,
        "storageStatusLabel": storage_status_label,
        "smsNotified": sms_notified,
        "smsContent": sms_content,
        "collectionProtocolId": collection_protocol_id,
    }


def _base_query(db: Session, tenant_no: str):
    return (
        db.query(
            BizStorageReleaseReview,
            BizStorageEntrySmsNotification,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
            BizProtocolSample,
        )
        .outerjoin(
            BizStorageEntrySmsNotification,
            and_(
                _unicode_ci(BizStorageEntrySmsNotification.storage_review_id)
                == _unicode_ci(BizStorageReleaseReview.id),
                _unicode_ci(BizStorageEntrySmsNotification.tenant_no) == tenant_no,
                BizStorageEntrySmsNotification.del_flag == "0",
            ),
        )
        .outerjoin(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == BizStorageReleaseReview.protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolSample,
            and_(
                _unicode_ci(BizProtocolSample.sample_code)
                == _unicode_ci(BizStorageReleaseReview.sample_code),
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            ),
        )
        .filter(
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            BizStorageReleaseReview.reception_source == "cell_storage",
            or_(
                BizStorageReleaseReview.storage_status == "yrk",
                BizStorageReleaseReview.review_conclusion == "btg",
            ),
        )
    )


def _ensure_notification(
    db: Session,
    review: BizStorageReleaseReview,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> BizStorageEntrySmsNotification:
    existing = (
        db.query(BizStorageEntrySmsNotification)
        .filter(
            BizStorageEntrySmsNotification.storage_review_id == review.id,
            BizStorageEntrySmsNotification.tenant_no == tenant_no,
            BizStorageEntrySmsNotification.del_flag == "0",
        )
        .first()
    )
    if existing:
        return existing

    obj = BizStorageEntrySmsNotification(
        id=generate_uuid(),
        storage_review_id=review.id,
        protocol_id=review.protocol_id or "",
        sms_notified=0,
        tenant_no=tenant_no,
        create_by=user_id,
        update_by=user_id,
    )
    db.add(obj)
    db.flush()
    db.commit()
    db.refresh(obj)
    return obj


def resolve_notification(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    storage_review_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[BizStorageEntrySmsNotification]:
    """Find notification by id; fallback to storage_review_id and ensure row exists."""
    notif = (
        db.query(BizStorageEntrySmsNotification)
        .filter(
            BizStorageEntrySmsNotification.id == notification_id,
            BizStorageEntrySmsNotification.tenant_no == tenant_no,
            BizStorageEntrySmsNotification.del_flag == "0",
        )
        .first()
    )
    if notif:
        return notif

    if not storage_review_id:
        return None

    review = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == storage_review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not review:
        return None
    return _ensure_notification(db, review, tenant_no=tenant_no, user_id=user_id)


def _fetch_sample_type_name(db: Session, sample_type_id: Optional[str], tenant_no: str) -> str:
    if not sample_type_id:
        return ""
    t = (
        db.query(BizCellSampleType)
        .filter(
            BizCellSampleType.id == sample_type_id,
            BizCellSampleType.del_flag == 0,
        )
        .first()
    )
    return (t.type_name or "") if t else ""


def list_notifications(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    keyword: Optional[str] = None,
    storage_status: Optional[str] = None,
    sms_notified: Optional[int] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0

    q = _base_query(db, tenant_no)

    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        q = q.filter(
            or_(
                BizCollectionProtocol.protocol_no.like(kw),
                BizProtocolPartyCustomer.mother_name.like(kw),
                BizStorageReleaseReview.sample_code.like(kw),
            )
        )

    if storage_status:
        if storage_status == "yrk":
            q = q.filter(BizStorageReleaseReview.storage_status == "yrk")
        elif storage_status == "btg":
            q = q.filter(BizStorageReleaseReview.review_conclusion == "btg")

    if sms_notified is not None:
        if sms_notified == 0:
            q = q.filter(
                or_(
                    BizStorageEntrySmsNotification.sms_notified == 0,
                    BizStorageEntrySmsNotification.id.is_(None),
                )
            )
        else:
            q = q.filter(BizStorageEntrySmsNotification.sms_notified == sms_notified)

    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizStorageReleaseReview, user_info, db, tenant_no)

    total = q.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)

    rows = (
        q.order_by(BizStorageReleaseReview.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    sample_type_ids = list({
        ps.sample_type_id
        for _, _, _, _, ps in rows
        if ps and ps.sample_type_id
    })
    type_map: Dict[str, str] = {}
    if sample_type_ids:
        for t in (
            db.query(BizCellSampleType)
            .filter(BizCellSampleType.id.in_(sample_type_ids), BizCellSampleType.del_flag == 0)
            .all()
        ):
            type_map[t.id] = t.type_name or ""

    items: List[Dict[str, Any]] = []
    for review, notif, proto, party, psample in rows:
        if notif is None:
            notif = _ensure_notification(db, review, tenant_no=tenant_no)

        sample_type_name = ""
        if psample and psample.sample_type_id:
            sample_type_name = type_map.get(psample.sample_type_id, "")

        # 入库时间：已入库取 storage_time，拒绝入库取 review_time
        display_time = ""
        if review.storage_status == "yrk" and review.storage_time:
            display_time = _fmt_dt(review.storage_time)
        elif review.review_time:
            display_time = _fmt_dt(review.review_time)

        items.append(
            _row_to_dict(
                notification_id=notif.id,
                storage_review_id=review.id,
                protocol_id=notif.protocol_id or "",
                sms_notified=notif.sms_notified or 0,
                sms_content=notif.sms_content or "",
                protocol_no=(proto.protocol_no or "") if proto else (review.protocol_no or ""),
                mother_name=(party.mother_name or "") if party else "",
                newborn_name=(party.newborn_name or "") if party else "",
                sample_code=review.sample_code or "",
                sample_type_name=sample_type_name,
                storage_time=display_time,
                storage_status_label=_storage_status_label(review),
                collection_protocol_id=(proto.id or "") if proto else (review.protocol_id or ""),
            )
        )
    return items, int(total)


def toggle_notification_status(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    sms_notified: int,
    user_id: Optional[str] = None,
    storage_review_id: Optional[str] = None,
) -> Optional[BizStorageEntrySmsNotification]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        storage_review_id=storage_review_id,
        user_id=user_id,
    )
    if not notif:
        return None

    notif.sms_notified = sms_notified
    notif.notified_by = user_id if sms_notified == 1 else None
    notif.notified_time = datetime.now() if sms_notified == 1 else None
    notif.update_by = user_id
    notif.update_time = datetime.now()
    db.commit()
    db.refresh(notif)
    return notif


def generate_sms_content(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    storage_review_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[str]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        storage_review_id=storage_review_id,
        user_id=user_id,
    )
    if not notif:
        return None

    review = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == notif.storage_review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not review:
        return None

    mother_name = ""
    proto = (
        db.query(BizCollectionProtocol)
        .filter(
            BizCollectionProtocol.id == review.protocol_id,
            BizCollectionProtocol.del_flag == 0,
        )
        .first()
    )
    if proto:
        party = (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id == proto.id,
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .first()
        )
        if party:
            mother_name = party.mother_name or ""

    sample_code = review.sample_code or ""

    # 短信模板：区分已入库和拒绝入库
    if review.storage_status == "yrk":
        storage_date = _fmt_dt(review.storage_time)[:10] if review.storage_time else ""
        content = (
            f"尊敬的{mother_name}女士，您的样本（编码：{sample_code}）"
            f"已于{storage_date}完成入库处理。如有疑问请联系客服。"
        )
    else:
        review_date = _fmt_dt(review.review_time)[:10] if review.review_time else ""
        content = (
            f"尊敬的{mother_name}女士，您的样本（编码：{sample_code}）"
            f"入库审核未通过。如有疑问请联系客服。"
        )

    notif.sms_content = content
    notif.update_time = datetime.now()
    db.commit()
    db.refresh(notif)
    return content

Step 2: 验证 import 无报错

Run: cd backend && source venv/bin/activate && python -c "from app.crud.storage_entry_sms_notification import list_notifications; print('CRUD OK')"

Expected: CRUD OK

Task 4: 后端 API 路由

Files:

Create: backend/app/api/storage_entry_sms_notification.py

参考文件: backend/app/api/sample_sms_notification.py

Step 1: 创建 API 路由文件

关键差异点（相对 sample_sms_notification.py）：

菜单码：cell-storage:storage-entry-info:list
CRUD 函数：参数名 storage_review_id、storage_status
URL 前缀：/storage-entry-sms-notifications

"""样本入库短信通知 API"""
from typing import Optional

from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.menu_guard import require_menu_code
from app.core.response import error_response, success_response
from app.crud import storage_entry_sms_notification as svc
from app.crud.collection_protocol import total_pages
from app.models.models import SysUser

router = APIRouter(prefix="/web/admin", tags=["样本入库短信通知"])


def _get_tenant_no(db: Session, user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失，无法确定租户")
    sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
    if not sys_user or not sys_user.tenant_no:
        raise PermissionError("用户租户信息缺失")
    return sys_user.tenant_no


def _get_current_user_id(user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失")
    return str(user_id)


def _page_payload(items: list, total: int, page: int, page_size: int) -> dict:
    return {
        "list": items,
        "total": total,
        "page": page,
        "pageSize": page_size,
        "totalPages": total_pages(total, page_size),
    }


class ToggleStatusBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    sms_notified: int = Field(..., alias="smsNotified")
    storage_review_id: Optional[str] = Field(None, alias="storageReviewId")


@router.get("/storage-entry-sms-notifications/list")
async def list_notifications(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    storageStatus: Optional[str] = None,
    smsNotified: Optional[int] = None,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("cell-storage:storage-entry-info:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = svc.list_notifications(
            db,
            page=page,
            page_size=pageSize,
            tenant_no=tenant_no,
            keyword=keyword,
            storage_status=storageStatus,
            sms_notified=smsNotified,
            user_info=user,
        )
        return success_response(_page_payload(items, total, page, pageSize))
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))


@router.put("/storage-entry-sms-notifications/{notification_id}/toggle-status")
async def toggle_status(
    notification_id: str,
    body: ToggleStatusBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("cell-storage:storage-entry-info:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        result = svc.toggle_notification_status(
            db,
            notification_id=notification_id,
            tenant_no=tenant_no,
            sms_notified=body.sms_notified,
            user_id=user_id,
            storage_review_id=body.storage_review_id,
        )
        if not result:
            return error_response(404, "通知记录不存在")
        return success_response(None)
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))


@router.post("/storage-entry-sms-notifications/{notification_id}/generate-sms")
async def generate_sms(
    notification_id: str,
    storageReviewId: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("cell-storage:storage-entry-info:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        content = svc.generate_sms_content(
            db,
            notification_id=notification_id,
            tenant_no=tenant_no,
            storage_review_id=storageReviewId,
            user_id=user_id,
        )
        if content is None:
            return error_response(404, "通知记录不存在")
        return success_response({"smsContent": content})
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))

Step 2: 验证 import 无报错

Run: cd backend && source venv/bin/activate && python -c "from app.api.storage_entry_sms_notification import router; print('API OK, routes:', len(router.routes))"

Expected: API OK, routes: 3

Task 5: 后端路由注册

Files:

Modify: backend/app/api/__init__.py
Modify: backend/app/main.py
[ ] Step 1: 在 __init__.py 追加 import 和 all 条目

在 sample_sms_notification_router 之后追加：

from app.api.storage_entry_sms_notification import router as storage_entry_sms_notification_router

在 __all__ 列表的 "sample_sms_notification_router" 之后追加：

"storage_entry_sms_notification_router",

Step 2: 在 main.py 追加 include_router

在 app.include_router(sample_sms_notification_router, prefix=settings.API_V1_PREFIX) 之后追加：

app.include_router(storage_entry_sms_notification_router, prefix=settings.API_V1_PREFIX)

Step 3: 启动后端验证路由挂载成功

Run: cd backend && source venv/bin/activate && python -c "from app.main import app; routes = [r.path for r in app.routes if hasattr(r, 'path') and 'storage-entry-sms' in r.path]; print('Found', len(routes), 'routes:', routes)"

Expected: Found 3 routes: ['/api/storage-entry-sms-notifications/list', ...]

Task 6: 前端 API 模块

Files:

Create: web-admin/src/api/storage-entry-sms-notification.ts
[ ] Step 1: 创建 API 文件

import request from '@/utils/http'

type PageList<T> = {
  list: T[]
  total: number
  page: number
  pageSize: number
  totalPages: number
}

/** 查询列表（联查 storage_release_review + sms_notification） */
export function fetchStorageEntrySmsList(params?: {
  page?: number
  pageSize?: number
  keyword?: string
  storageStatus?: string
  smsNotified?: string
}) {
  return request.get<PageList<Record<string, unknown>>>({
    url: '/web/admin/storage-entry-sms-notifications/list',
    params
  })
}

/** 切换通知状态 */
export function fetchToggleStorageEntrySmsNotified(
  id: string,
  data: { smsNotified: number; storageReviewId?: string }
) {
  return request.put({
    url: `/web/admin/storage-entry-sms-notifications/${id}/toggle-status`,
    data
  })
}

/** 生成短信内容 */
export function fetchGenerateStorageEntrySms(id: string, storageReviewId?: string) {
  return request.post<{ smsContent: string }>({
    url: `/web/admin/storage-entry-sms-notifications/${id}/generate-sms`,
    params: storageReviewId ? { storageReviewId } : undefined
  })
}

Task 7: 前端页面组件

Files:

Create: web-admin/src/views/cell-storage/storage-entry-info/index.vue

参考文件: web-admin/src/views/cell-storage/sample-reception-info/index.vue

Step 1: 创建页面组件

关键差异点（相对 sample-reception-info/index.vue）：

页面标题、描述 → 样本入库信息
面板标题 → 样本入库列表
筛选：acceptConclusion → storageStatus（已入库/拒绝入库）
API → storage-entry-sms-notification 的 3 个函数
表格列：接收时间→入库时间，接收结论→入库状态
入库状态 Tag：已入库=success，拒绝入库=danger
组件名 → CellStorageStorageEntryInfo

<!-- 样本入库信息 -->
<template>
  <div class="page-container">
    <header class="page-head">
      <h1>样本入库信息</h1>
      <p class="page-desc">查看样本入库信息，提取短信通知内容并管理通知状态。</p>
    </header>

    <section class="panel">
      <div class="panel-top">
        <h2 class="title-bar-heading">样本入库列表</h2>
      </div>

      <div class="filter-block">
        <el-form :model="searchForm" class="filter-form" @submit.prevent>
          <div class="filter-grid">
            <el-form-item label="关键词">
              <el-input v-model="searchForm.keyword" placeholder="协议号/姓名/样本编码" clearable />
            </el-form-item>
            <el-form-item label="入库状态">
              <el-select v-model="searchForm.storageStatus" placeholder="全部" clearable style="width: 100%">
                <el-option label="已入库" value="yrk" />
                <el-option label="拒绝入库" value="btg" />
              </el-select>
            </el-form-item>
            <el-form-item label="通知状态">
              <el-select v-model="searchForm.smsNotified" placeholder="全部" clearable style="width: 100%">
                <el-option label="已通知" value="1" />
                <el-option label="未通知" value="0" />
              </el-select>
            </el-form-item>
          </div>
          <div class="filter-actions-row">
            <div class="filter-actions-end">
              <el-button class="btn-ghost" @click="handleReset">
                <el-icon><RefreshLeft /></el-icon>重置
              </el-button>
              <el-button class="btn-main" type="primary" @click="handleSearch">
                <el-icon><Search /></el-icon>搜索
              </el-button>
            </div>
          </div>
        </el-form>
      </div>

      <div class="list-body list-wrap">
        <el-table
          class="list-table"
          :data="tableData"
          v-loading="loading"
          stripe
          row-key="id"
          @row-dblclick="handleView"
        >
          <el-table-column
            type="index"
            label="序号"
            min-width="68"
            align="center"
            header-align="center"
            :index="(i) => (pagination.page - 1) * pagination.pageSize + i + 1"
          />
          <el-table-column prop="protocolNo" label="协议号" min-width="140" align="center" header-align="center" />
          <el-table-column prop="motherName" label="母亲姓名" min-width="100" align="center" header-align="center" />
          <el-table-column prop="newbornName" label="新生儿姓名" min-width="110" align="center" header-align="center" show-overflow-tooltip />
          <el-table-column prop="sampleCode" label="样本编码" min-width="130" align="center" header-align="center" />
          <el-table-column prop="sampleTypeName" label="样本类型" min-width="110" align="center" header-align="center" />
          <el-table-column prop="storageTime" label="入库时间" min-width="170" align="center" header-align="center" />
          <el-table-column prop="storageStatusLabel" label="入库状态" min-width="100" align="center" header-align="center">
            <template #default="{ row }">
              <el-tag :type="row.storageStatusLabel === '已入库' ? 'success' : 'danger'" size="small">
                {{ row.storageStatusLabel }}
              </el-tag>
            </template>
          </el-table-column>
          <el-table-column prop="smsNotified" label="通知状态" min-width="100" align="center" header-align="center">
            <template #default="{ row }">
              <el-tag :type="row.smsNotified == 1 ? 'success' : 'info'" size="small">
                {{ row.smsNotified == 1 ? '已通知' : '未通知' }}
              </el-tag>
            </template>
          </el-table-column>
          <el-table-column label="操作" min-width="260" fixed="right" align="center" header-align="center">
            <template #default="{ row }">
              <el-button class="btn-op-item primary" @click="handleView(row)">
                <el-icon><View /></el-icon>查看
              </el-button>
              <el-button class="btn-op-item" @click="handleExtractSms(row)">
                <el-icon><ChatDotRound /></el-icon>提取短信
              </el-button>
              <el-button class="btn-op-item" @click="handleToggleNotified(row)">
                <el-icon><Bell /></el-icon>{{ row.smsNotified == 1 ? '标记未通知' : '标记已通知' }}
              </el-button>
            </template>
          </el-table-column>
        </el-table>
        <div class="pagination list-pager-bar">
          <el-pagination
            :current-page="pagination.page"
            :page-size="pagination.pageSize"
            :total="pagination.total"
            :page-sizes="[10, 20, 50]"
            layout="total, sizes, prev, pager, next"
            @size-change="(val) => ((pagination.page = 1), (pagination.pageSize = val), loadData())"
            @current-change="(val) => ((pagination.page = val), loadData())"
          />
        </div>
      </div>

      <ProtocolViewDialog
        v-model="viewVisible"
        :protocol-id="viewProtocolId"
        :list-row="viewRow"
        show-audit-history
        show-storage-cert
        show-fullmoon-insurance
      />
    </section>
  </div>
</template>

<script setup lang="ts">
  import { ref, reactive, onMounted } from 'vue'
  import { ElMessage } from 'element-plus'
  import { Search, RefreshLeft, View, ChatDotRound, Bell } from '@element-plus/icons-vue'
  import { useClipboard } from '@vueuse/core'
  import ProtocolViewDialog from '../protocol/ProtocolViewDialog.vue'
  import {
    fetchStorageEntrySmsList,
    fetchToggleStorageEntrySmsNotified,
    fetchGenerateStorageEntrySms
  } from '@/api/storage-entry-sms-notification'
  import { type CollectionProtocolRow } from '@/api/collection-protocol'

  defineOptions({ name: 'CellStorageStorageEntryInfo' })

  const searchForm = reactive({ keyword: '', storageStatus: '', smsNotified: '' })
  const tableData = ref<any[]>([])
  const loading = ref(false)
  const pagination = reactive({ page: 1, pageSize: 10, total: 0 })

  const viewVisible = ref(false)
  const viewProtocolId = ref<string | null>(null)
  const viewRow = ref<CollectionProtocolRow | null>(null)

  const { copy } = useClipboard()

  async function loadData() {
    loading.value = true
    try {
      const result = await fetchStorageEntrySmsList({
        page: pagination.page,
        pageSize: pagination.pageSize,
        keyword: searchForm.keyword || undefined,
        storageStatus: searchForm.storageStatus || undefined,
        smsNotified: searchForm.smsNotified || undefined
      })
      const res = (result as any) || {}
      tableData.value = res.list || []
      pagination.total = res.total || 0
    } finally {
      loading.value = false
    }
  }

  function handleSearch() {
    pagination.page = 1
    loadData()
  }

  function handleReset() {
    Object.assign(searchForm, { keyword: '', storageStatus: '', smsNotified: '' })
    handleSearch()
  }

  function handleView(row: Record<string, unknown>) {
    const pid = String(row.collectionProtocolId || row.protocolId || '')
    if (!pid) {
      ElMessage.warning('缺少关联协议单，无法查看')
      return
    }
    viewProtocolId.value = pid
    viewRow.value = {
      id: pid,
      protocolNo: row.protocolNo,
      motherName: row.motherName,
      newbornName: row.newbornName
    }
    viewVisible.value = true
  }

  async function handleExtractSms(row: any) {
    try {
      const result = await fetchGenerateStorageEntrySms(row.id, row.storageReviewId)
      const content = (result as any)?.smsContent || ''
      if (!content) {
        ElMessage.warning('短信内容为空')
        return
      }
      await copy(content)
      ElMessage.success('短信内容已复制到剪贴板')
    } catch {
      ElMessage.error('提取短信失败')
    }
  }

  async function handleToggleNotified(row: any) {
    const newVal = row.smsNotified == 1 ? 0 : 1
    try {
      await fetchToggleStorageEntrySmsNotified(row.id, {
        smsNotified: newVal,
        storageReviewId: row.storageReviewId
      })
      ElMessage.success(newVal === 1 ? '已标记为已通知' : '已标记为未通知')
      loadData()
    } catch {
      ElMessage.error('操作失败')
    }
  }

  onMounted(() => {
    loadData()
  })
</script>

Task 8: 前端路由注册

Files:

Modify: web-admin/src/router/modules/sqxp.ts
[ ] Step 1: 在 CellStorage 下的 sample-reception-info 条目之后追加

在 sample-reception-info 路由块（约第 260 行）之后，storage-cert 路由块之前，插入：

      {
        id: '50000000000000000000000000000012',
        name: 'CellStorageStorageEntryInfo',
        path: 'storage-entry-info',
        component: '/cell-storage/storage-entry-info/index',
        meta: { title: '样本入库信息', keepAlive: true, menuCode: 'cell-storage:storage-entry-info:list' }
      },

Step 2: 验证前端编译无报错

Run: cd web-admin && pnpm build 2>&1 | tail -5

Expected: 构建成功，无 TypeScript 错误。

Task 9: 端到端验证

Step 1: 启动后端，通过 Swagger 验证 API

Run: cd backend && source venv/bin/activate && ./start-local.sh

访问 http://localhost:8000/docs，检查：

GET /api/storage-entry-sms-notifications/list 存在
PUT /api/storage-entry-sms-notifications/{id}/toggle-status 存在
POST /api/storage-entry-sms-notifications/{id}/generate-sms 存在
[ ] Step 2: 启动前端，访问页面

Run: cd web-admin && pnpm dev

访问 http://localhost:3006/#/cell-storage/storage-entry-info，验证：

页面正常渲染，筛选区和表格列正确
列表数据加载正常（需后台有已入库记录）
双击行 / 点击「查看」打开 ProtocolViewDialog
「提取短信」生成内容并复制到剪贴板
「标记已/未通知」状态切换正常
浏览器窗口缩放时，各列等比伸缩（min-width 生效）
[ ] Step 3: 提交

git add docs/sql/biz_storage_entry_sms_notification.sql
git add backend/app/models/models.py
git add backend/app/crud/storage_entry_sms_notification.py
git add backend/app/api/storage_entry_sms_notification.py
git add backend/app/api/__init__.py
git add backend/app/main.py
git add web-admin/src/api/storage-entry-sms-notification.ts
git add web-admin/src/views/cell-storage/storage-entry-info/index.vue
git add web-admin/src/router/modules/sqxp.ts
git commit -m "feat: 新增样本入库信息页面，支持短信通知管理（克隆样本接收信息模式）"

2025-05-23-storage-entry-sms-notification.md 38 KB История Исходник

样本入库信息短信通知 — 实施计划

File Structure

新建文件

修改文件

Task 1: 建表 DDL

Task 2: ORM 模型

Task 3: 后端 CRUD 层

Task 4: 后端 API 路由

Task 5: 后端路由注册

Task 6: 前端 API 模块

Task 7: 前端页面组件

Task 8: 前端路由注册

Task 9: 端到端验证

2025-05-23-storage-entry-sms-notification.md 38 KB

История Исходник