kushidou/YaoXiang
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
YaoXiang/docs/backend.md
T
tang1219 ead13f863c 首次提交by MimoCode
2026-06-15 14:50:15 +08:00

1440 lines
44 KiB
Markdown
Raw Permalink Blame History

# 后端开发文档
## 1. 项目结构
```
backend/
├── app/
│ ├── __init__.py
│ ├── main.py # 应用入口
│ ├── config.py # 配置管理
│ ├── database.py # 数据库连接
│ ├── api/ # 路由层
│ │ ├── __init__.py
│ │ ├── router.py # 路由汇总
│ │ └── v1/ # API 版本 v1
│ │ ├── __init__.py
│ │ ├── auth.py # 认证路由
│ │ ├── medicines.py # 药品路由
│ │ ├── batches.py # 批次路由
│ │ ├── categories.py # 分类路由
│ │ ├── search.py # 搜索路由
│ │ ├── notifications.py # 通知路由
│ │ ├── ai.py # AI 识别路由
│ │ ├── users.py # 用户管理路由
│ │ └── settings.py # 设置路由
│ ├── models/ # SQLAlchemy 模型
│ │ ├── __init__.py
│ │ ├── user.py # 用户模型
│ │ ├── medicine.py # 药品模型
│ │ ├── batch.py # 批次模型
│ │ ├── category.py # 分类模型
│ │ ├── audit_log.py # 审计日志模型
│ │ ├── notification.py # 通知模型
│ │ └── setting.py # 设置模型
│ ├── schemas/ # Pydantic 模型
│ │ ├── __init__.py
│ │ ├── user.py # 用户 Schema
│ │ ├── medicine.py # 药品 Schema
│ │ ├── batch.py # 批次 Schema
│ │ ├── category.py # 分类 Schema
│ │ ├── audit_log.py # 审计日志 Schema
│ │ ├── notification.py # 通知 Schema
│ │ └── auth.py # 认证 Schema
│ ├── services/ # 服务层
│ │ ├── __init__.py
│ │ ├── auth.py # 认证服务
│ │ ├── user.py # 用户服务
│ │ ├── medicine.py # 药品服务
│ │ ├── batch.py # 批次服务
│ │ ├── category.py # 分类服务
│ │ ├── notification.py # 通知服务
│ │ ├── search.py # 搜索服务
│ │ └── audit.py # 审计服务
│ ├── repositories/ # 数据访问层
│ │ ├── __init__.py
│ │ ├── user.py # 用户仓库
│ │ ├── medicine.py # 药品仓库
│ │ ├── batch.py # 批次仓库
│ │ ├── category.py # 分类仓库
│ │ ├── audit_log.py # 审计日志仓库
│ │ └── notification.py # 通知仓库
│ ├── ai/ # AI Provider
│ │ ├── __init__.py
│ │ ├── base.py # 抽象基类
│ │ ├── openai_provider.py # OpenAI 实现
│ │ ├── gemini_provider.py # Gemini 实现
│ │ ├── claude_provider.py # Claude 实现
│ │ ├── deepseek_provider.py # DeepSeek 实现
│ │ ├── ollama_provider.py # Ollama 实现
│ │ └── manager.py # Provider 管理器
│ ├── notifications/ # 通知系统
│ │ ├── __init__.py
│ │ ├── base.py # 抽象基类
│ │ ├── serverchan.py # Server酱
│ │ ├── pushplus.py # PushPlus
│ │ ├── bark.py # Bark
│ │ ├── wechat.py # 企业微信
│ │ ├── telegram.py # Telegram
│ │ ├── email.py # 邮件
│ │ └── manager.py # 通知管理器
│ ├── storage/ # 文件存储
│ │ ├── __init__.py
│ │ ├── base.py # 抽象基类
│ │ ├── local.py # 本地存储
│ │ └── manager.py # 存储管理器
│ ├── core/ # 核心功能
│ │ ├── __init__.py
│ │ ├── security.py # 安全工具（密码哈希、JWT）
│ │ ├── deps.py # 依赖注入
│ │ └── exceptions.py # 自定义异常
│ └── tasks/ # 异步任务
│ ├── __init__.py
│ ├── expiry_check.py # 到期检查任务
│ └── stock_check.py # 库存检查任务
├── alembic/ # 数据库迁移
│ ├── versions/
│ ├── env.py
│ └── script.py.mako
├── tests/ # 测试文件
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_auth.py
│ ├── test_medicines.py
│ └── test_batches.py
├── migrations/ # 迁移脚本
├── requirements.txt # 依赖配置
├── alembic.ini # Alembic 配置
├── Dockerfile # Docker 配置
├── docker-compose.yml # Docker Compose 配置
├── .env.example # 环境变量示例
├── .env # 环境变量（不提交）
├── pytest.ini # Pytest 配置
└── README.md # 后端说明
```
## 2. 核心依赖
```txt
# requirements.txt
# Web 框架
fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
# 数据库
sqlalchemy==2.0.23
alembic==1.13.0
aiosqlite==0.19.0
# 认证
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
# 数据验证
pydantic==2.5.2
pydantic-settings==2.1.0
# AI 服务
openai==1.6.1
google-generativeai==0.3.2
anthropic==0.8.0
httpx==0.25.2
# 通知
aiohttp==3.9.1
# 文件处理
aiofiles==23.2.1
Pillow==10.1.0
# 工具
python-dotenv==1.0.0
loguru==0.7.2
apscheduler==3.10.4
# 测试
pytest==7.4.3
pytest-asyncio==0.23.2
httpx==0.25.2
```
## 3. 应用入口
```python
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import settings
from app.database import engine, SessionLocal
from app.api.router import api_router
from app.core.exceptions import register_exception_handlers
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时
print("Starting YaoXiang API...")
yield
# 关闭时
print("Shutting down YaoXiang API...")
app = FastAPI(
title="药箱 API",
description="家庭药品与应急物资管理系统 API",
version="1.0.0",
lifespan=lifespan,
docs_url="/docs" if settings.DEBUG else None,
redoc_url="/redoc" if settings.DEBUG else None,
)
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 注册路由
app.include_router(api_router, prefix="/api")
# 注册异常处理器
register_exception_handlers(app)
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "version": "1.0.0"}
```
## 4. 配置管理
```python
# app/config.py
from pydantic_settings import BaseSettings
from typing import List
from functools import lru_cache
class Settings(BaseSettings):
"""应用配置"""
# 应用配置
APP_NAME: str = "药箱"
APP_VERSION: str = "1.0.0"
DEBUG: bool = False
# 数据库配置
DATABASE_URL: str = "sqlite+aiosqlite:///./data/yaoxiang.db"
# 安全配置
JWT_SECRET_KEY: str = "your-secret-key-change-in-production"
JWT_ALGORITHM: str = "HS256"
JWT_EXPIRATION_HOURS: int = 24
# AI Provider 配置
AI_PROVIDER: str = "openai"
OPENAI_API_KEY: str = ""
OPENAI_MODEL: str = "gpt-4o"
GEMINI_API_KEY: str = ""
GEMINI_MODEL: str = "gemini-pro-vision"
ANTHROPIC_API_KEY: str = ""
ANTHROPIC_MODEL: str = "claude-3-opus-20240229"
DEEPSEEK_API_KEY: str = ""
DEEPSEEK_MODEL: str = "deepseek-chat"
OLLAMA_BASE_URL: str = "http://localhost:11434"
OLLAMA_MODEL: str = "llava"
# 通知配置
NOTIFICATION_PROVIDERS: List[str] = []
SERVERCHAN_KEY: str = ""
PUSHPLUS_TOKEN: str = ""
BARK_URL: str = ""
WECHAT_WEBHOOK_URL: str = ""
TELEGRAM_BOT_TOKEN: str = ""
TELEGRAM_CHAT_ID: str = ""
SMTP_HOST: str = ""
SMTP_PORT: int = 587
SMTP_USER: str = ""
SMTP_PASSWORD: str = ""
SMTP_FROM: str = ""
# 文件存储配置
UPLOAD_DIR: str = "./data/uploads"
MAX_UPLOAD_SIZE: int = 10485760 # 10MB
# 到期提醒配置
EXPIRY_WARNING_DAYS: List[int] = [90, 30, 7]
# 低库存阈值
LOW_STOCK_THRESHOLD: int = 5
# CORS 配置
CORS_ORIGINS: List[str] = ["http://localhost:5173", "http://localhost:3000"]
# 宽限天数最大值
EXPIRY_GRACE_DAYS_MAX: int = 60
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
@lru_cache()
def get_settings() -> Settings:
return Settings()
settings = get_settings()
```
## 5. 数据库配置
```python
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.config import settings
# 创建异步引擎
engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
future=True,
)
# 创建会话工厂
async_session_factory = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
class Base(DeclarativeBase):
"""模型基类"""
pass
async def get_db() -> AsyncSession:
"""获取数据库会话"""
async with async_session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def init_db():
"""初始化数据库"""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
```
## 6. 数据模型
### 6.1 用户模型
```python
# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, index=True, nullable=False)
password_hash = Column(String(255), nullable=False)
role = Column(String(20), nullable=False, default="user")
display_name = Column(String(100))
email = Column(String(100))
notification_level = Column(String(20), default="normal")
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
medicines = relationship("Medicine", back_populates="creator")
audit_logs = relationship("AuditLog", back_populates="user")
notifications = relationship("Notification", back_populates="user")
```
### 6.2 药品模型
```python
# app/models/medicine.py
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, JSON
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base
class Medicine(Base):
__tablename__ = "medicines"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(200), nullable=False, index=True)
generic_name = Column(String(200), index=True)
brand_name = Column(String(200))
manufacturer = Column(String(200))
specification = Column(String(200))
category_id = Column(Integer, ForeignKey("categories.id"))
description = Column(Text)
indications = Column(Text)
adult_dose = Column(Text)
child_dose = Column(Text)
contraindications = Column(Text)
notes = Column(Text)
image_front_path = Column(String(500))
image_expiry_path = Column(String(500))
image_leaflet_paths = Column(JSON)
expiry_grace_days = Column(Integer, default=0)
created_by = Column(Integer, ForeignKey("users.id"))
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
category = relationship("Category", back_populates="medicines")
creator = relationship("User", back_populates="medicines")
batches = relationship("Batch", back_populates="medicine", cascade="all, delete-orphan")
audit_logs = relationship("AuditLog", back_populates="medicine")
```
### 6.3 批次模型
```python
# app/models/batch.py
from sqlalchemy import Column, Integer, String, Date, Boolean, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base
class Batch(Base):
__tablename__ = "batches"
id = Column(Integer, primary_key=True, index=True)
medicine_id = Column(Integer, ForeignKey("medicines.id"), nullable=False, index=True)
batch_no = Column(String(100))
production_date = Column(Date)
expiry_date = Column(Date, nullable=False)
quantity = Column(Integer, nullable=False, default=0)
location = Column(String(200))
is_expired = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
medicine = relationship("Medicine", back_populates="batches")
audit_logs = relationship("AuditLog", back_populates="batch")
```
### 6.4 分类模型
```python
# app/models/category.py
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base
class Category(Base):
__tablename__ = "categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
parent_id = Column(Integer, ForeignKey("categories.id"))
level = Column(Integer, nullable=False, default=1)
icon = Column(String(50))
sort_order = Column(Integer, default=0)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
parent = relationship("Category", remote_side=[id])
children = relationship("Category", back_populates="parent")
medicines = relationship("Medicine", back_populates="category")
```
### 6.5 审计日志模型
```python
# app/models/audit_log.py
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
medicine_id = Column(Integer, ForeignKey("medicines.id"), nullable=False, index=True)
batch_id = Column(Integer, ForeignKey("batches.id"))
user_id = Column(Integer, ForeignKey("users.id"))
action = Column(String(50), nullable=False)
quantity_change = Column(Integer, nullable=False)
quantity_after = Column(Integer, nullable=False)
remark = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
# 关系
medicine = relationship("Medicine", back_populates="audit_logs")
batch = relationship("Batch", back_populates="audit_logs")
user = relationship("User", back_populates="audit_logs")
```
## 7. Pydantic Schema
### 7.1 用户 Schema
```python
# app/schemas/user.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class UserBase(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
display_name: Optional[str] = None
email: Optional[str] = None
role: str = Field(default="user", pattern="^(admin|user|readonly)$")
notification_level: str = Field(default="normal", pattern="^(none|low|normal|high)$")
class UserCreate(UserBase):
password: str = Field(..., min_length=6)
class UserUpdate(BaseModel):
display_name: Optional[str] = None
email: Optional[str] = None
role: Optional[str] = None
notification_level: Optional[str] = None
is_active: Optional[bool] = None
class UserResponse(UserBase):
id: int
is_active: bool
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class UserLogin(BaseModel):
username: str
password: str
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
user: UserResponse
```
### 7.2 药品 Schema
```python
# app/schemas/medicine.py
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime, date
class MedicineBase(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
generic_name: Optional[str] = None
brand_name: Optional[str] = None
manufacturer: Optional[str] = None
specification: Optional[str] = None
category_id: Optional[int] = None
description: Optional[str] = None
indications: Optional[str] = None
adult_dose: Optional[str] = None
child_dose: Optional[str] = None
contraindications: Optional[str] = None
notes: Optional[str] = None
expiry_grace_days: int = Field(default=0, ge=0, le=60)
class MedicineCreate(MedicineBase):
pass
class MedicineUpdate(BaseModel):
name: Optional[str] = None
generic_name: Optional[str] = None
brand_name: Optional[str] = None
manufacturer: Optional[str] = None
specification: Optional[str] = None
category_id: Optional[int] = None
description: Optional[str] = None
indications: Optional[str] = None
adult_dose: Optional[str] = None
child_dose: Optional[str] = None
contraindications: Optional[str] = None
notes: Optional[str] = None
expiry_grace_days: Optional[int] = None
class MedicineResponse(MedicineBase):
id: int
image_front_path: Optional[str] = None
image_expiry_path: Optional[str] = None
image_leaflet_paths: Optional[List[str]] = None
created_by: Optional[int] = None
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class MedicineWithStock(MedicineResponse):
total_quantity: int = 0
nearest_expiry_date: Optional[date] = None
batch_count: int = 0
```
### 7.3 批次 Schema
```python
# app/schemas/batch.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import date, datetime
class BatchBase(BaseModel):
batch_no: Optional[str] = None
production_date: Optional[date] = None
expiry_date: date
quantity: int = Field(default=0, ge=0)
location: Optional[str] = None
class BatchCreate(BatchBase):
pass
class BatchUpdate(BaseModel):
batch_no: Optional[str] = None
production_date: Optional[date] = None
expiry_date: Optional[date] = None
quantity: Optional[int] = None
location: Optional[str] = None
class BatchResponse(BatchBase):
id: int
medicine_id: int
is_expired: bool
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class BatchDispense(BaseModel):
quantity: int = Field(..., gt=0)
class BatchAddStock(BaseModel):
quantity: int = Field(..., gt=0)
```
## 8. 服务层设计
### 8.1 药品服务
```python
# app/services/medicine.py
from typing import List, Optional
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models.medicine import Medicine
from app.models.batch import Batch
from app.schemas.medicine import MedicineCreate, MedicineUpdate, MedicineWithStock
from app.repositories.medicine import MedicineRepository
class MedicineService:
def __init__(self, db: AsyncSession):
self.db = db
self.repo = MedicineRepository(db)
async def get_medicines(
self,
category_id: Optional[int] = None,
search: Optional[str] = None,
page: int = 1,
page_size: int = 20
) -> tuple[List[MedicineWithStock], int]:
"""获取药品列表"""
medicines, total = await self.repo.get_list(
category_id=category_id,
search=search,
page=page,
page_size=page_size
)
result = []
for medicine in medicines:
# 计算库存
total_quantity = sum(b.quantity for b in medicine.batches if not b.is_expired)
# 获取最近过期日期
nearest_expiry = None
for batch in medicine.batches:
if not batch.is_expired:
if nearest_expiry is None or batch.expiry_date < nearest_expiry:
nearest_expiry = batch.expiry_date
medicine_with_stock = MedicineWithStock(
**medicine.__dict__,
total_quantity=total_quantity,
nearest_expiry_date=nearest_expiry,
batch_count=len([b for b in medicine.batches if not b.is_expired])
)
result.append(medicine_with_stock)
return result, total
async def get_medicine(self, medicine_id: int) -> Optional[Medicine]:
"""获取药品详情"""
return await self.repo.get_by_id(medicine_id)
async def create_medicine(self, data: MedicineCreate, user_id: int) -> Medicine:
"""创建药品"""
medicine_data = data.model_dump()
medicine_data['created_by'] = user_id
return await self.repo.create(medicine_data)
async def update_medicine(self, medicine_id: int, data: MedicineUpdate) -> Optional[Medicine]:
"""更新药品"""
update_data = data.model_dump(exclude_unset=True)
return await self.repo.update(medicine_id, update_data)
async def delete_medicine(self, medicine_id: int) -> bool:
"""删除药品"""
return await self.repo.delete(medicine_id)
async def search_medicines(self, query: str) -> List[Medicine]:
"""搜索药品"""
return await self.repo.search(query)
```
### 8.2 批次服务
```python
# app/services/batch.py
from typing import List, Optional
from datetime import date, timedelta
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.batch import Batch
from app.schemas.batch import BatchCreate, BatchUpdate
from app.repositories.batch import BatchRepository
from app.services.audit import AuditService
class BatchService:
def __init__(self, db: AsyncSession):
self.db = db
self.repo = BatchRepository(db)
self.audit_service = AuditService(db)
async def get_batches_by_medicine(self, medicine_id: int) -> List[Batch]:
"""获取药品的所有批次"""
return await self.repo.get_by_medicine_id(medicine_id)
async def get_batch(self, batch_id: int) -> Optional[Batch]:
"""获取批次详情"""
return await self.repo.get_by_id(batch_id)
async def create_batch(self, medicine_id: int, data: BatchCreate) -> Batch:
"""创建批次"""
batch_data = data.model_dump()
batch_data['medicine_id'] = medicine_id
return await self.repo.create(batch_data)
async def update_batch(self, batch_id: int, data: BatchUpdate) -> Optional[Batch]:
"""更新批次"""
update_data = data.model_dump(exclude_unset=True)
return await self.repo.update(batch_id, update_data)
async def delete_batch(self, batch_id: int) -> bool:
"""删除批次"""
return await self.repo.delete(batch_id)
async def dispense(self, batch_id: int, quantity: int, user_id: int) -> Optional[Batch]:
"""取药（扣减库存）"""
batch = await self.repo.get_by_id(batch_id)
if not batch:
raise ValueError("批次不存在")
if batch.quantity < quantity:
raise ValueError("库存不足")
# 记录审计日志
await self.audit_service.log_action(
medicine_id=batch.medicine_id,
batch_id=batch_id,
user_id=user_id,
action="dispense",
quantity_change=-quantity,
quantity_after=batch.quantity - quantity
)
# 扣减库存
batch.quantity -= quantity
await self.db.commit()
return batch
async def add_stock(self, batch_id: int, quantity: int, user_id: int) -> Optional[Batch]:
"""入库（增加库存）"""
batch = await self.repo.get_by_id(batch_id)
if not batch:
raise ValueError("批次不存在")
# 记录审计日志
await self.audit_service.log_action(
medicine_id=batch.medicine_id,
batch_id=batch_id,
user_id=user_id,
action="add_stock",
quantity_change=quantity,
quantity_after=batch.quantity + quantity
)
# 增加库存
batch.quantity += quantity
await self.db.commit()
return batch
async def check_expiring_batches(self, warning_days: List[int]) -> List[dict]:
"""检查即将过期的批次"""
expiring = []
today = date.today()
for days in warning_days:
target_date = today + timedelta(days=days)
batches = await self.repo.get_expiring_before(target_date)
for batch in batches:
expiring.append({
'batch': batch,
'days_until_expiry': days
})
return expiring
```
## 9. AI Provider 设计
### 9.1 抽象基类
```python
# app/ai/base.py
from abc import ABC, abstractmethod
from typing import Optional
from pydantic import BaseModel
class VisionResult(BaseModel):
"""视觉识别结果"""
generic_name: Optional[str] = None
brand_name: Optional[str] = None
manufacturer: Optional[str] = None
specification: Optional[str] = None
class DateResult(BaseModel):
"""日期识别结果"""
production_date: Optional[str] = None
expiry_date: Optional[str] = None
class LeafletResult(BaseModel):
"""说明书识别结果"""
indications: str
adult_dose: str
child_dose: Optional[str] = None
contraindications: str
notes: Optional[str] = None
class VisionProvider(ABC):
"""视觉模型提供者抽象基类"""
@abstractmethod
async def recognize_medicine(self, image_bytes: bytes) -> VisionResult:
"""识别药盒信息"""
pass
@abstractmethod
async def recognize_dates(self, image_bytes: bytes) -> DateResult:
"""识别日期信息"""
pass
class TextProvider(ABC):
"""文本模型提供者抽象基类"""
@abstractmethod
async def summarize_leaflet(self, text: str) -> LeafletResult:
"""总结说明书内容"""
pass
@abstractmethod
async def natural_language_search(self, query: str, medicines: list) -> list:
"""自然语言搜索"""
pass
```
### 9.2 OpenAI 实现
```python
# app/ai/openai_provider.py
import base64
from openai import AsyncOpenAI
from app.ai.base import VisionProvider, TextProvider, VisionResult, DateResult, LeafletResult
from app.config import settings
class OpenAIVisionProvider(VisionProvider):
"""OpenAI 视觉模型提供者"""
def __init__(self):
self.client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
self.model = settings.OPENAI_MODEL
async def recognize_medicine(self, image_bytes: bytes) -> VisionResult:
"""识别药盒信息"""
base64_image = base64.b64encode(image_bytes).decode('utf-8')
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": """请识别这张药品包装图片中的信息，返回JSON格式：
{
"generic_name": "通用名称",
"brand_name": "商品名称",
"manufacturer": "生产厂家",
"specification": "规格"
}
只提取图片中真实出现的内容，不要猜测。"""
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return VisionResult(**result)
async def recognize_dates(self, image_bytes: bytes) -> DateResult:
"""识别日期信息"""
base64_image = base64.b64encode(image_bytes).decode('utf-8')
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": """请识别这张图片中的日期信息，返回JSON格式：
{
"production_date": "生产日期(YYYY-MM-DD格式，如果无法识别则为null)",
"expiry_date": "有效期/过期日期(YYYY-MM-DD格式，如果无法识别则为null)"
}
只提取图片中真实出现的日期，不要猜测或推理。"""
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return DateResult(**result)
class OpenAITextProvider(TextProvider):
"""OpenAI 文本模型提供者"""
def __init__(self):
self.client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
self.model = "gpt-4"
async def summarize_leaflet(self, text: str) -> LeafletResult:
"""总结说明书内容"""
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "你是一个医疗信息提取助手。请从药品说明书中提取关键信息。"
},
{
"role": "user",
"content": f"""请从以下药品说明书中提取关键信息，返回JSON格式：
{{
"indications": "适应症",
"adult_dose": "成人用法用量",
"child_dose": "儿童用法用量(如果没有则为null)",
"contraindications": "禁忌",
"notes": "注意事项(如果有)"
}}
说明书内容：
{text}"""
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return LeafletResult(**result)
async def natural_language_search(self, query: str, medicines: list) -> list:
"""自然语言搜索"""
medicines_text = "\n".join([
f"- {m['name']}: {m.get('indications', '')}"
for m in medicines
])
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "你是一个药品搜索助手。根据用户描述的症状，从药品列表中找出可能适用的药品。"
},
{
"role": "user",
"content": f"""用户描述：{query}
可用药品列表：
{medicines_text}
请返回JSON格式的搜索结果：
{{
"results": [
{{
"medicine_id": 药品ID,
"name": "药品名称",
"reason": "匹配原因"
}}
]
}}"""
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return result.get('results', [])
```
### 9.3 Provider 管理器
```python
# app/ai/manager.py
from typing import Optional
from app.ai.base import VisionProvider, TextProvider
class AIManager:
"""AI Provider 管理器"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
self.vision_providers: dict[str, VisionProvider] = {}
self.text_providers: dict[str, TextProvider] = {}
self._initialized = True
def register_vision_provider(self, name: str, provider: VisionProvider):
"""注册视觉模型提供者"""
self.vision_providers[name] = provider
def register_text_provider(self, name: str, provider: TextProvider):
"""注册文本模型提供者"""
self.text_providers[name] = provider
def get_vision_provider(self, name: str) -> Optional[VisionProvider]:
"""获取视觉模型提供者"""
return self.vision_providers.get(name)
def get_text_provider(self, name: str) -> Optional[TextProvider]:
"""获取文本模型提供者"""
return self.text_providers.get(name)
# 全局管理器实例
ai_manager = AIManager()
```
## 10. 认证与授权
### 10.1 安全工具
```python
# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""验证密码"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""获取密码哈希"""
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""创建访问令牌"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(hours=settings.JWT_EXPIRATION_HOURS)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
return encoded_jwt
def decode_access_token(token: str) -> Optional[dict]:
"""解码访问令牌"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
return payload
except JWTError:
return None
```
### 10.2 依赖注入
```python
# app/core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.core.security import decode_access_token
from app.services.user import UserService
from app.models.user import User
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db)
) -> User:
"""获取当前用户"""
token = credentials.credentials
payload = decode_access_token(token)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的认证令牌"
)
user_id = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的认证令牌"
)
user_service = UserService(db)
user = await user_service.get_user(int(user_id))
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户不存在"
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="用户已被禁用"
)
return user
def require_role(roles: list[str]):
"""要求特定角色的依赖"""
async def role_checker(current_user: User = Depends(get_current_user)):
if current_user.role not in roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="权限不足"
)
return current_user
return role_checker
```
## 11. 路由设计
### 11.1 路由汇总
```python
# app/api/router.py
from fastapi import APIRouter
from app.api.v1 import auth, medicines, batches, categories, search, notifications, ai, users, settings
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/v1/auth", tags=["认证"])
api_router.include_router(medicines.router, prefix="/v1/medicines", tags=["药品管理"])
api_router.include_router(batches.router, prefix="/v1/batches", tags=["批次管理"])
api_router.include_router(categories.router, prefix="/v1/categories", tags=["分类管理"])
api_router.include_router(search.router, prefix="/v1/search", tags=["搜索"])
api_router.include_router(notifications.router, prefix="/v1/notifications", tags=["通知"])
api_router.include_router(ai.router, prefix="/v1/ai", tags=["AI 识别"])
api_router.include_router(users.router, prefix="/v1/users", tags=["用户管理"])
api_router.include_router(settings.router, prefix="/v1/settings", tags=["系统设置"])
```
### 11.2 药品路由示例
```python
# app/api/v1/medicines.py
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.core.deps import get_current_user, require_role
from app.models.user import User
from app.schemas.medicine import MedicineCreate, MedicineUpdate, MedicineResponse, MedicineWithStock
from app.services.medicine import MedicineService
router = APIRouter()
@router.get("/", response_model=dict)
async def list_medicines(
category_id: Optional[int] = Query(None),
search: Optional[str] = Query(None),
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""获取药品列表"""
service = MedicineService(db)
medicines, total = await service.get_medicines(
category_id=category_id,
search=search,
page=page,
page_size=page_size
)
return {
"data": medicines,
"total": total,
"page": page,
"page_size": page_size
}
@router.get("/{medicine_id}", response_model=MedicineResponse)
async def get_medicine(
medicine_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""获取药品详情"""
service = MedicineService(db)
medicine = await service.get_medicine(medicine_id)
if not medicine:
raise HTTPException(status_code=404, detail="药品不存在")
return medicine
@router.post("/", response_model=MedicineResponse)
async def create_medicine(
data: MedicineCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "user"]))
):
"""创建药品"""
service = MedicineService(db)
medicine = await service.create_medicine(data, current_user.id)
return medicine
@router.put("/{medicine_id}", response_model=MedicineResponse)
async def update_medicine(
medicine_id: int,
data: MedicineUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "user"]))
):
"""更新药品"""
service = MedicineService(db)
medicine = await service.update_medicine(medicine_id, data)
if not medicine:
raise HTTPException(status_code=404, detail="药品不存在")
return medicine
@router.delete("/{medicine_id}")
async def delete_medicine(
medicine_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin"]))
):
"""删除药品"""
service = MedicineService(db)
success = await service.delete_medicine(medicine_id)
if not success:
raise HTTPException(status_code=404, detail="药品不存在")
return {"message": "删除成功"}
```
## 12. 通知系统
### 12.1 通知提供者
```python
# app/notifications/base.py
from abc import ABC, abstractmethod
class NotificationProvider(ABC):
"""通知提供者抽象基类"""
@abstractmethod
async def send(self, title: str, content: str) -> bool:
"""发送通知"""
pass
@abstractmethod
def validate_config(self) -> bool:
"""验证配置"""
pass
```
### 12.2 Server酱实现
```python
# app/notifications/serverchan.py
import httpx
from app.notifications.base import NotificationProvider
from app.config import settings
class ServerChanProvider(NotificationProvider):
"""Server酱通知提供者"""
def __init__(self):
self.key = settings.SERVERCHAN_KEY
def validate_config(self) -> bool:
return bool(self.key)
async def send(self, title: str, content: str) -> bool:
"""发送通知"""
if not self.validate_config():
return False
url = f"https://sctapi.ftqq.com/{self.key}.send"
data = {
"title": title,
"desp": content
}
async with httpx.AsyncClient() as client:
response = await client.post(url, data=data)
return response.status_code == 200
```
## 13. 异步任务
### 13.1 到期检查任务
```python
# app/tasks/expiry_check.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from sqlalchemy import select
from datetime import date, timedelta
from app.database import async_session_factory
from app.models.batch import Batch
from app.models.medicine import Medicine
from app.services.notification import NotificationService
from app.config import settings
scheduler = AsyncIOScheduler()
async def check_expiring_medicines():
"""检查即将过期的药品"""
async with async_session_factory() as db:
today = date.today()
for days in settings.EXPIRY_WARNING_DAYS:
target_date = today + timedelta(days=days)
# 查询即将过期的批次
query = select(Batch, Medicine).join(
Medicine, Batch.medicine_id == Medicine.id
).where(
Batch.expiry_date <= target_date,
Batch.is_expired == False,
Batch.quantity > 0
)
result = await db.execute(query)
batches = result.all()
if batches:
notification_service = NotificationService()
title = f"药品过期提醒 ({days}天内)"
content = "以下药品即将过期，请及时处理：\n\n"
for batch, medicine in batches:
content += f"- {medicine.name}: {batch.batch_no or '默认批次'} "
content += f"(过期日期: {batch.expiry_date})\n"
await notification_service.send_notification(title, content)
def start_expiry_check_task():
"""启动到期检查任务"""
scheduler.add_job(
check_expiring_medicines,
'cron',
hour=9,
minute=0,
id='expiry_check'
)
scheduler.start()
```
Reference in New Issue View Git Blame Copy Permalink