"""B2B 订货商城模型(批次 D1)。 设计: - `StorefrontAccount`:客户在商城的登录凭证(与后台用户体系分离,**不占用户席**, 不消耗 billing 的 users 配额)。 - `CustomerProductAuth`:客户可见商品的白名单(默认不可见,授予才可见 —— 白名单比 黑名单安全,漏配不会误暴露)。 - 价格复用 `partner.CustomerProductPrice` + `quote_price` 的既有优先级, 不在商城侧另建价格体系(避免两处价格打架)。 下单走 **SalesOrder 草稿**(内部确认后才转销售单),客户不能直接过账出库。 """ from __future__ import annotations import hashlib import secrets from django.db import models from apps.core.base_models import TenantScopedModel def _hash_token(raw: str) -> str: return hashlib.sha256(raw.encode("utf-8")).hexdigest() class StorefrontAccount(TenantScopedModel): """商城客户账号(手机号 + 密码,独立于后台用户)。""" customer = models.ForeignKey( "partner.Customer", on_delete=models.CASCADE, related_name="storefront_accounts" ) phone = models.CharField(max_length=32, help_text="登录手机号") password_hash = models.CharField(max_length=128) display_name = models.CharField(max_length=64, blank=True, default="") is_active = models.BooleanField(default=True) last_login_at = models.DateTimeField(null=True, blank=True) class Meta: db_table = "storefront_account" unique_together = [("tenant", "phone")] verbose_name_plural = "商城客户账号" ordering = ["-created_at"] def __str__(self): return f"{self.phone} → {self.customer.code}" # ---- 口令 ---- def set_password(self, raw: str) -> None: salt = secrets.token_hex(8) digest = hashlib.pbkdf2_hmac( "sha256", raw.encode("utf-8"), salt.encode("utf-8"), 120_000 ).hex() self.password_hash = f"pbkdf2${salt}${digest}" def check_password(self, raw: str) -> bool: try: algo, salt, digest = self.password_hash.split("$", 2) except ValueError: return False if algo != "pbkdf2": return False candidate = hashlib.pbkdf2_hmac( "sha256", raw.encode("utf-8"), salt.encode("utf-8"), 120_000 ).hex() return secrets.compare_digest(candidate, digest) class CustomerProductAuth(TenantScopedModel): """客户可见商品授权(白名单)。""" customer = models.ForeignKey( "partner.Customer", on_delete=models.CASCADE, related_name="product_auths" ) product = models.ForeignKey( "catalog.Product", on_delete=models.CASCADE, related_name="customer_auths" ) is_active = models.BooleanField(default=True) class Meta: db_table = "storefront_product_auth" unique_together = [("tenant", "customer", "product")] verbose_name_plural = "客户可见商品授权" ordering = ["customer", "product"] def __str__(self): return f"{self.customer.code} → {self.product.code}" class StorefrontOrder(TenantScopedModel): """商城订单(客户自助提交,对内是 SalesOrder 草稿的来源单)。""" STATUS_SUBMITTED = "submitted" STATUS_CONFIRMED = "confirmed" STATUS_REJECTED = "rejected" STATUS_CHOICES = [ (STATUS_SUBMITTED, "已提交待确认"), (STATUS_CONFIRMED, "已确认转单"), (STATUS_REJECTED, "已驳回"), ] order_no = models.CharField(max_length=64) customer = models.ForeignKey( "partner.Customer", on_delete=models.PROTECT, related_name="storefront_orders" ) account = models.ForeignKey( StorefrontAccount, null=True, blank=True, on_delete=models.SET_NULL, related_name="orders", ) status = models.CharField(max_length=16, choices=STATUS_CHOICES, default=STATUS_SUBMITTED) total_amount = models.DecimalField(max_digits=18, decimal_places=4, default=0) remark = models.TextField(blank=True, default="") sales_order = models.ForeignKey( "sales.SalesOrder", null=True, blank=True, on_delete=models.SET_NULL, related_name="storefront_orders", ) class Meta: db_table = "storefront_order" unique_together = [("tenant", "order_no")] verbose_name_plural = "商城订单" ordering = ["-created_at"] def __str__(self): return f"{self.order_no} ({self.get_status_display()})" class StorefrontOrderLine(TenantScopedModel): """商城订单行(价格在下单时快照,避免事后调价影响历史单)。""" order = models.ForeignKey( StorefrontOrder, on_delete=models.CASCADE, related_name="lines" ) product = models.ForeignKey( "catalog.Product", on_delete=models.PROTECT, related_name="storefront_lines" ) quantity = models.DecimalField(max_digits=18, decimal_places=4) unit_price = models.DecimalField(max_digits=18, decimal_places=4) amount = models.DecimalField(max_digits=18, decimal_places=4, default=0) source_unit = models.ForeignKey( "catalog.Unit", null=True, blank=True, on_delete=models.SET_NULL, related_name="storefront_lines", ) source_quantity = models.DecimalField( max_digits=18, decimal_places=4, null=True, blank=True ) class Meta: db_table = "storefront_order_line" ordering = ["id"] def __str__(self): return f"{self.product.code} × {self.quantity}" # ------------------------------------------------------------ # 会话 token(无状态签名,省一张表;密钥取 SECRET_KEY) # ------------------------------------------------------------ def issue_session_token(account: StorefrontAccount, *, ttl_hours: int = 72) -> str: """签发商城会话 token:`account_id.expires.sig`(HMAC-SHA256)。""" import time from django.conf import settings expires = int(time.time()) + ttl_hours * 3600 payload = f"{account.id}.{expires}" sig = hashlib.sha256( f"{payload}.{settings.SECRET_KEY}".encode("utf-8") ).hexdigest()[:32] return f"{payload}.{sig}" def verify_session_token(token: str): """校验 token,返回 StorefrontAccount 或 None。""" import time from django.conf import settings if not token or token.count(".") != 2: return None raw_id, raw_exp, sig = token.split(".") expect = hashlib.sha256( f"{raw_id}.{raw_exp}.{settings.SECRET_KEY}".encode("utf-8") ).hexdigest()[:32] if not secrets.compare_digest(expect, sig): return None try: if int(raw_exp) < int(time.time()): return None account_id = int(raw_id) except ValueError: return None return StorefrontAccount.objects.filter( pk=account_id, is_active=True ).select_related("customer", "tenant").first()