beige/backend/scripts/seed_data.py

"""
scripts/seed_data.py — 初始化北哥 workspace 种子数据
建3账号(admin/supervisor/operator) + workspace_members + 预置素颜霜 + 5条违禁词
幂等：已存在则跳过，可重复跑。
用法：cd backend && python3 scripts/seed_data.py
"""
import json
import os
import sys

# 把 backend/ 加入 path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.core.config import get_settings  # noqa: E402 — path must be set first
from app.core.database import SessionLocal  # noqa: E402
from app.models.user import User  # noqa: E402
from app.models.workspace import Workspace, WorkspaceMember  # noqa: E402
from app.models.product import Product, BannedWord  # noqa: E402
from app.services.auth_service import DEFAULT_SEED_PASSWORD, hash_password  # noqa: E402

get_settings()  # 触发 .env 校验，缺变量早报错


WORKSPACE_NAME = "北哥小红书车间"
WORKSPACE_SLUG = "beige-xhs"

SEED_PASSWORDS = {
    "admin": os.environ.get("CLOVER_SEED_PASSWORD_ADMIN", os.environ.get("CLOVER_SEED_PASSWORD", DEFAULT_SEED_PASSWORD)),
    "supervisor": os.environ.get("CLOVER_SEED_PASSWORD_SUPERVISOR", os.environ.get("CLOVER_SEED_PASSWORD", DEFAULT_SEED_PASSWORD)),
    "operator": os.environ.get("CLOVER_SEED_PASSWORD_OPERATOR", os.environ.get("CLOVER_SEED_PASSWORD", DEFAULT_SEED_PASSWORD)),
}
if get_settings().APP_ENV == "production" and any(p == DEFAULT_SEED_PASSWORD for p in SEED_PASSWORDS.values()):
    raise RuntimeError("Production seed requires CLOVER_SEED_PASSWORD; default password is forbidden")

USERS = [
    {"username": "admin",      "email": "admin@clover.local",      "password": SEED_PASSWORDS["admin"], "role": "admin"},
    {"username": "supervisor", "email": "supervisor@clover.local",  "password": SEED_PASSWORDS["supervisor"], "role": "supervisor"},
    {"username": "operator",   "email": "operator@clover.local",   "password": SEED_PASSWORDS["operator"], "role": "operator"},
]

PRODUCT = {
    "name": "素颜霜（预置样本）",
    "category": "素颜霜",
    "source": "preset",
    "selling_points": json.dumps(["遮瑕提亮", "持妆", "养肤成分"], ensure_ascii=False),
    "style_tone": "素人分享、真实不浮夸",
    "text_angles": json.dumps(["痛点切入", "成分党", "使用场景", "质地肤感", "平价替代"], ensure_ascii=False),
    "custom_prompt": None,
    "is_active": True,
}

BANNED_WORDS = [
    {"word": "美白",  "level": "auto_fix",   "replacement": "提亮肤色"},
    {"word": "祛斑",  "level": "auto_fix",   "replacement": "改善暗沉"},
    {"word": "速效",  "level": "soft_warn",  "replacement": "温和渐进"},
    {"word": "医用",  "level": "hard_block", "replacement": None},
    {"word": "药妆",  "level": "hard_block", "replacement": None},
]


def run():
    db = SessionLocal()
    try:
        # ── workspace ──────────────────────────────────────
        ws = db.query(Workspace).filter(Workspace.slug == WORKSPACE_SLUG).first()
        if not ws:
            ws = Workspace(name=WORKSPACE_NAME, slug=WORKSPACE_SLUG, is_active=True)
            db.add(ws)
            db.flush()
            print(f"[+] workspace 已建: {ws.name} (id={ws.id})")
        else:
            print(f"[=] workspace 已存在: id={ws.id}")

        # ── users + members ────────────────────────────────
        for u_spec in USERS:
            user = db.query(User).filter(User.username == u_spec["username"]).first()
            if not user:
                user = User(
                    username=u_spec["username"],
                    email=u_spec["email"],
                    hashed_password=hash_password(u_spec["password"]),
                    is_active=True,
                    must_change_password=True,
                )
                db.add(user)
                db.flush()
                print(f"[+] user: {user.username}")
            else:
                print(f"[=] user 已存在: {user.username}")

            mem = db.query(WorkspaceMember).filter(
                WorkspaceMember.workspace_id == ws.id,
                WorkspaceMember.user_id == user.id,
            ).first()
            if not mem:
                db.add(WorkspaceMember(workspace_id=ws.id, user_id=user.id, role=u_spec["role"]))
                print(f"    -> 绑定 role={u_spec['role']}")

        # ── 预置素颜霜 ─────────────────────────────────────
        prod = db.query(Product).filter(
            Product.workspace_id == ws.id, Product.name == PRODUCT["name"]
        ).first()
        if not prod:
            prod = Product(workspace_id=ws.id, **PRODUCT)
            db.add(prod)
            print(f"[+] product: {prod.name}")
        else:
            print(f"[=] product 已存在: {prod.name}")

        # ── 违禁词 ─────────────────────────────────────────
        for bw_spec in BANNED_WORDS:
            existing = db.query(BannedWord).filter(
                BannedWord.workspace_id == ws.id, BannedWord.word == bw_spec["word"]
            ).first()
            if not existing:
                db.add(BannedWord(workspace_id=ws.id, **bw_spec))
                print(f"[+] banned_word: {bw_spec['word']} ({bw_spec['level']})")
            else:
                print(f"[=] banned_word 已存在: {bw_spec['word']}")

        # ── user_api_keys：不再预置（倩倩姐2026-06-12拍板：系统埋的key彻底拔干净）──
        # 每个用户登录后到「设置 → API Key」自录自用，录一次即记住。
        # 种子不再从 APIPORTS_KEY 环境变量批量塞 key，避免"免key可用"。

        db.commit()
        print("\n种子数据初始化完成。")
    except Exception as e:
        db.rollback()
        print(f"[ERROR] {e}")
        raise
    finally:
        db.close()


if __name__ == "__main__":
    run()