SmartShop 智能购物系统

后端

前端

11 个后端子域（模块化单体 + 事件驱动）

双端页面（参照淘宝网信息架构）

• C 端商城 MallLayout：顶栏搜索 + 类目导航 + 金刚区 + 限时秒杀 + 猜你喜欢瀑布流；商品详情/购物车/确认订单/收银台/我的订单/优惠券/积分/识别/室内导航/直播间/客服
• B 端后台 AdminLayout：侧边菜单 + 面包屑；Dashboard 报表、门店/铺位/路网编辑、商品/库存、订单/退款、活动/券模板、数字人/视频审核/直播管理、识别记录/纠错、会话/工单/知识库、用户角色、字典与能力降级可视化

目录树

SmartShop/
├── app/                                  # ========== 后端（FastAPI 模块化单体）==========
│   ├── main.py                           # 应用装配：lifespan / 中间件 / 静态资源 / 路由
│   ├── core/                             # 基建层
│   │   ├── config.py                     # Settings（pydantic-settings，.env 驱动）
│   │   ├── database.py                   # async engine / Base / Mixin / session_scope
│   │   ├── redis.py                      # Redis 连接 + RedisKeys 集中管理
│   │   ├── event_bus.py                  # 进程内事件总线（订阅 + 幂等消费）
│   │   ├── outbox.py                     # Outbox 发件箱 + 投递器（本地事务一致性）
│   │   ├── security.py                   # JWT / bcrypt / 手机号 AES-256-GCM + HMAC 指纹
│   │   ├── response.py                   # 统一信封 {code,msg,data,traceId}
│   │   ├── exceptions.py                 # ErrorCode + BizError + 全局异常处理
│   │   ├── middleware.py                 # TraceId 链路 / 限流中间件
│   │   ├── idempotency.py                # 分布式锁 + 幂等守卫（SET NX EX + 结果回放）
│   │   ├── dicts.py · cache.py · deps.py · pagination.py · ws.py · context.py · types.py
│   ├── api/v1/
│   │   ├── router.py                     # 子域路由自动挂载（MODULE_ORDER）
│   │   └── system.py                     # health / capabilities / dicts
│   ├── integrations/base.py              # 能力注册表 + real/local/stub 降级链
│   └── modules/                          # ===== 11 子域（六件套：models/schemas/repo/service/api/events）=====
│       ├── identity/                     # 用户/登录/RBAC/字典/积分/地址（P1 进行中）
│       ├── store/                        # 门店/铺位/库存/附近门店
│       ├── product/                      # 类目/SPU/SKU/ngram 全文搜索
│       ├── recognition/                  # 拍照识别：阈值分流 + 纠错回流
│       ├── payment/                      # 下单幂等/金额重算/微信支付回调/退款对账
│       ├── marketing/                    # trade(券/规则引擎/推荐) + avatar + video + live
│       ├── mapguide/                     # 室内导航：NetworkX APSP + 跨楼层拼接
│       └── customer_service/             # 智能客服：意图路由 + RAG + WS
├── alembic/                              # 迁移：0001_init_schema / 0002_seed_dict_and_demo
├── scripts/  check_env.py · init_test_db.py · seed_demo.py
├── tests/    test_00_health.py …（P9 补齐 test_00~test_17）
├── storage/  images/ videos/ covers/ audio/ hls/ chroma/ exports/
├── requirements.txt · pyproject.toml · .env.example · docker-compose.yml · alembic.ini
└── frontend/                             # ========== 前端（Vue3 + Element Plus）==========
    ├── vite.config.ts                    # dev proxy：/api、/static → 127.0.0.1:8000
    ├── package.json
    └── src/
        ├── main.ts · App.vue · env.d.ts
        ├── api/          auth.ts · system.ts · index.ts
        ├── router/       index.ts（登录守卫）· routes/mall.ts · routes/admin.ts
        ├── stores/       auth.ts · user.ts · store.ts
        ├── layouts/      MallLayout.vue（淘宝式顶栏）· AdminLayout.vue（商家后台）
        ├── components/   common/(SearchBar/CategoryTree/ProductCard/Waterfall/PriceBlock/CountDown)
        │                 admin/(CrudTable/QueryForm/DictSelect/StatusTag/UploadImage/RichEditor)
        ├── views/        mall/Home.vue · common/(Login/Forbidden/NotFound)
        │                 admin/(Login/Dashboard/System/Capabilities)
        ├── utils/        request.ts（信封解包+401 刷新重放）· auth.ts · format.ts · dict.ts · validate.ts · ws.ts
        └── styles/       variables.scss（蓝色主题）· index.scss · mall.scss · admin.scss

必需环境

说明

• RabbitMQ / Docker 非必需：一期事件走 Outbox + 进程内总线，二期可替换 Transport
• 无第三方凭证也可端到端跑通：支付/短信/登录/LLM/视觉/TTS 等 13 项能力支持 real / local / stub 三档降级
• Windows 平台：aiomysql 纯 Python 免编译；Celery 使用 -P solo 池

1. 安装依赖

后端依赖见 requirements.txt，前端依赖见 package.json：

# SmartShop 后端依赖（Python 3.11）
# Web
fastapi>=0.115
uvicorn[standard]>=0.30
python-multipart>=0.0.9
# 配置 / 校验
pydantic>=2.9
pydantic-settings>=2.5
# 数据库
SQLAlchemy[asyncio]>=2.0.35
alembic>=1.13
aiomysql>=0.2.0
greenlet>=3.0
# 缓存 / 异步
redis>=5.0
celery>=5.4
# 网络 / 安全
httpx>=0.27
PyJWT>=2.9
cryptography>=43.0
bcrypt>=4.2
# 工具
orjson>=3.10
loguru>=0.7
# 算法 / 内容
networkx>=3.3
websockets>=12.0
jieba>=0.42
chromadb>=0.5
edge-tts>=6.1
imageio-ffmpeg>=0.5
# 测试
pytest>=8.3
pytest-asyncio>=0.24

{
  "name": "smartshop-frontend",
  "private": true,
  "version": "0.1.0",
  "type": "module",
  "scripts": {
    "dev": "vite",
    "build": "vue-tsc --noEmit && vite build",
    "type-check": "vue-tsc --noEmit",
    "preview": "vite preview"
  },
  "dependencies": {
    "@element-plus/icons-vue": "^2.3.1",
    "axios": "^1.7.7",
    "echarts": "^5.5.1",
    "element-plus": "^2.8.8",
    "hls.js": "^1.5.17",
    "pinia": "^2.2.6",
    "vue": "^3.5.13",
    "vue-router": "^4.4.5"
  },
  "devDependencies": {
    "@types/node": "^22.9.0",
    "@vitejs/plugin-vue": "^5.1.4",
    "sass": "^1.79.4",
    "typescript": "^5.6.3",
    "vite": "^5.4.10",
    "vue-tsc": "^2.1.6"
  }
}

2. 环境配置（.env）

复制 .env.example 为 .env，关键配置如下：

# ============ 基础 ============
ENV=dev
DEBUG=true
APP_NAME=SmartShop

# ============ MySQL ============
MYSQL_HOST=127.0.0.1
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=root
MYSQL_DB=smartshopdb

# ============ Redis ============
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=

# ============ JWT ============
JWT_SECRET=smartshop-dev-secret-change-me-in-production
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=120
REFRESH_TOKEN_EXPIRE_DAYS=14

# ============ 敏感字段加密（手机号 AES-256-GCM + HMAC 指纹）============
# 32 字节密钥的 hex（64 个 hex 字符）
MOBILE_AES_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
MOBILE_HMAC_KEY=smartshop-mobile-hmac-key

# ============ CORS ============
CORS_ORIGINS=["http://127.0.0.1:5173","http://localhost:5173"]

# ============ 静态资源 ============
STATIC_URL_PREFIX=/static
STATIC_BASE_URL=http://127.0.0.1:8000

# ============ Outbox ============
OUTBOX_BATCH_SIZE=100
OUTBOX_MAX_RETRY=5
OUTBOX_POLL_INTERVAL=2.0

# ============ Celery ============
CELERY_BROKER_URL=redis://127.0.0.1:6379/1
CELERY_RESULT_BACKEND=redis://127.0.0.1:6379/2
TASKS_ALWAYS_EAGER=false

# ============ 业务阈值 ============
RECOG_HIGH_CONF=0.90
RECOG_LOW_CONF=0.60

# ============ 能力注册表（auto|real|stub）============
STORAGE_MODE=auto
WECHAT_PAY_MODE=auto
WECHAT_LOGIN_MODE=auto
SMS_MODE=auto
MAP_MODE=auto
LLM_MODE=auto
VISION_MODE=auto
EMBEDDING_MODE=auto
VECTOR_MODE=auto
TTS_MODE=auto
AVATAR_MODE=auto
VIDEO_MODE=auto
RTC_MODE=auto

# ============ 第三方凭证（留空即走降级）============
WECHAT_PAY_MCH_ID=
WECHAT_PAY_APP_ID=
WECHAT_PAY_API_V3_KEY=
WECHAT_PAY_CERT_SERIAL_NO=
WECHAT_PAY_PRIVATE_KEY_PATH=
WECHAT_PAY_NOTIFY_URL=http://127.0.0.1:8000/api/v1/pay/notify/wechat
WECHAT_MINI_APP_ID=
WECHAT_MINI_APP_SECRET=

SMS_ACCESS_KEY_ID=
SMS_ACCESS_KEY_SECRET=
SMS_SIGN_NAME=
SMS_TEMPLATE_CODE=

AMAP_KEY=

LLM_BASE_URL=https://api.openai.com/v1
LLM_API_KEY=
LLM_MODEL=gpt-4o-mini
EMBEDDING_DIM=384

3. 初始化与启动

# ===== 后端准备（Windows PowerShell）=====
python -m pip install -r requirements.txt
copy .env.example .env                     # 按需修改 MySQL/Redis 密码
python scripts\check_env.py                # 检查 MySQL/Redis/能力注册表
python scripts\init_test_db.py             # 创建测试库 smartshopdb_test
python -m alembic upgrade head             # 主库建表（约 45 张）
python scripts\seed_demo.py                # 演示数据：门店/商品/券/FAQ/账号

# ===== 启动后端 =====
uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload
# Swagger 文档：http://127.0.0.1:8000/docs   健康检查：/api/v1/health

# ===== 启动前端（另开终端）=====
cd frontend
npm install
npm run dev                                # http://127.0.0.1:5173

# ===== 测试与构建 =====
$env:ENV="test"; $env:MYSQL_DB="smartshopdb_test"
python -m alembic upgrade head
pytest -q
cd frontend; npm run build                 # 生产构建校验

1. 应用装配 — app/main.py

lifespan 中依次初始化 MySQL/Redis、注册事件订阅、启动 OutboxDispatcher；挂载 /static 静态资源、统一异常处理、TraceId 与限流中间件。

"""FastAPI 应用装配。

lifespan：init redis → init db → 注册事件订阅 → 启动 OutboxDispatcher
         → StaticFiles(/static) → 异常处理器 → CORS → 路由
"""

from __future__ import annotations

import asyncio
import importlib
import importlib.util
import pkgutil
from contextlib import asynccontextmanager, suppress

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

import app.modules as modules_pkg
from app.api.v1.router import api_router, health_router
from app.core.config import settings
from app.core.database import close_db, init_db_connection
from app.core.logging import logger, setup_logging
from app.core.middleware import RateLimitMiddleware, TraceIdMiddleware
from app.core.outbox import dispatcher
from app.core.redis import close_redis, init_redis
from app.core.response import register_exception_handlers

STORAGE_SUBDIRS = ("images", "videos", "covers", "audio", "hls", "chroma", "exports")


def _register_subscriptions() -> dict[str, list[str]]:
    """导入各子域 `events.py`，触发 `@bus.subscribe` 注册。"""
    available = sorted({m.name for m in pkgutil.iter_modules(modules_pkg.__path__)})
    for name in available:
        path = f"app.modules.{name}.events"
        if importlib.util.find_spec(path) is None:
            continue
        importlib.import_module(path)
    from app.core.event_bus import bus

    return bus.registered()


def _ensure_storage() -> None:
    base = settings.storage_dir
    for sub in STORAGE_SUBDIRS:
        (base / sub).mkdir(parents=True, exist_ok=True)


@asynccontextmanager
async def lifespan(app: FastAPI):  # noqa: ARG001
    setup_logging()
    logger.info(f"=== {settings.app_name} starting | env={settings.env} ===")

    try:
        await init_db_connection()
        logger.info(f"MySQL connected: {settings.mysql_host}:{settings.mysql_port}/{settings.mysql_db}")
    except Exception as exc:  # noqa: BLE001
        logger.error(f"MySQL unavailable, degraded mode: {exc}")

    try:
        await init_redis()
        logger.info(f"Redis connected: {settings.redis_host}:{settings.redis_port}")
    except Exception as exc:  # noqa: BLE001
        logger.error(f"Redis unavailable, degraded mode: {exc}")

    _ensure_storage()

    subscriptions = _register_subscriptions()
    logger.info(f"event consumers: {subscriptions}")

    dispatch_task = asyncio.create_task(dispatcher.run_forever())

    try:
        yield
    finally:
        logger.info("=== shutting down ===")
        dispatcher.stop()
        dispatch_task.cancel()
        with suppress(asyncio.CancelledError, Exception):
            await dispatch_task
        await close_redis()
        await close_db()


app = FastAPI(
    title=f"{settings.app_name} API",
    version="0.1.0",
    description="智能购物系统（模块化单体 + 事件驱动 + 领域分包）",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
)

_ensure_storage()
app.mount(settings.static_url_prefix, StaticFiles(directory=str(settings.storage_dir)), name="static")

register_exception_handlers(app)

# 中间件：后添加者更靠外层 → TraceId 最外层，保证限流响应也带 traceId
app.add_middleware(RateLimitMiddleware, limit=600, window=60)
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Trace-Id"],
)
app.add_middleware(TraceIdMiddleware)

app.include_router(health_router, prefix=settings.api_prefix, include_in_schema=False)
app.include_router(health_router)
app.include_router(api_router, prefix=settings.api_prefix)


@app.get("/", include_in_schema=False)
async def root() -> dict:
    return {
        "app": settings.app_name,
        "version": "0.1.0",
        "docs": "/docs",
        "health": "/health",
        "apiPrefix": settings.api_prefix,
    }

2. 子域路由聚合 — app/api/v1/router.py

按 MODULE_ORDER 顺序扫描 app/modules/*/api.py 自动挂载子域路由；EnvelopeAPIRouter 自动包装统一信封。

"""API v1 聚合路由：系统路由 + 各子域路由（按固定顺序自动挂载）。"""

from __future__ import annotations

import importlib
import pkgutil

import app.modules as modules_pkg
from app.api.v1 import system
from app.api.v1.system import health_router
from app.core.logging import logger
from app.core.response import EnvelopeAPIRouter

api_router = EnvelopeAPIRouter()

# 系统（health 单独挂在根，其余在 /system）
api_router.include_router(system.router)

# 子域挂载顺序（决定 OpenAPI 中的分组顺序）
MODULE_ORDER = (
    "identity",
    "store",
    "product",
    "recognition",
    "payment",
    "marketing",
    "mapguide",
    "customer_service",
)


def _mount_module_routers() -> list[str]:
    """扫描 `app/modules/*`，导入其 `api.py` 中已带前缀的 `router`。"""
    available = {m.name for m in pkgutil.iter_modules(modules_pkg.__path__)}
    mounted: list[str] = []
    for name in MODULE_ORDER:
        if name not in available:
            continue
        try:
            mod = importlib.import_module(f"app.modules.{name}.api")
        except ModuleNotFoundError as exc:
            logger.debug(f"module {name} has no api yet: {exc}")
            continue
        sub = getattr(mod, "router", None)
        if sub is not None:
            api_router.include_router(sub)
            mounted.append(name)
    return mounted


mounted_modules = _mount_module_routers()

__all__ = ["api_router", "health_router", "mounted_modules"]

3. 异步数据库基建 — app/core/database.py

async engine（pool_pre_ping + orjson 序列化）；Base 带命名约定；Timestamp/SoftDelete/Tenant/Trace 四个 Mixin；get_session 不隐式 commit，session_scope 供任务使用。

"""数据库：async engine / session / Base / Mixin / get_session。"""

from __future__ import annotations

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import datetime

import orjson
from sqlalchemy import (
    BigInteger,
    DateTime,
    MetaData,
    String,
    func,
    text,
)
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from app.core.config import settings

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uk_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}


def _dumps(obj) -> str:
    return orjson.dumps(obj).decode()


engine = create_async_engine(
    settings.database_url,
    echo=settings.mysql_echo,
    pool_pre_ping=True,
    pool_size=settings.mysql_pool_size,
    max_overflow=settings.mysql_max_overflow,
    pool_recycle=1800,
    json_serializer=_dumps,
    json_deserializer=orjson.loads,
    future=True,
)

SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

    def to_dict(self, exclude: set[str] | None = None) -> dict:
        exclude = exclude or set()
        return {
            col.name: getattr(self, col.name)
            for col in self.__table__.columns  # type: ignore[attr-defined]
            if col.name not in exclude
        }


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.now(), comment="创建时间"
    )
    updated_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True, server_default=func.now(), onupdate=func.now(), comment="更新时间"
    )


class SoftDeleteMixin:
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True, default=None, comment="软删时间"
    )


class TenantMixin:
    tenant_id: Mapped[int] = mapped_column(
        BigInteger, nullable=False, server_default=text("0"), comment="租户"
    )


class TraceMixin:
    trace_id: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="链路ID")


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """FastAPI 依赖：异常回滚，**不隐式 commit**。"""
    async with SessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise


@asynccontextmanager
async def session_scope() -> AsyncGenerator[AsyncSession, None]:
    """脚本 / Celery 任务用：自动 commit / rollback。"""
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def init_db_connection() -> bool:
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))
    return True


async def close_db() -> None:
    await engine.dispose()

4. 进程内事件总线 — app/core/event_bus.py

@bus.subscribe 注册消费者；dispatch 逐个订阅者独立事务执行；通过 sys_event_consume_log(event_id, consumer) 唯一键实现幂等消费。

"""进程内事件总线：订阅注册 + 幂等消费（`sys_event_consume_log` 唯一键）。

一期为进程内实现；Transport 抽象预留二期 RabbitMQ 替换点。
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any

from sqlalchemy.exc import IntegrityError

from app.core.context import get_trace_id
from app.core.logging import logger

Handler = Callable[["DomainEvent"], Awaitable[None]]


@dataclass
class DomainEvent:
    event_type: str
    payload: dict[str, Any] = field(default_factory=dict)
    aggregate_type: str = ""
    aggregate_id: str = ""
    event_id: int | None = None
    trace_id: str | None = None

    def to_envelope_json(self) -> dict[str, Any]:
        return {
            "event_type": self.event_type,
            "payload": self.payload,
            "aggregate_type": self.aggregate_type,
            "aggregate_id": self.aggregate_id,
            "trace_id": self.trace_id,
        }


@dataclass
class Consumer:
    name: str
    handler: Handler


class EventBus:
    """单进程事件总线。订阅方在模块 `events.py` 中通过装饰器注册。"""

    def __init__(self) -> None:
        self._handlers: dict[str, list[Consumer]] = {}

    def subscribe(self, event_type: str, consumer: str | None = None):
        """`@bus.subscribe("OrderPaid", consumer="identity.points")`"""

        def deco(fn: Handler) -> Handler:
            name = consumer or fn.__name__
            self._handlers.setdefault(event_type, []).append(Consumer(name=name, handler=fn))
            logger.debug(f"event subscribe: {event_type} <- {name}")
            return fn

        return deco

    def registered(self) -> dict[str, list[str]]:
        return {k: [c.name for c in v] for k, v in self._handlers.items()}

    async def _consume(self, session, event: DomainEvent, consumer: Consumer) -> bool:
        """幂等消费：先占坑（唯一键），再执行 handler。返回是否实际执行。"""
        from app.core.outbox import SysEventConsumeLog

        if event.event_id is not None:
            try:
                async with session.begin_nested():
                    session.add(
                        SysEventConsumeLog(event_id=event.event_id, consumer=consumer.name)
                    )
                    await session.flush()
            except IntegrityError:
                logger.debug(f"event {event.event_id} already consumed by {consumer.name}")
                return False
        await consumer.handler(event)
        return True

    async def dispatch(self, event: DomainEvent, session=None) -> int:
        """分发事件；返回成功执行的订阅者数量。"""
        handlers = self._handlers.get(event.event_type, [])
        if not handlers:
            logger.debug(f"event {event.event_type} no subscriber")
            return 0
        if event.trace_id is None:
            event.trace_id = get_trace_id()

        from app.core.database import session_scope

        executed = 0
        for c in handlers:
            try:
                if session is not None:
                    if await self._consume(session, event, c):
                        executed += 1
                else:
                    # 每个订阅者独立事务，隔离失败
                    async with session_scope() as s:
                        if await self._consume(s, event, c):
                            executed += 1
            except Exception as exc:  # 单个订阅者失败不影响其它
                logger.exception(f"consumer {c.name} failed for {event.event_type}: {exc}")
        return executed


bus = EventBus()

5. Outbox 发件箱 — app/core/outbox.py

publish_in_tx 与业务写同事务落 sys_event_outbox；OutboxDispatcher 用 Redis 锁保证单投递者，SELECT … FOR UPDATE SKIP LOCKED 批量拉取，失败指数退避、超限进入死信，支持 replay_failed 重放。

"""Outbox：本地事务一致性发件箱 + 投递器 + 幂等消费日志模型。

- `publish_in_tx()`：与业务写**同事务**写入 `sys_event_outbox`（不 commit）。
- `OutboxDispatcher.dispatch_once()`：
  BEGIN → SELECT … status=0 AND retry<max AND next_retry_at<=NOW()
        ORDER BY id LIMIT :batch FOR UPDATE SKIP LOCKED
  → 逐条分发 → 成功 status=1 / 失败指数退避 / retry>=max→status=2 死信 → COMMIT
- `replay_failed()`：重置死信/失败事件以重放。
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timedelta

from sqlalchemy import (
    JSON,
    BigInteger,
    DateTime,
    Index,
    Integer,
    SmallInteger,
    String,
    UniqueConstraint,
    func,
    text,
)
from sqlalchemy.orm import Mapped, mapped_column

from app.core.config import settings
from app.core.context import get_trace_id
from app.core.database import Base, SessionLocal, session_scope
from app.core.event_bus import DomainEvent, bus
from app.core.idempotency import DistributedLock, LockNotAcquired
from app.core.logging import logger


class SysEventOutbox(Base):
    __tablename__ = "sys_event_outbox"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    event_type: Mapped[str] = mapped_column(String(64), nullable=False, comment="事件类型")
    aggregate_type: Mapped[str | None] = mapped_column(String(32), comment="聚合根类型")
    aggregate_id: Mapped[str | None] = mapped_column(String(64), comment="聚合根ID")
    payload: Mapped[dict | None] = mapped_column(JSON, comment="事件载荷 JSON")
    status: Mapped[int] = mapped_column(
        SmallInteger, nullable=False, default=0, server_default=text("0"), comment="0待投递 1已投递 2失败"
    )
    retry: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"), comment="重试次数"
    )
    next_retry_at: Mapped[datetime | None] = mapped_column(DateTime, comment="[补充] 下次重试时间")
    last_error: Mapped[str | None] = mapped_column(String(512), comment="[补充] 最近错误")
    trace_id: Mapped[str | None] = mapped_column(String(64), comment="链路ID")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), comment="创建时间")
    updated_at: Mapped[datetime | None] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(), comment="[补充] 更新时间"
    )

    __table_args__ = (Index("idx_status", "status", "id"),)


class SysEventConsumeLog(Base):
    __tablename__ = "sys_event_consume_log"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    event_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="事件ID")
    consumer: Mapped[str] = mapped_column(String(64), nullable=False, comment="消费者标识")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    __table_args__ = (UniqueConstraint("event_id", "consumer", name="uk_event_consumer"),)


async def publish_in_tx(
    session,
    event_type: str,
    payload: dict | None = None,
    aggregate_type: str = "",
    aggregate_id: str | int = "",
) -> SysEventOutbox:
    """与业务写同事务写入 Outbox。**调用方负责 commit**。"""
    row = SysEventOutbox(
        event_type=event_type,
        aggregate_type=aggregate_type or None,
        aggregate_id=str(aggregate_id) if aggregate_id != "" else None,
        payload=payload or {},
        status=0,
        retry=0,
        trace_id=get_trace_id(),
    )
    # 立即 flush 以拿到自增 id（供消费日志使用）
    session.add(row)
    await session.flush()
    return row


class OutboxDispatcher:
    """单投递者：Redis 锁 + FOR UPDATE SKIP LOCKED 批量拉取。"""

    def __init__(self, batch_size: int | None = None, max_retry: int | None = None):
        self.batch_size = batch_size or settings.outbox_batch_size
        self.max_retry = max_retry or settings.outbox_max_retry
        self._running = False

    async def dispatch_once(self) -> int:
        """投递一批，返回处理条数。"""
        lock = DistributedLock("outbox:dispatch", ttl=30)
        if not await lock.acquire():
            return 0
        handled = 0
        try:
            async with SessionLocal() as session:
                async with session.begin():
                    rows = (
                        await session.execute(
                            text(
                                """
                                SELECT id FROM sys_event_outbox
                                WHERE status = 0 AND retry < :max_retry
                                  AND (next_retry_at IS NULL OR next_retry_at <= NOW())
                                ORDER BY id
                                LIMIT :batch
                                FOR UPDATE SKIP LOCKED
                                """
                            ),
                            {"max_retry": self.max_retry, "batch": self.batch_size},
                        )
                    ).scalars().all()
                    ids = list(rows)
                    if not ids:
                        return 0

                    events: list[DomainEvent] = []
                    for eid in ids:
                        row = await session.get(SysEventOutbox, eid)
                        events.append(
                            DomainEvent(
                                event_type=row.event_type,
                                payload=row.payload or {},
                                aggregate_type=row.aggregate_type or "",
                                aggregate_id=row.aggregate_id or "",
                                event_id=row.id,
                                trace_id=row.trace_id,
                            )
                        )

                # 事务外分发（handler 独立事务）
                for ev, eid in zip(events, ids):
                    try:
                        await bus.dispatch(ev)
                        async with SessionLocal() as s:
                            async with s.begin():
                                row = await s.get(SysEventOutbox, eid)
                                row.status = 1
                                row.last_error = None
                    except Exception as exc:
                        await self._mark_failed(eid, exc)
                    handled += 1
        except LockNotAcquired:
            pass
        finally:
            await lock.release()
        return handled

    async def _mark_failed(self, event_id: int, exc: Exception) -> None:
        msg = str(exc)[:500]
        async with SessionLocal() as s:
            async with s.begin():
                row = await s.get(SysEventOutbox, event_id)
                if row is None:
                    return
                row.retry += 1
                row.last_error = msg
                if row.retry >= self.max_retry:
                    row.status = 2  # 死信
                else:
                    row.next_retry_at = datetime.now() + timedelta(seconds=2 ** row.retry)
        logger.warning(f"outbox event {event_id} failed: {msg}")

    async def replay_failed(self) -> int:
        """重置失败事件（status=2）与到期的失败事件，供重新投递。"""
        async with session_scope() as s:
            result = await s.execute(
                text(
                    "UPDATE sys_event_outbox SET status=0, retry=0, next_retry_at=NULL "
                    "WHERE status=2"
                )
            )
            return int(result.rowcount or 0)

    async def run_forever(self, interval: float | None = None) -> None:
        self._running = True
        interval = interval or settings.outbox_poll_interval
        logger.info(f"OutboxDispatcher started, interval={interval}s")
        while self._running:
            try:
                n = await self.dispatch_once()
                if n:
                    logger.debug(f"outbox dispatched {n} events")
            except Exception as exc:
                logger.exception(f"outbox dispatch error: {exc}")
            await asyncio.sleep(interval)

    def stop(self) -> None:
        self._running = False


dispatcher = OutboxDispatcher()


async def replay_failed_events() -> int:
    return await dispatcher.replay_failed()


__all__ = [
    "OutboxDispatcher",
    "SysEventConsumeLog",
    "SysEventOutbox",
    "dispatcher",
    "publish_in_tx",
    "replay_failed_events",
]

6. 安全组件 — app/core/security.py

JWT 双令牌（access/refresh，jti 可撤销）；bcrypt 密码哈希；手机号 AES-256-GCM 加密存储 + HMAC-SHA256 指纹列（可建唯一索引检索）+ 脱敏展示。

"""安全组件：JWT（HS256）/ 密码哈希（bcrypt）/ 手机号 AES-256-GCM + HMAC 指纹 + 脱敏。"""

from __future__ import annotations

import base64
import hashlib
import hmac
import os
import time
import uuid
from dataclasses import dataclass
from typing import Any

import bcrypt
import jwt
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from app.core.config import settings
from app.core.exceptions import BizError, ErrorCode


class PasswordHasher:
    @staticmethod
    def hash(password: str) -> str:
        raw = password.encode("utf-8")[:72]
        return bcrypt.hashpw(raw, bcrypt.gensalt(rounds=10)).decode()

    @staticmethod
    def verify(password: str, hashed: str | None) -> bool:
        if not hashed:
            return False
        try:
            return bcrypt.checkpw(password.encode("utf-8")[:72], hashed.encode("utf-8"))
        except ValueError:
            return False


@dataclass
class TokenPair:
    access_token: str
    refresh_token: str
    expires_in: int

    def to_dict(self) -> dict[str, Any]:
        return {
            "accessToken": self.access_token,
            "refreshToken": self.refresh_token,
            "expiresIn": self.expires_in,
        }


class JWTManager:
    """access / refresh 双令牌，jti 支持撤销。type: access|refresh"""

    @staticmethod
    def _encode(payload: dict, expire_seconds: int) -> str:
        now = int(time.time())
        body = {
            **payload,
            "iat": now,
            "exp": now + expire_seconds,
            "jti": uuid.uuid4().hex,
        }
        return jwt.encode(body, settings.jwt_secret, algorithm=settings.jwt_algorithm)

    @classmethod
    def create_tokens(cls, user_id: int, roles: list[str] | None = None, extra: dict | None = None) -> TokenPair:
        base = {"sub": str(user_id), "roles": roles or [], **(extra or {})}
        access_expire = settings.access_token_expire_minutes * 60
        access = cls._encode({**base, "type": "access"}, access_expire)
        refresh = cls._encode(
            {"sub": str(user_id), "type": "refresh"},
            settings.refresh_token_expire_days * 86400,
        )
        return TokenPair(access, refresh, access_expire)

    @classmethod
    def decode(cls, token: str, expected_type: str | None = None) -> dict:
        try:
            payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
        except jwt.ExpiredSignatureError as exc:
            raise BizError(ErrorCode.TOKEN_EXPIRED, "登录已过期，请重新登录") from exc
        except jwt.PyJWTError as exc:
            raise BizError(ErrorCode.UNAUTHORIZED, "无效的令牌") from exc
        if expected_type and payload.get("type") != expected_type:
            raise BizError(ErrorCode.UNAUTHORIZED, "令牌类型不匹配")
        return payload

    @classmethod
    def refresh(cls, refresh_token: str, roles: list[str] | None = None) -> TokenPair:
        payload = cls.decode(refresh_token, expected_type="refresh")
        return cls.create_tokens(int(payload["sub"]), roles)


class MobileCipher:
    """手机号：AES-256-GCM 加密 + HMAC-SHA256 指纹（唯一索引列）+ 脱敏展示。"""

    @staticmethod
    def _key() -> bytes:
        raw = settings.mobile_aes_key
        try:
            key = bytes.fromhex(raw)
        except ValueError:
            key = hashlib.sha256(raw.encode()).digest()
        if len(key) != 32:
            key = hashlib.sha256(key).digest()
        return key

    @classmethod
    def encrypt(cls, mobile: str) -> str:
        nonce = os.urandom(12)
        ct = AESGCM(cls._key()).encrypt(nonce, mobile.encode("utf-8"), None)
        return base64.b64encode(nonce + ct).decode()

    @classmethod
    def decrypt(cls, token: str) -> str:
        raw = base64.b64decode(token)
        nonce, ct = raw[:12], raw[12:]
        return AESGCM(cls._key()).decrypt(nonce, ct, None).decode("utf-8")

    @staticmethod
    def fingerprint(mobile: str) -> str:
        return hmac.new(
            settings.mobile_hmac_key.encode(), mobile.encode(), hashlib.sha256
        ).hexdigest()

    @staticmethod
    def mask(mobile: str) -> str:
        if not mobile or len(mobile) < 7:
            return mobile or ""
        return f"{mobile[:3]}****{mobile[-4:]}"


security = type("Security", (), {"jwt": JWTManager, "pwd": PasswordHasher, "mobile": MobileCipher})

7. 系统接口 — app/api/v1/system.py

健康检查 /health（返回 db/redis 连通性）、能力注册表 /system/capabilities（降级状态可视化）、字典 /system/dicts。

"""系统接口：健康检查 / 能力降级状态 / 字典。"""

from __future__ import annotations

from fastapi import Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import dicts as dicts_module
from app.core.config import settings
from app.core.database import SessionLocal, get_session
from app.core.redis import get_redis
from app.core.response import EnvelopeAPIRouter
from app.integrations.base import registry

health_router = EnvelopeAPIRouter(tags=["系统"])
router = EnvelopeAPIRouter(prefix="/system", tags=["系统"])

VERSION = "0.1.0"


@health_router.get("/health", summary="健康检查")
async def health() -> dict:
    db_ok = redis_ok = False
    db_error = redis_error = None

    try:
        async with SessionLocal() as session:
            await session.execute(text("SELECT 1"))
        db_ok = True
    except Exception as exc:  # noqa: BLE001
        db_error = str(exc)

    try:
        await get_redis().ping()
        redis_ok = True
    except Exception as exc:  # noqa: BLE001
        redis_error = str(exc)

    data: dict = {
        "db": db_ok,
        "redis": redis_ok,
        "env": settings.env,
        "version": VERSION,
        "app": settings.app_name,
    }
    if not db_ok and settings.debug:
        data["dbError"] = db_error
    if not redis_ok and settings.debug:
        data["redisError"] = redis_error
    return data


@router.get("/capabilities", summary="能力注册表与降级状态")
async def capabilities() -> dict:
    return {
        "modes": settings.capability_modes(),
        "registered": registry.registered(),
        "items": await registry.snapshot(),
    }


@router.get("/dicts", summary="全部字典类型")
async def all_dict_types() -> list[dict]:
    return [{"dictType": k, "count": len(v)} for k, v in dicts_module.BUILTIN_DICTS.items()]


@router.get("/dicts/{dict_type}", summary="按类型读取字典项")
async def get_dict_items(
    dict_type: str,
    session: AsyncSession = Depends(get_session),
) -> list[dict]:
    return await dicts_module.get_dict(session, dict_type)

8. 用户模型 — app/modules/identity/models.py（节选）

SysUser 表：手机号存密文 + HMAC 指纹（uk_mobile_hash 唯一索引），积分、会员等级、软删除、租户字段齐备。

class SysUser(Base):
    __tablename__ = "sys_user"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    open_id: Mapped[str | None] = mapped_column(String(64), comment="微信openid")
    union_id: Mapped[str | None] = mapped_column(String(64))
    mobile: Mapped[str | None] = mapped_column(String(255), comment="手机号密文")
    mobile_hash: Mapped[str | None] = mapped_column(String(64), comment="手机号HMAC指纹")
    mobile_mask: Mapped[str | None] = mapped_column(String(20))
    password_hash: Mapped[str | None] = mapped_column(String(128))
    nickname: Mapped[str | None] = mapped_column(String(64))
    avatar: Mapped[str | None] = mapped_column(String(512))
    member_level_id: Mapped[int] = mapped_column(Integer, default=1, server_default=text("1"))
    points: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"))
    status: Mapped[int] = mapped_column(SmallInteger, default=1, server_default=text("1"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime | None] = mapped_column(DateTime, onupdate=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime)
    tenant_id: Mapped[int] = mapped_column(BigInteger, default=0, server_default=text("0"))

    __table_args__ = (
        UniqueConstraint("open_id", name="uk_openid"),
        UniqueConstraint("mobile_hash", name="uk_mobile_hash"),
        Index("idx_mobile", "mobile_hash"),
    )

9. 统一请求层 — frontend/src/utils/request.ts

Axios 拦截器注入 Bearer + X-Store-Id + X-Trace-Id；响应解包信封，code≠0 抛 BizError 并 toast；40100/40101 触发单飞刷新并重放原请求，失败跳登录。

import axios, { type AxiosError, type AxiosRequestConfig, type InternalAxiosRequestConfig } from 'axios'
import { ElMessage } from 'element-plus'
import {
  clearAuth,
  getRefreshToken,
  getStoreId,
  getToken,
  setToken,
} from './auth'

/** 后端统一响应信封：{ code, msg, data, traceId } */
export interface Envelope<T = any> {
  code: number
  msg: string
  data: T
  traceId?: string
}

/** 业务异常（code !== 0），供调用方按 code 分支处理 */
export class BizError extends Error {
  code: number
  data: any
  traceId?: string
  constructor(code: number, msg: string, data: any = null, traceId?: string) {
    super(msg)
    this.name = 'BizError'
    this.code = code
    this.data = data
    this.traceId = traceId
  }
}

export const UNAUTHORIZED = 40100
export const TOKEN_EXPIRED = 40101

const BASE_URL = import.meta.env.VITE_API_BASE || '/api/v1'

/** 当前门店 ID（B 端运营后台使用，C 端默认 0 由后端推断） */
let currentStoreId = getStoreId()
export function setStoreIdHeader(id: number | string | null): void {
  currentStoreId = Number(id || 0)
}

function genTraceId(): string {
  const rnd = Math.random().toString(36).slice(2, 10)
  return `web-${Date.now().toString(36)}-${rnd}`
}

const service = axios.create({
  baseURL: BASE_URL,
  timeout: 20000,
  headers: { 'Content-Type': 'application/json' },
})

service.interceptors.request.use((config: InternalAxiosRequestConfig) => {
  const token = getToken()
  if (token) config.headers.set('Authorization', `Bearer ${token}`)
  if (currentStoreId) config.headers.set('X-Store-Id', String(currentStoreId))
  config.headers.set('X-Trace-Id', genTraceId())
  return config
})

/** 刷新 token 单飞：并发 401 只触发一次 refresh，其余请求复用同一 Promise */
let refreshing: Promise<string> | null = null

async function refreshTokenOnce(): Promise<string> {
  if (!refreshing) {
    refreshing = (async () => {
      const rt = getRefreshToken()
      if (!rt) throw new BizError(UNAUTHORIZED, '登录状态已失效，请重新登录')
      // 走原生 axios，避免再次触发本实例拦截器造成递归
      const resp = await axios.post<Envelope<{ accessToken: string; refreshToken: string }>>(
        `${BASE_URL}/identity/auth/refresh`,
        { refreshToken: rt },
        { timeout: 15000 },
      )
      const body = resp.data
      if (!body || body.code !== 0) {
        throw new BizError(body?.code ?? UNAUTHORIZED, body?.msg || '刷新登录状态失败')
      }
      setToken(body.data.accessToken, body.data.refreshToken)
      return body.data.accessToken
    })().finally(() => {
      refreshing = null
    })
  }
  return refreshing
}

function redirectToLogin(): void {
  clearAuth()
  const isAdmin = location.pathname.startsWith('/admin')
  const loginPath = isAdmin ? '/admin/login' : '/login'
  if (location.pathname !== loginPath) {
    const redirect = encodeURIComponent(location.pathname + location.search)
    location.href = `${loginPath}?redirect=${redirect}`
  }
}

/** 解包信封：code === 0 直接返回 data，否则抛 BizError 并 toast */
function unwrap<T>(body: Envelope<T>, silent: boolean): T {
  if (!body || typeof body.code === 'undefined') {
    throw new BizError(-1, '响应格式异常', body)
  }
  if (body.code === 0) return body.data
  const err = new BizError(body.code, body.msg || '请求失败', body.data, body.traceId)
  if (!silent) ElMessage.error(err.message)
  throw err
}

service.interceptors.response.use(
  async (response) => {
    const config = response.config as InternalAxiosRequestConfig & { silent?: boolean; _retried?: boolean }
    const body = response.data as Envelope
    if (body && (body.code === TOKEN_EXPIRED || body.code === UNAUTHORIZED)) {
      if (!config._retried) {
        config._retried = true
        try {
          const fresh = await refreshTokenOnce()
          config.headers.set('Authorization', `Bearer ${fresh}`)
          return service.request(config)
        } catch {
          redirectToLogin()
          return Promise.reject(new BizError(TOKEN_EXPIRED, '登录状态已失效，请重新登录'))
        }
      }
      redirectToLogin()
      return Promise.reject(new BizError(TOKEN_EXPIRED, '登录状态已失效，请重新登录'))
    }
    return unwrap(body, !!config.silent)
  },
  (error: AxiosError) => {
    const status = error.response?.status
    if (status === 401) {
      redirectToLogin()
      return Promise.reject(new BizError(UNAUTHORIZED, '登录状态已失效，请重新登录'))
    }
    const respBody = error.response?.data as Envelope | undefined
    if (respBody && typeof respBody.code !== 'undefined') {
      const err = new BizError(respBody.code, respBody.msg || '请求失败', respBody.data, respBody.traceId)
      ElMessage.error(err.message)
      return Promise.reject(err)
    }
    const msg =
      error.code === 'ECONNABORTED' ? '请求超时，请稍后重试' : error.message || '网络异常，请检查服务是否启动'
    ElMessage.error(msg)
    return Promise.reject(new BizError(-1, msg))
  },
)

export interface RequestOptions extends AxiosRequestConfig {
  /** 静默模式：业务错误不弹 toast */
  silent?: boolean
}

export const request = {
  get<T = any>(url: string, params?: any, options: RequestOptions = {}): Promise<T> {
    return service.get(url, { params, ...options }) as unknown as Promise<T>
  },
  post<T = any>(url: string, data?: any, options: RequestOptions = {}): Promise<T> {
    return service.post(url, data, options) as unknown as Promise<T>
  },
  put<T = any>(url: string, data?: any, options: RequestOptions = {}): Promise<T> {
    return service.put(url, data, options) as unknown as Promise<T>
  },
  patch<T = any>(url: string, data?: any, options: RequestOptions = {}): Promise<T> {
    return service.patch(url, data, options) as unknown as Promise<T>
  },
  delete<T = any>(url: string, options: RequestOptions = {}): Promise<T> {
    return service.delete(url, options) as unknown as Promise<T>
  },
}

export default service

10. 登录态管理 — frontend/src/stores/auth.ts

Pinia store：登录编排、token 落盘、登出清理；登录后联动拉取用户资料。

/** 当前登录态（token / 登录登出编排） */
import { defineStore } from 'pinia'
import * as authApi from '@/api/auth'
import { clearAuth, getRefreshToken, getToken, setToken } from '@/utils/auth'
import { useUserStore } from './user'

interface State {
  token: string
  refreshToken: string
  logging: boolean
}

export const useAuthStore = defineStore('auth', {
  state: (): State => ({
    token: getToken(),
    refreshToken: getRefreshToken(),
    logging: false,
  }),
  getters: {
    isLogin: (s): boolean => !!s.token,
  },
  actions: {
    /** 账号密码登录（C/B 端共用） */
    async login(payload: { account: string; password: string }) {
      this.logging = true
      try {
        const result = await authApi.login(payload)
        this.applyToken(result.accessToken, result.refreshToken)
        const userStore = useUserStore()
        await userStore.fetchProfile()
        return result
      } finally {
        this.logging = false
      }
    },

    applyToken(token: string, refreshToken?: string) {
      this.token = token
      this.refreshToken = refreshToken || this.refreshToken
      setToken(token, refreshToken)
    },

    async logout() {
      try {
        await authApi.logout()
      } catch {
        // 忽略登出接口异常，本地态必须清理
      }
      this.reset()
    },

    reset() {
      this.token = ''
      this.refreshToken = ''
      clearAuth()
      useUserStore().reset()
    },
  },
})

11. 路由与守卫 — frontend/src/router/index.ts

Mall/Admin/Common 三组路由；全局守卫设置标题，requiresAuth 页面未登录跳转并携带 redirect。

import { createRouter, createWebHistory, type RouteRecordRaw } from 'vue-router'
import { mallRoutes } from './routes/mall'
import { adminRoutes } from './routes/admin'
import { getToken } from '@/utils/auth'

export const commonRoutes: RouteRecordRaw[] = [
  {
    path: '/login',
    name: 'login',
    component: () => import('@/views/common/Login.vue'),
    meta: { title: '会员登录', requiresAuth: false },
  },
  {
    path: '/403',
    name: 'forbidden',
    component: () => import('@/views/common/Forbidden.vue'),
    meta: { title: '无访问权限', requiresAuth: false },
  },
  {
    path: '/:pathMatch(.*)*',
    name: 'not-found',
    component: () => import('@/views/common/NotFound.vue'),
    meta: { title: '页面不存在', requiresAuth: false },
  },
]

const router = createRouter({
  history: createWebHistory(),
  routes: [...mallRoutes, ...adminRoutes, ...commonRoutes],
  scrollBehavior: () => ({ top: 0 }),
})

const APP_TITLE = import.meta.env.VITE_APP_TITLE || 'SmartShop'

router.beforeEach((to) => {
  const title = to.meta.title as string | undefined
  document.title = title ? `${title} - ${APP_TITLE}` : APP_TITLE

  if (to.meta.requiresAuth && !getToken()) {
    const isAdmin = to.path.startsWith('/admin')
    return {
      path: isAdmin ? '/admin/login' : '/login',
      query: { redirect: to.fullPath },
    }
  }
  return true
})

export default router

12. 全局主题变量 — frontend/src/styles/variables.scss

蓝色主色 #1677ff 完整色阶 + B 端侧边栏色板；通过 CSS 变量 --ss-primary 与 Element Plus --el-color-primary 全局生效。

// SCSS 变量：供 .scss 文件 @use 使用；组件内部优先使用 CSS 变量（--ss-* / --el-*）

// 蓝色主色（C 端商城）
$mall-primary: #1677ff;
$mall-primary-light: #4d94ff;
$mall-primary-lighter: #8ab8ff;
$mall-primary-pale: #e6f0ff;
$mall-primary-dark: #0e5ce0;

// 中性色
$text-main: #333333;
$text-sub: #666666;
$text-muted: #999999;
$border-color: #e8e8e8;
$page-bg: #f4f4f4;
$card-bg: #ffffff;

// B 端后台
$admin-sidebar-bg: #0b2b45;
$admin-sidebar-active: #1677ff;
$admin-sidebar-width: 220px;
$admin-header-height: 56px;

// 圆角与阴影
$radius-sm: 4px;
$radius-md: 8px;
$radius-lg: 12px;
$shadow-card: 0 2px 8px rgba(0, 0, 0, 0.08);
$shadow-hover: 0 6px 18px rgba(22, 119, 255, 0.18);

13. 首页演示数据 — frontend/src/views/mall/Home.vue（节选）

金刚区/轮播/秒杀/瀑布流分区；演示商品引用 storage/images 实物图，P1 接入 /api/v1/product 后替换为真实接口数据。

/** 演示商品（P1 接入商品域后替换为 /product/spu 与推荐接口数据） */
const demoProducts: ProductCardItem[] = [
  { id: 5001, title: '乐事薯片 原味 大包装 145g', price: 12.90, marketPrice: 15.90, sales: 1286, tags: ['识别即买'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5001.png' },
  { id: 5002, title: '可口可乐 汽水 330ml*6罐', price: 19.90, marketPrice: 24.90, sales: 3562, tags: ['直播同款'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5002.png' },
  { id: 5003, title: '农夫山泉 天然水 550ml*12瓶', price: 24.90, marketPrice: 29.90, sales: 8931, tags: ['门店有货'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5003.png' },
  { id: 5004, title: '蓝月亮 深层洁净洗衣液 3kg', price: 39.90, marketPrice: 49.90, sales: 645, tags: ['识别即买'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5004.png' },
  { id: 5005, title: '星辰 氮化镓快充充电器 65W', price: 89.90, marketPrice: 129.00, sales: 432, tags: ['直播同款'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5005.png' },
  { id: 5006, title: '星辰 无线蓝牙耳机 主动降噪', price: 199.00, marketPrice: 299.00, sales: 218, tags: ['门店有货'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5006.png' },
  { id: 5007, title: '花漾 玻尿酸保湿面霜 50g', price: 129.00, marketPrice: 169.00, sales: 567, tags: ['识别即买'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5007.png' },
  { id: 5008, title: '婴儿纸尿裤 L码 68片', price: 89.00, marketPrice: 109.00, sales: 1024, tags: ['门店有货'], storeName: 'SmartShop 旗舰店', image: '/static/images/spu-5008.png' },
]

const feed: ProductCardItem[] = Array.from({ length: 15 }).map((_, i) => demoProducts[i % 8])

const seckill = feed.slice(0, 5)

1. 运行结果截图

后端（127.0.0.1:8000）与前端（127.0.0.1:5173）同时启动后，C 端首页实际运行效果：首屏为蓝色主题轮播、金刚区入口与限时秒杀实物商品图；下滑为「猜你喜欢」瀑布流推荐与页脚。

2. 启动命令

见「安装步骤」第 3 节。启动后访问：

• C 端商城：http://127.0.0.1:5173
• B 端后台：http://127.0.0.1:5173/admin/login
• 接口文档：http://127.0.0.1:8000/docs
• 健康检查：http://127.0.0.1:8000/api/v1/health

3. 演示账号（seed_demo.py 生成）

4. 关键配置说明

5. 安全说明

• 手机号不落明文：AES-256-GCM 密文 + HMAC 指纹列用于检索与唯一约束，列表展示脱敏（138****8000）
• 密码使用 bcrypt（rounds=10）；JWT 双令牌，refresh 14 天、access 120 分钟，jti 支持撤销
• 支付金额一律服务端重算（Decimal ROUND_HALF_UP），回调验签 + 金额一致性校验
• 所有业务写通过 publish_in_tx 与事件同事务落库，保证状态与消息最终一致

6. 常见问题

• /health 返回 db:false —— 检查 MySQL 是否启动、.env 密码是否正确；服务会进入降级模式但不崩溃
• 图片 404 —— 确认 storage/images 下文件存在，后端 /static 由 StaticFiles 挂载到 storage/
• Celery 在 Windows 报错 —— 必须使用 -P solo 池；测试可设 TASKS_ALWAYS_EAGER=true 同步执行
• 前端 401 循环 —— 检查 JWT_SECRET 是否前后一致；刷新接口走 /identity/auth/refresh，避免拦截器递归
• npm 脚本执行被策略拦截 —— PowerShell 下使用 npm.cmd 运行

十阶段计划（每阶段独立可验证）

二期演进方向

• 消息中间件：事件 Transport 抽象已预留，可平滑替换 RabbitMQ / Kafka
• 对象存储：storage 能力切 real 即接入 MinIO / OSS，本地存储零改动保留
• 向量检索：vector 能力切 real 接入 Milvus，替换 ChromaDB PersistentClient
• 全文检索：商品搜索可由 ngram 迁移至 Elasticsearch / OpenSearch
• GPU 内容生产：avatar/video/live 三子域队列已隔离（live>video>avatar），接 GPU 后启用真实引擎
• 微服务拆分：模块化单体按子域边界拆分，事件契约（Outbox 载荷）即服务间契约