625275e55b
Add the Healthcare platform backend (phases 13.0–13.7) with Alembic migration revision fix, production deploy script, validation reports, shared UI exports, and loyalty list page client directives so Next.js build succeeds. Co-authored-by: Cursor <cursoragent@cursor.com>
293 lines
11 KiB
Python
293 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate healthcare phases 13.2-13.7 implementation files."""
|
|
from __future__ import annotations
|
|
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FILES: dict[str, str] = {}
|
|
|
|
|
|
def add(path: str, content: str) -> None:
|
|
FILES[path] = textwrap.dedent(content).lstrip() + "\n"
|
|
|
|
|
|
# =============================================================================
|
|
# Phase 13.2 — Doctor Panel
|
|
# =============================================================================
|
|
|
|
add(
|
|
"app/schemas/doctor_panel.py",
|
|
'''\
|
|
"""Doctor panel DTOs — Phase 13.2."""
|
|
from __future__ import annotations
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from app.models.types import VisitSessionStatus
|
|
from app.schemas.common import ORMBase
|
|
from app.schemas.appointments import AppointmentRead
|
|
|
|
_FORBID = ConfigDict(extra="forbid")
|
|
|
|
class VisitSessionCreate(BaseModel):
|
|
model_config = _FORBID
|
|
clinic_id: UUID
|
|
doctor_id: UUID
|
|
patient_id: UUID
|
|
appointment_id: UUID
|
|
code: str = Field(max_length=50)
|
|
queue_position: int = Field(default=100, ge=1, le=10000)
|
|
metadata_json: dict | None = None
|
|
|
|
class VisitSessionActionRequest(BaseModel):
|
|
model_config = _FORBID
|
|
version: int = Field(ge=1)
|
|
reason: str | None = Field(default=None, max_length=500)
|
|
|
|
class VisitSessionRead(ORMBase):
|
|
id: UUID
|
|
tenant_id: UUID
|
|
clinic_id: UUID
|
|
doctor_id: UUID
|
|
patient_id: UUID
|
|
appointment_id: UUID
|
|
code: str
|
|
status: VisitSessionStatus
|
|
queue_position: int
|
|
started_at: datetime | None
|
|
finished_at: datetime | None
|
|
metadata_json: dict | None
|
|
version: int
|
|
is_deleted: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class VisitNoteCreate(BaseModel):
|
|
model_config = _FORBID
|
|
visit_session_id: UUID
|
|
doctor_id: UUID
|
|
code: str = Field(max_length=50)
|
|
title: str = Field(max_length=255)
|
|
body: str | None = None
|
|
storage_file_ref: str | None = Field(default=None, max_length=255)
|
|
metadata_json: dict | None = None
|
|
|
|
class VisitNoteUpdate(BaseModel):
|
|
model_config = _FORBID
|
|
title: str | None = Field(default=None, max_length=255)
|
|
body: str | None = None
|
|
storage_file_ref: str | None = Field(default=None, max_length=255)
|
|
version: int = Field(ge=1)
|
|
metadata_json: dict | None = None
|
|
|
|
class VisitNoteRead(ORMBase):
|
|
id: UUID
|
|
tenant_id: UUID
|
|
visit_session_id: UUID
|
|
doctor_id: UUID
|
|
code: str
|
|
title: str
|
|
body: str | None
|
|
storage_file_ref: str | None
|
|
metadata_json: dict | None
|
|
version: int
|
|
is_deleted: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class DoctorDashboardRead(BaseModel):
|
|
doctor_id: UUID
|
|
queued_count: int
|
|
in_progress_count: int
|
|
completed_today_count: int
|
|
upcoming_appointments: list[AppointmentRead]
|
|
|
|
class DoctorQueueRead(BaseModel):
|
|
doctor_id: UUID
|
|
items: list[VisitSessionRead]
|
|
''',
|
|
)
|
|
|
|
add(
|
|
"app/validators/doctor_panel.py",
|
|
'''\
|
|
"""Doctor panel validators — Phase 13.2."""
|
|
from __future__ import annotations
|
|
from shared.exceptions import AppError
|
|
from app.models.types import VisitSessionAction, VisitSessionStatus
|
|
|
|
ALLOWED: dict[VisitSessionAction, set[VisitSessionStatus]] = {
|
|
VisitSessionAction.CALL_NEXT: {VisitSessionStatus.QUEUED},
|
|
VisitSessionAction.START: {VisitSessionStatus.QUEUED},
|
|
VisitSessionAction.FINISH: {VisitSessionStatus.IN_PROGRESS},
|
|
VisitSessionAction.DEFER: {VisitSessionStatus.QUEUED, VisitSessionStatus.IN_PROGRESS},
|
|
}
|
|
TARGET: dict[VisitSessionAction, VisitSessionStatus] = {
|
|
VisitSessionAction.CALL_NEXT: VisitSessionStatus.QUEUED,
|
|
VisitSessionAction.START: VisitSessionStatus.IN_PROGRESS,
|
|
VisitSessionAction.FINISH: VisitSessionStatus.COMPLETED,
|
|
VisitSessionAction.DEFER: VisitSessionStatus.DEFERRED,
|
|
}
|
|
TERMINAL = frozenset({VisitSessionStatus.COMPLETED, VisitSessionStatus.CANCELLED})
|
|
|
|
def ensure_visit_session_transition(*, action: VisitSessionAction, current: VisitSessionStatus) -> None:
|
|
if current in TERMINAL:
|
|
raise AppError("جلسه ویزیت در وضعیت پایانی است", status_code=409, error_code="visit_session_terminal")
|
|
allowed = ALLOWED.get(action, set())
|
|
if current not in allowed:
|
|
raise AppError(
|
|
"انتقال وضعیت جلسه ویزیت مجاز نیست",
|
|
status_code=409,
|
|
error_code="invalid_visit_session_transition",
|
|
details={"from_status": current.value, "action": action.value},
|
|
)
|
|
|
|
def target_status_for(action: VisitSessionAction) -> VisitSessionStatus:
|
|
return TARGET[action]
|
|
''',
|
|
)
|
|
|
|
add(
|
|
"app/policies/doctor_panel.py",
|
|
'''\
|
|
"""Visit session lifecycle policies — Phase 13.2."""
|
|
from __future__ import annotations
|
|
from app.models.types import VisitSessionAction, VisitSessionStatus
|
|
from app.validators.doctor_panel import ensure_visit_session_transition, target_status_for
|
|
|
|
class VisitSessionPolicy:
|
|
def ensure_transition(self, *, action: VisitSessionAction, current: VisitSessionStatus) -> VisitSessionStatus:
|
|
ensure_visit_session_transition(action=action, current=current)
|
|
return target_status_for(action)
|
|
''',
|
|
)
|
|
|
|
add(
|
|
"app/specifications/doctor_panel.py",
|
|
'''\
|
|
"""Doctor panel list specifications — Phase 13.2."""
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
from app.models.types import VisitSessionStatus
|
|
|
|
ALLOWED_SORT_FIELDS = frozenset({"queue_position", "created_at", "started_at"})
|
|
|
|
@dataclass
|
|
class VisitSessionListSpec:
|
|
doctor_id: UUID | None = None
|
|
clinic_id: UUID | None = None
|
|
status: VisitSessionStatus | None = None
|
|
sort_by: str = "queue_position"
|
|
sort_dir: str = "asc"
|
|
''',
|
|
)
|
|
|
|
add(
|
|
"app/repositories/doctor_panel.py",
|
|
'''\
|
|
"""Doctor panel repositories — Phase 13.2."""
|
|
from __future__ import annotations
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID
|
|
from sqlalchemy import and_, func, select
|
|
from app.models.doctor_panel import VisitNote, VisitSession
|
|
from app.models.types import VisitSessionStatus
|
|
from app.repositories.base import TenantBaseRepository
|
|
from app.specifications.doctor_panel import VisitSessionListSpec
|
|
|
|
class VisitSessionRepository(TenantBaseRepository[VisitSession]):
|
|
model = VisitSession
|
|
|
|
async def get_by_code(self, tenant_id: UUID, code: str) -> VisitSession | None:
|
|
stmt = select(VisitSession).where(
|
|
VisitSession.tenant_id == tenant_id,
|
|
VisitSession.code == code,
|
|
VisitSession.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_appointment(self, tenant_id: UUID, appointment_id: UUID) -> VisitSession | None:
|
|
stmt = select(VisitSession).where(
|
|
VisitSession.tenant_id == tenant_id,
|
|
VisitSession.appointment_id == appointment_id,
|
|
VisitSession.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
def _filter_clauses(self, tenant_id: UUID, spec: VisitSessionListSpec | None):
|
|
clauses = [VisitSession.tenant_id == tenant_id, VisitSession.is_deleted.is_(False)]
|
|
if spec is None:
|
|
return clauses
|
|
if spec.doctor_id is not None:
|
|
clauses.append(VisitSession.doctor_id == spec.doctor_id)
|
|
if spec.clinic_id is not None:
|
|
clauses.append(VisitSession.clinic_id == spec.clinic_id)
|
|
if spec.status is not None:
|
|
clauses.append(VisitSession.status == spec.status)
|
|
return clauses
|
|
|
|
async def list_filtered(self, tenant_id: UUID, *, offset: int = 0, limit: int = 20, spec: VisitSessionListSpec | None = None):
|
|
clauses = self._filter_clauses(tenant_id, spec)
|
|
sort_col = getattr(VisitSession, spec.sort_by if spec else "queue_position", VisitSession.queue_position)
|
|
order = sort_col.asc() if spec and spec.sort_dir == "asc" else sort_col.desc()
|
|
stmt = select(VisitSession).where(and_(*clauses)).order_by(order).offset(offset).limit(limit)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
async def count_by_status(self, tenant_id: UUID, *, doctor_id: UUID, status: VisitSessionStatus) -> int:
|
|
stmt = select(func.count()).select_from(VisitSession).where(
|
|
VisitSession.tenant_id == tenant_id,
|
|
VisitSession.doctor_id == doctor_id,
|
|
VisitSession.status == status,
|
|
VisitSession.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return int(result.scalar_one())
|
|
|
|
async def count_completed_today(self, tenant_id: UUID, *, doctor_id: UUID) -> int:
|
|
today = datetime.now(timezone.utc).date()
|
|
stmt = select(func.count()).select_from(VisitSession).where(
|
|
VisitSession.tenant_id == tenant_id,
|
|
VisitSession.doctor_id == doctor_id,
|
|
VisitSession.status == VisitSessionStatus.COMPLETED,
|
|
VisitSession.is_deleted.is_(False),
|
|
func.date(VisitSession.finished_at) == today,
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return int(result.scalar_one())
|
|
|
|
class VisitNoteRepository(TenantBaseRepository[VisitNote]):
|
|
model = VisitNote
|
|
|
|
async def get_by_code(self, tenant_id: UUID, code: str) -> VisitNote | None:
|
|
stmt = select(VisitNote).where(
|
|
VisitNote.tenant_id == tenant_id,
|
|
VisitNote.code == code,
|
|
VisitNote.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_session(self, tenant_id: UUID, visit_session_id: UUID):
|
|
stmt = select(VisitNote).where(
|
|
VisitNote.tenant_id == tenant_id,
|
|
VisitNote.visit_session_id == visit_session_id,
|
|
VisitNote.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
''',
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
for rel, content in FILES.items():
|
|
path = ROOT / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
print(f"wrote {rel}")
|