feat(market): 市场分类字典 + 商品标识 slug 规范化

市场分类字典（admin 端弹窗维护，已运行生效）：
- 新增 MarketCategory 模型 + 0050 迁移 + seed
- categories.py 重构为数据库驱动（list/valid/CRUD）
- admin 分类管理端点：增删改/排序/启停；商品列表支持分类筛选

商品标识 slug 规范化（修复云超服市场安装下载链路）：
- 新增 _normalize_slug：标识不含 '/' 自动补 pineagents/ 前缀，
  避免单段 slug 在 owner/name 拆分时产生 undefined
- admin_create_item：创建时 slug 规范化
- admin_update_item：修复标识首次提交后再次编辑无效
  （此前 handler 未把 slug 传给 update_item）

配套：存量单段 slug 数据已批量规范化为 pineagents/<slug>

alembic/versions/0050_market_categories.py

@@ -0,0 +1,42 @@
+"""市场分类字典：skill/plugin/app 各自独立的分类维护表（admin 端弹窗维护）。
+
+- market_categories：分类字典（item_type + key 唯一），技能分类与前端技能
+  市场下拉固定分类（应用/工程开发/数据研究/文档办公/设计创作/自动化集成/
+  产品管理）按 key 对齐；admin 端可在线增删改排序/启停。
+
+Revision ID: 0050
+Revises: 0049
+Create Date: 2026-09-03
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0050"
+down_revision = "0049"
+branch_labels = None
+depends_on = None
+
+S = sa.String
+
+
+def upgrade() -> None:
+    op.create_table(
+        "market_categories",
+        sa.Column("id", S(), primary_key=True),          # mc_<hex>
+        sa.Column("item_type", S(), nullable=True, server_default="skill"),
+        sa.Column("key", S(), nullable=True, server_default=""),          # 稳定编码（存入 market_items.category）
+        sa.Column("label_zh", S(), nullable=True, server_default=""),
+        sa.Column("label_en", S(), nullable=True, server_default=""),
+        sa.Column("sort", sa.Integer(), nullable=True, server_default="0"),
+        sa.Column("enabled", sa.Integer(), nullable=True, server_default="1"),
+        sa.Column("created_at", S(), nullable=True, server_default=""),
+        sa.Column("updated_at", S(), nullable=True, server_default=""),
+    )
+    op.create_index("ix_market_categories_item_type", "market_categories", ["item_type"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_market_categories_item_type", table_name="market_categories")
+    op.drop_table("market_categories")

app/infrastructure/models.py

@@ -1341,6 +1341,25 @@ class MarketItem(Base):
     updated_at: Mapped[str] = mapped_column(String, default="")
 
 
+class MarketCategory(Base):
+    """市场分类字典（skill / plugin / app 各自独立，admin 端弹窗维护）。
+
+    分类在「market_items.category」中以稳定编码 ``key`` 存储；
+    展示名分中英文（label_zh / label_en），可在线增删改排序。
+    """
+    __tablename__ = "market_categories"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)                    # mc_<hex>
+    item_type: Mapped[str] = mapped_column(String, default="skill", index=True)  # skill|plugin|app
+    key: Mapped[str] = mapped_column(String, default="")                         # 稳定编码（存入 market_items.category）
+    label_zh: Mapped[str] = mapped_column(String, default="")
+    label_en: Mapped[str] = mapped_column(String, default="")
+    sort: Mapped[int] = mapped_column(Integer, default=0)
+    enabled: Mapped[int] = mapped_column(Integer, default=1)
+    created_at: Mapped[str] = mapped_column(String, default="")
+    updated_at: Mapped[str] = mapped_column(String, default="")
+
+
 class MarketPurchaseOrder(Base):
     """市场购买订单（插件 / 应用定价消费，微信支付 V3，复用充值 mp 扫码→小程序支付）。
 

app/infrastructure/seed.py

@@ -17,6 +17,7 @@ from .models import (
     FinanceRecord,
     InvestmentIntent,
     InvestorPreference,
+    MarketCategory,
     Message,
     Notification,
     OpcProfile,
@@ -270,6 +271,7 @@ async def seed_data(session: AsyncSession) -> None:
         await _seed_task_categories(session, now)
         await _seed_port_dashboards(session, now)
         await _seed_market(session, now)
+        await _seed_market_categories(session, now)
         await _seed_port_pages(session, now)
         await _seed_org_members(session, now)
         await _seed_carrier_park_binding(session, now)
@@ -326,6 +328,7 @@ async def seed_data(session: AsyncSession) -> None:
     await _seed_opc_business(session, now)
     await _seed_port_dashboards(session, now)
     await _seed_market(session, now)
+    await _seed_market_categories(session, now)
     await _seed_port_pages(session, now)
     await _seed_org_members(session, now)
     await _seed_carrier_park_binding(session, now)
@@ -615,6 +618,15 @@ async def _seed_market(session: AsyncSession, now: str) -> None:
     ])
 
 
+# ---------------------------------------------------------------------------
+# 市场分类（skill/plugin/app 各类型独立，admin 端弹窗在线维护）
+# ---------------------------------------------------------------------------
+async def _seed_market_categories(session: AsyncSession, now: str) -> None:
+    """幂等补种市场分类字典（空表才写；此后由 admin 端维护）。"""
+    from ..market.categories import ensure_default_categories
+    await ensure_default_categories(session, now)
+
+
 # ---------------------------------------------------------------------------
 # 端口子页面演示数据（企业人才/载体OPC与服务/服务商订单与服务/政务OPC·政策·载体·数据）
 # ---------------------------------------------------------------------------

app/market/categories.py

@@ -1,48 +1,122 @@
 # -*- coding: utf-8 -*-
-"""市场分类字典（三类商品各自的分类，与前端 / qwenpaw 外部源对齐）。"""
+"""市场分类字典（skill / plugin / app 三类商品各自的分类）。
+
+分类定义存储在 ``market_categories`` 表（admin 端弹窗在线维护），
+本模块提供读取 / 校验辅助函数，并保留默认分类常量用于幂等种子。
+"""
 from __future__ import annotations
 
-MARKET_CATEGORIES: dict[str, list[str]] = {
-    "skill": [
-        "productivity",   # 效率
-        "automation",     # 自动化
-        "media",          # 内容创作
-        "developer",      # 开发
-        "analysis",       # 数据分析
-        "general",        # 通用
-    ],
-    "plugin": [
-        "app",            # 应用
-        "productivity",   # 效率
-        "developer",      # 开发
-        "automation",     # 自动化
-        "media",          # 内容创作
-        "general",        # 通用
-    ],
-    "app": [
-        "app",            # 应用
-        "productivity",   # 效率
-        "developer",      # 开发
-        "automation",     # 自动化
-        "media",          # 内容创作
-        "general",        # 通用
-    ],
-}
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..infrastructure.models import MarketCategory
 
 VALID_TYPES = ("skill", "plugin", "app")
 
+# 默认分类种子（首次建库/空表时写入；此后由 admin 端在线维护）。
+# key 为稳定编码（存入 market_items.category）；技能分类与前端（agent-desktop
+# 技能市场）按 key 对齐，即截图中的固定分类：应用/工程开发/数据研究/文档办公/
+# 设计创作/自动化集成/产品管理。
+DEFAULT_CATEGORIES: dict[str, list[tuple[str, str, str]]] = {
+    "skill": [
+        # (key, label_zh, label_en)
+        ("app", "应用", "Apps"),
+        ("engineering-development", "工程开发", "Engineering"),
+        ("data-research", "数据研究", "Data & Research"),
+        ("document-office", "文档办公", "Docs & Office"),
+        ("design-creation", "设计创作", "Design"),
+        ("automation-integration", "自动化集成", "Automation"),
+        ("product-management", "产品管理", "Product"),
+    ],
+    "plugin": [
+        ("app", "应用", "Apps"),
+        ("productivity", "效率", "Productivity"),
+        ("developer", "开发", "Developer"),
+        ("automation", "自动化", "Automation"),
+        ("media", "内容创作", "Media"),
+        ("general", "通用", "General"),
+    ],
+    "app": [
+        ("app", "应用", "Apps"),
+        ("productivity", "效率", "Productivity"),
+        ("developer", "开发", "Developer"),
+        ("automation", "自动化", "Automation"),
+        ("media", "内容创作", "Media"),
+        ("general", "通用", "General"),
+    ],
+}
+
 
 def valid_type(item_type: str) -> bool:
     return item_type in VALID_TYPES
 
 
-def valid_category(item_type: str, category: str) -> bool:
+def _row(r: MarketCategory) -> dict:
+    return {
+        "id": r.id,
+        "item_type": r.item_type,
+        "key": r.key,
+        "label_zh": r.label_zh,
+        "label_en": r.label_en,
+        "sort": r.sort,
+        "enabled": bool(r.enabled),
+        "created_at": r.created_at,
+        "updated_at": r.updated_at,
+    }
+
+
+async def list_categories(
+    db: AsyncSession, item_type: str | None = None, lang: str = "zh",
+) -> dict:
+    """分类字典：默认返回全部类型；指定 item_type 时返回该类型的分类。
+
+    返回 ``{type?, items: [{key, label_zh, label_en, sort, enabled}]}``。
+    ``lang`` 影响返回中的 ``label`` 字段（zh/en），供前端直接展示。
+    """
+    rows = (
+        await db.session.execute(
+            select(MarketCategory)
+            .where(MarketCategory.item_type == item_type if item_type else True)
+            .order_by(MarketCategory.sort.asc(), MarketCategory.created_at.asc())
+        )
+    ).scalars().all()
+    items = [_row(r) for r in rows]
+    lang_key = "zh" if str(lang).lower().startswith("zh") else "en"
+    for it in items:
+        it["label"] = it["label_zh"] if lang_key == "zh" else it["label_en"]
+    if item_type and valid_type(item_type):
+        return {"type": item_type, "items": items}
+    return {"types": VALID_TYPES, "items": items}
+
+
+async def valid_category(db: AsyncSession, item_type: str, category: str) -> bool:
+    """校验商品分类是否属于该类型的分类字典（空分类视为合法=未分类）。"""
     if not category:
         return True
-    return category in MARKET_CATEGORIES.get(item_type, [])
+    if not valid_type(item_type):
+        return False
+    row = await db.session.scalar(
+        select(MarketCategory).where(
+            MarketCategory.item_type == item_type,
+            MarketCategory.key == category,
+            MarketCategory.enabled == 1,
+        ).limit(1)
+    )
+    return row is not None
 
 
-def categories(item_type: str | None = None) -> dict:
-    if item_type and valid_type(item_type):
-        return {"type": item_type, "items": MARKET_CATEGORIES.get(item_type, [])}
-    return {"types": VALID_TYPES, "items": MARKET_CATEGORIES}
+async def ensure_default_categories(session: AsyncSession, now: str) -> None:
+    """幂等补种默认分类（仅当 market_categories 表为空时写入）。"""
+    from ..infrastructure.repositories import new_id
+
+    if await session.scalar(select(MarketCategory.id).limit(1)) is not None:
+        return
+    rows: list[MarketCategory] = []
+    for item_type, cats in DEFAULT_CATEGORIES.items():
+        for sort, (key, label_zh, label_en) in enumerate(cats, start=1):
+            rows.append(MarketCategory(
+                id=new_id("mc"), item_type=item_type, key=key,
+                label_zh=label_zh, label_en=label_en, sort=sort,
+                enabled=1, created_at=now, updated_at=now,
+            ))
+    session.add_all(rows)

app/market/routers.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import json
 import logging
+import secrets
 
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
 from fastapi.responses import JSONResponse, RedirectResponse
@@ -16,8 +17,9 @@ from pydantic import BaseModel, Field
 from sqlalchemy import func, select
 
 from ..api.dependencies import Database, get_current_user, get_db, optional_current_user
-from ..infrastructure.models import MarketItem
+from ..infrastructure.models import MarketCategory, MarketItem
 from ..infrastructure.oss import oss
+from ..infrastructure.repositories import utcnow_iso
 from ..rbac import require_permission, require_roles
 from . import categories, service
 
@@ -38,6 +40,21 @@ async def _audit(db, user: dict, action: str, item_id: str, detail: str = "") ->
         await db.session.rollback()
 
 
+def _normalize_slug(slug: str, name: str, fallback: str = "") -> str:
+    """标识规范化：统一为 ``owner/name`` 双段结构。
+
+    - 为空时回退 fallback（有值）或名称 name；
+    - 不含 ``/`` 时自动补 ``pineagents/`` 前缀（平台默认发布方），
+      避免单段 slug 在下载链路（owner/name 拆分）产生 ``undefined``。
+    """
+    s = (slug or fallback or name or "").strip().strip("/")
+    if not s:
+        raise HTTPException(status_code=400, detail="标识不能为空")
+    if "/" not in s:
+        s = f"pineagents/{s}"
+    return s
+
+
 # ===========================================================================
 # 公开读
 # ===========================================================================
@@ -149,8 +166,12 @@ async def market_item_bundle(
 
 
 @router.get("/categories", summary="分类字典")
-async def market_categories(item_type: str | None = None):
-    return categories.categories(item_type)
+async def market_categories(
+    item_type: str | None = None,
+    lang: str = "zh",
+    db: Database = Depends(get_db),
+):
+    return await categories.list_categories(db, item_type, lang)
 
 
 # ===========================================================================
@@ -248,19 +269,40 @@ class ItemUpdateRequest(ItemCreateRequest):
     pass
 
 
+class CategoryCreateRequest(BaseModel):
+    item_type: str = Field(..., description="skill|plugin|app")
+    key: str = Field(..., description="稳定编码（商品分类存储值）")
+    label_zh: str = ""
+    label_en: str = ""
+    sort: int = 0
+
+
+class CategoryUpdateRequest(BaseModel):
+    key: str | None = None
+    label_zh: str | None = None
+    label_en: str | None = None
+    sort: int | None = None
+    enabled: int | None = None
+
+
 @admin_router.get("/items", summary="管理列表（全状态）")
 async def admin_items(
     item_type: str | None = None, status: str | None = None, q: str = "",
-    page: int = 1, page_size: int = 50,
+    category: str = "", page: int = 1, page_size: int = 50,
     db: Database = Depends(get_db),
     _u: dict = Depends(require_permission("menu:admin_market")),
 ):
-    data = await service.list_items(db, item_type=item_type, q=q, status=status or "", page=page, page_size=page_size)
+    data = await service.list_items(db, item_type=item_type, q=q, category=category, status=status or "", page=page, page_size=page_size)
     stmt = select(func.count()).select_from(MarketItem)
     if item_type:
         stmt = stmt.where(MarketItem.item_type == item_type)
     if status:
         stmt = stmt.where(MarketItem.status == status)
+    if category:
+        stmt = stmt.where(MarketItem.category == category)
+    if q:
+        like = f"%{q}%"
+        stmt = stmt.where((MarketItem.name.like(like)) | (MarketItem.description.like(like)) | (MarketItem.slug.like(like)))
     data["total"] = int((await db.session.execute(stmt)).scalar() or 0)
     return data
 
@@ -273,9 +315,12 @@ async def admin_create_item(
 ):
     if not categories.valid_type(body.item_type):
         raise HTTPException(status_code=400, detail="item_type 仅支持 skill/plugin/app")
+    if not await categories.valid_category(db, body.item_type, body.category):
+        raise HTTPException(status_code=400, detail="分类不属于该类型，请先维护分类字典")
+    slug = _normalize_slug(body.slug, body.name)
     try:
         item = await service.create_item(
-            db, item_type=body.item_type, slug=body.slug or body.name, name=body.name,
+            db, item_type=body.item_type, slug=slug, name=body.name,
             publisher_id=user["id"], description=body.description, version=body.version,
             author=body.author, owner=body.owner, icon_url=body.icon_url,
             category=body.category, tags=body.tags, price_fen=body.price_fen, bundle=body.bundle,
@@ -304,9 +349,15 @@ async def admin_update_item(
     db: Database = Depends(get_db),
     user: dict = Depends(require_permission("menu:admin_market")),
 ):
+    existing = await service.get_item(db, item_id)
+    if existing is None:
+        raise HTTPException(status_code=404, detail="商品不存在")
+    if not await categories.valid_category(db, existing["item_type"], body.category):
+        raise HTTPException(status_code=400, detail="分类不属于该类型，请先维护分类字典")
+    slug = _normalize_slug(body.slug, body.name, existing["slug"])
     try:
         item = await service.update_item(
-            db, item_id, name=body.name, description=body.description, version=body.version,
+            db, item_id, slug=slug, name=body.name, description=body.description, version=body.version,
             author=body.author, owner=body.owner, icon_url=body.icon_url,
             category=body.category, tags=body.tags, price_fen=body.price_fen, bundle=body.bundle,
         )
@@ -331,6 +382,109 @@ async def admin_delete_item(
     return {"ok": True}
 
 
+# ---------------------------------------------------------------------------
+# 分类管理（admin 端弹窗维护：增删改 + 排序/启停）
+# ---------------------------------------------------------------------------
+@admin_router.get("/categories", summary="分类列表（admin，含停用项）")
+async def admin_list_categories(
+    item_type: str | None = None,
+    db: Database = Depends(get_db),
+    _u: dict = Depends(require_permission("menu:admin_market")),
+):
+    return await categories.list_categories(db, item_type, "zh")
+
+
+@admin_router.post("/categories", summary="新建分类")
+async def admin_create_category(
+    body: CategoryCreateRequest,
+    db: Database = Depends(get_db),
+    user: dict = Depends(require_permission("menu:admin_market")),
+):
+    if not categories.valid_type(body.item_type):
+        raise HTTPException(status_code=400, detail="item_type 仅支持 skill/plugin/app")
+    key = (body.key or "").strip()
+    if not key:
+        raise HTTPException(status_code=400, detail="key（分类编码）不能为空")
+    dup = await db.session.scalar(
+        select(MarketCategory.id).where(
+            MarketCategory.item_type == body.item_type,
+            MarketCategory.key == key,
+        ).limit(1)
+    )
+    if dup:
+        raise HTTPException(status_code=400, detail="该分类编码已存在")
+    now = utcnow_iso()
+    row = MarketCategory(
+        id=f"mc_{secrets.token_hex(8)}", item_type=body.item_type, key=key,
+        label_zh=body.label_zh or "", label_en=body.label_en or "",
+        sort=body.sort, enabled=1, created_at=now, updated_at=now,
+    )
+    db.session.add(row)
+    await db.session.commit()
+    await _audit(db, user, "market.category.create", row.id, f"{body.item_type}:{key}")
+    return categories._row(row)
+
+
+@admin_router.put("/categories/{cat_id}", summary="编辑分类")
+async def admin_update_category(
+    cat_id: str, body: CategoryUpdateRequest,
+    db: Database = Depends(get_db),
+    user: dict = Depends(require_permission("menu:admin_market")),
+):
+    row = await db.session.get(MarketCategory, cat_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="分类不存在")
+    if body.key is not None:
+        key = body.key.strip()
+        if not key:
+            raise HTTPException(status_code=400, detail="key（分类编码）不能为空")
+        dup = await db.session.scalar(
+            select(MarketCategory.id).where(
+                MarketCategory.item_type == row.item_type,
+                MarketCategory.key == key,
+                MarketCategory.id != cat_id,
+            ).limit(1)
+        )
+        if dup:
+            raise HTTPException(status_code=400, detail="该分类编码已存在")
+        row.key = key
+    if body.label_zh is not None:
+        row.label_zh = body.label_zh
+    if body.label_en is not None:
+        row.label_en = body.label_en
+    if body.sort is not None:
+        row.sort = body.sort
+    if body.enabled is not None:
+        row.enabled = 1 if body.enabled else 0
+    row.updated_at = utcnow_iso()
+    await db.session.commit()
+    await _audit(db, user, "market.category.update", cat_id, row.key)
+    return categories._row(row)
+
+
+@admin_router.delete("/categories/{cat_id}", summary="删除分类")
+async def admin_delete_category(
+    cat_id: str,
+    db: Database = Depends(get_db),
+    user: dict = Depends(require_permission("menu:admin_market")),
+):
+    row = await db.session.get(MarketCategory, cat_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="分类不存在")
+    used = await db.session.scalar(
+        select(MarketItem.id).where(
+            MarketItem.item_type == row.item_type,
+            MarketItem.category == row.key,
+        ).limit(1)
+    )
+    if used:
+        raise HTTPException(status_code=400, detail="该分类下存在商品，无法删除；可先停用")
+    await db.session.delete(row)
+    await db.session.commit()
+    await _audit(db, user, "market.category.delete", cat_id, row.key)
+    return {"ok": True}
+
+
 @admin_router.post("/items/{item_id}/submit", summary="提交审核 draft→pending")
 async def admin_submit_item(
     item_id: str,