64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
|
||
from sqlalchemy.dialects.mysql import DECIMAL
|
||
from sqlalchemy.sql import func
|
||
from pydantic import BaseModel, Field
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
from .database import Base
|
||
import random
|
||
import time
|
||
import enum
|
||
|
||
class OrderStatus(str, enum.Enum):
|
||
CREATED = "created" # 已下单
|
||
UNVERIFIED = "unverified" # 未核销
|
||
VERIFIED = "verified" # 已核销
|
||
REFUNDING = "refunding" # 退款中
|
||
REFUNDED = "refunded" # 已退款
|
||
|
||
class MerchantOrderDB(Base):
|
||
__tablename__ = "merchant_orders"
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||
order_id = Column(String(15), unique=True, nullable=False)
|
||
user_id = Column(Integer, ForeignKey("users.userid"), nullable=False)
|
||
merchant_product_id = Column(Integer, ForeignKey("merchant_products.id"), nullable=False)
|
||
order_amount = Column(DECIMAL(10,2), nullable=False)
|
||
status = Column(Enum(OrderStatus), nullable=False, default=OrderStatus.CREATED)
|
||
order_verify_code = Column(String(21), unique=True, nullable=False)
|
||
verify_time = Column(DateTime(timezone=True), nullable=True)
|
||
verify_user_id = Column(Integer, ForeignKey("users.userid"), nullable=True)
|
||
create_time = Column(DateTime(timezone=True), server_default=func.now())
|
||
update_time = Column(DateTime(timezone=True), onupdate=func.now())
|
||
|
||
class MerchantOrderCreate(BaseModel):
|
||
merchant_product_id: int
|
||
order_amount: float = Field(..., gt=0)
|
||
|
||
class MerchantOrderInfo(BaseModel):
|
||
id: int
|
||
order_id: str
|
||
user_id: int
|
||
merchant_product_id: int
|
||
order_amount: float
|
||
status: OrderStatus
|
||
order_verify_code: str
|
||
verify_time: Optional[datetime]
|
||
verify_user_id: Optional[int]
|
||
create_time: datetime
|
||
update_time: Optional[datetime]
|
||
|
||
class Config:
|
||
from_attributes = True
|
||
|
||
def generate_order_id() -> str:
|
||
"""生成订单号：8位日期 + 7位时间戳"""
|
||
now = datetime.now()
|
||
date_str = now.strftime('%Y%m%d')
|
||
# 取时间戳后7位
|
||
timestamp = str(int(time.time() * 1000))[-7:]
|
||
return f"{date_str}{timestamp}"
|
||
|
||
def generate_verify_code() -> str:
|
||
"""生成21位数字核销码"""
|
||
return ''.join(random.choices('0123456789', k=21)) |