"""DePIN 控制面数据模型。 核心表： - User：用户与钱包地址 - ComputeNode：算力节点、授权金额、12 个月授权周期、在线状态 - NodeYield：每日算力产出与积分流水 - WithdrawalLog：数字资产提现申请、链上哈希与状态 支撑表（业务计算所依赖，不单独对外当作资产账本）： - UptimeInterval：一次连续在线区间，用来计算有效在线时长 - ComputeTask：下发给节点的模拟计算任务及回传校验结果 - SchedulerLock：多实例部署时，定时任务靠数据库互斥，避免重复下发和重复清算 """ from __future__ import annotations import enum from datetime import date, datetime from decimal import Decimal from sqlalchemy import ( CheckConstraint, Enum, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.types import JSON, TypeDecorator from app.clock import utcnow from app.database import Base, UtcDateTime from app.money import money class PreciseDecimal(TypeDecorator): """金额与比率。SQLite 会把 Numeric 读成 float，这里统一还原成 8 位小数。""" impl = Numeric(24, 8) cache_ok = True def process_bind_param(self, value, dialect): if value is None: return None return money(value) def process_result_value(self, value, dialect): if value is None: return None return money(value) MONEY = PreciseDecimal() RATE = PreciseDecimal() def _enum(enum_cls: type[enum.Enum], length: int = 16) -> Enum: """把枚举的 value（online/offline）写入数据库，而不是成员名。""" return Enum( enum_cls, native_enum=False, length=length, values_callable=lambda obj: [item.value for item in obj], ) class NodeStatus(str, enum.Enum): ONLINE = "online" OFFLINE = "offline" class YieldStatus(str, enum.Enum): SETTLED = "settled" VOID = "void" class WithdrawalStatus(str, enum.Enum): PENDING = "pending" COMPLETED = "completed" REJECTED = "rejected" FAILED = "failed" class TaskStatus(str, enum.Enum): PENDING = "pending" COMPLETED = "completed" FAILED = "failed" EXPIRED = "expired" class PushStatus(str, enum.Enum): QUEUED = "queued" PUSHED = "pushed" FAILED = "failed" class User(Base): """用户。一个钱包地址对应一名用户，可绑定多台算力节点。""" __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) wallet_address: Mapped[str] = mapped_column(String(128), unique=True, index=True) created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) nodes: Mapped[list[ComputeNode]] = relationship(back_populates="user") yields: Mapped[list[NodeYield]] = relationship(back_populates="user") withdrawals: Mapped[list[WithdrawalLog]] = relationship(back_populates="user") class ComputeNode(Base): """算力节点。购买金额在授权周期内按天线性释放，实际到账再乘当日表现系数。""" __tablename__ = "compute_nodes" __table_args__ = ( CheckConstraint("purchase_amount > 0", name="ck_node_purchase_positive"), CheckConstraint("license_months > 0", name="ck_node_license_months"), Index("ix_node_status_heartbeat", "status", "last_heartbeat_at"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) sn: Mapped[str] = mapped_column(String(64), unique=True, index=True, comment="节点唯一序列号") purchase_amount: Mapped[Decimal] = mapped_column(MONEY, comment="授权购买金额") license_months: Mapped[int] = mapped_column(Integer, default=12) licensed_at: Mapped[datetime] = mapped_column(UtcDateTime, comment="授权开始时间") license_expires_at: Mapped[datetime] = mapped_column(UtcDateTime, comment="授权结束时间") status: Mapped[NodeStatus] = mapped_column(_enum(NodeStatus), default=NodeStatus.OFFLINE, index=True) last_heartbeat_at: Mapped[datetime | None] = mapped_column(UtcDateTime) callback_url: Mapped[str | None] = mapped_column(String(512), comment="可选的任务推送地址") created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) updated_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow, onupdate=utcnow) user: Mapped[User] = relationship(back_populates="nodes") yields: Mapped[list[NodeYield]] = relationship(back_populates="node") tasks: Mapped[list[ComputeTask]] = relationship(back_populates="node") uptime_intervals: Mapped[list[UptimeInterval]] = relationship(back_populates="node") class UptimeInterval(Base): """一段连续在线时间。心跳首次上线时打开，超时或授权到期时关闭。""" __tablename__ = "uptime_intervals" __table_args__ = (Index("ix_uptime_node_window", "node_id", "started_at", "ended_at"),) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) started_at: Mapped[datetime] = mapped_column(UtcDateTime) ended_at: Mapped[datetime | None] = mapped_column(UtcDateTime) close_reason: Mapped[str | None] = mapped_column(String(32)) node: Mapped[ComputeNode] = relationship(back_populates="uptime_intervals") class ComputeTask(Base): """下发给在线节点的模拟计算任务。客户端须回传 task_id 与一次性 nonce。""" __tablename__ = "compute_tasks" __table_args__ = ( Index("ix_task_node_status", "node_id", "status"), Index("ix_task_dispatched_at", "dispatched_at"), ) id: Mapped[str] = mapped_column(String(36), primary_key=True, comment="task_id") node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) task_type: Mapped[str] = mapped_column(String(32), default="simulated_compute") payload: Mapped[dict] = mapped_column(JSON) status: Mapped[TaskStatus] = mapped_column(_enum(TaskStatus), default=TaskStatus.PENDING, index=True) push_status: Mapped[PushStatus] = mapped_column(_enum(PushStatus), default=PushStatus.QUEUED) push_error: Mapped[str | None] = mapped_column(String(255)) dispatched_at: Mapped[datetime] = mapped_column(UtcDateTime) deadline_at: Mapped[datetime] = mapped_column(UtcDateTime) responded_at: Mapped[datetime | None] = mapped_column(UtcDateTime) success: Mapped[bool | None] = mapped_column() node: Mapped[ComputeNode] = relationship(back_populates="tasks") class NodeYield(Base): """每日产出流水。同一节点同一天只有一条结算记录。""" __tablename__ = "node_yields" __table_args__ = ( UniqueConstraint("node_id", "yield_date", name="uq_node_yield_date"), Index("ix_node_yield_user_date", "user_id", "yield_date"), CheckConstraint("points_delta >= 0", name="ck_yield_points_non_negative"), CheckConstraint("compute_output >= 0", name="ck_yield_output_non_negative"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) yield_date: Mapped[date] = mapped_column(index=True, comment="清算日，UTC") compute_output: Mapped[Decimal] = mapped_column(MONEY, comment="当日有效算力产出，等于授权金额乘表现系数") points_delta: Mapped[Decimal] = mapped_column(MONEY, comment="当日入账积分") online_seconds: Mapped[int] = mapped_column(Integer, default=0) eligible_seconds: Mapped[int] = mapped_column(Integer, default=0) online_rate: Mapped[Decimal] = mapped_column(RATE, default=0) tasks_dispatched: Mapped[int] = mapped_column(Integer, default=0) tasks_completed: Mapped[int] = mapped_column(Integer, default=0) task_response_rate: Mapped[Decimal] = mapped_column(RATE, default=0) online_weight: Mapped[Decimal] = mapped_column(RATE, default=0) task_weight: Mapped[Decimal] = mapped_column(RATE, default=0) performance: Mapped[Decimal] = mapped_column(RATE, default=0, comment="当日表现系数，0 到 1") cycle_days: Mapped[int] = mapped_column(Integer) status: Mapped[YieldStatus] = mapped_column(_enum(YieldStatus), default=YieldStatus.SETTLED) is_partial: Mapped[bool] = mapped_column(default=False, comment="结算时该 UTC 日尚未结束") created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) node: Mapped[ComputeNode] = relationship(back_populates="yields") user: Mapped[User] = relationship(back_populates="yields") class WithdrawalLog(Base): """提现申请。待处理和已完成的金额都会占用可用积分，拒绝后释放。""" __tablename__ = "withdrawal_logs" __table_args__ = ( Index("ix_withdrawal_user_status", "user_id", "status"), CheckConstraint("amount > 0", name="ck_withdrawal_amount_positive"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) amount: Mapped[Decimal] = mapped_column(MONEY) asset_symbol: Mapped[str] = mapped_column(String(32), default="POINT") tx_hash: Mapped[str | None] = mapped_column(String(128), unique=True) status: Mapped[WithdrawalStatus] = mapped_column( _enum(WithdrawalStatus), default=WithdrawalStatus.PENDING, index=True, ) note: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) updated_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow, onupdate=utcnow) user: Mapped[User] = relationship(back_populates="withdrawals") class SchedulerLock(Base): """定时任务锁。锁到期后其他实例才能接手，进程崩溃不会永久占住任务。""" __tablename__ = "scheduler_locks" name: Mapped[str] = mapped_column(String(64), primary_key=True) locked_until: Mapped[datetime] = mapped_column(UtcDateTime) owner: Mapped[str] = mapped_column(String(128))