cjw
aaaac95b53
has_alembic = await conn.execute(...).scalar() 因 await 优先级低于
属性访问，实际解析为 await (conn.execute(...).scalar())——在协程对象
上调 .scalar() 必抛 AttributeError('coroutine' object has no attribute
'scalar')。启动迁移每次容器启动都在第一行炸掉，再被 except 吞成一行
日志（4a5dc61 已改为 logger.exception）。两行均改为
(await conn.execute(...)).scalar()。
已于生产库实测：主库路径（has_alembic=True→upgrade no-op）与临时
历史库路径（stamp+upgrade）全部跑通，临时库已删除。
至此 2026-09-26 事故根因链闭合：启动迁移从未生效 → prod schema 长期
落后于模型 → 新镜像上线即上传 500（缺 product_id）。
Co-Authored-By: Claude Code <noreply@anthropic.com>
293 lines
13 KiB
Python
293 lines
13 KiB
Python
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
from sqlalchemy import text, select
|
||
from shared.database.database import db_manager
|
||
from shared.models.identity import User, Role, Permission, UserRole, RolePermission
|
||
from shared.services.auth_service import get_password_hash
|
||
from shared.config.settings import settings
|
||
from shared.utils.logger import get_logger
|
||
from alembic.config import Config
|
||
from alembic import command
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
_ALEMBIC_INI = Path(__file__).resolve().parents[3] / "alembic.ini"
|
||
|
||
|
||
def _alembic_stamp_head() -> None:
|
||
"""将当前 DB 标记为已到最新版本（基线既有 DB，不执行 SQL）"""
|
||
cfg = Config(str(_ALEMBIC_INI))
|
||
command.stamp(cfg, "head")
|
||
|
||
|
||
def _alembic_upgrade_head() -> None:
|
||
"""执行 alembic 迁移到最新版本"""
|
||
cfg = Config(str(_ALEMBIC_INI))
|
||
command.upgrade(cfg, "head")
|
||
|
||
|
||
async def _run_alembic_migrations() -> None:
|
||
"""以 alembic 管理 schema：既有未纳入管理的 DB 自动 stamp 基线，再 upgrade head
|
||
|
||
- 全新 DB：upgrade head 执行初始迁移，创建全部表
|
||
- 既有已纳入管理：upgrade head 为 no-op
|
||
- 既有但无 alembic_version（历史 DB）：先 stamp head 基线，再 upgrade（no-op）
|
||
"""
|
||
async with db_manager.engine.begin() as conn:
|
||
# 注意括号：await 优先级低于属性访问，await conn.execute(...).scalar()
|
||
# 实际是 await (conn.execute(...).scalar())——会在协程对象上调 .scalar()
|
||
# 直接 AttributeError。必须 (await conn.execute(...)).scalar()。
|
||
# 曾因漏括号让启动迁移自诞生起一次都没跑通过（2026-09-26 事故根因）。
|
||
has_alembic = (await conn.execute(text("SELECT to_regclass('public.alembic_version')"))).scalar()
|
||
if not has_alembic:
|
||
table_count = (await conn.execute(
|
||
text("SELECT count(*) FROM information_schema.tables WHERE table_schema='public' AND table_name <> 'alembic_version'")
|
||
)).scalar()
|
||
if table_count and table_count > 0:
|
||
logger.info("检测到既有 DB 未纳入 alembic 管理，自动 stamp head 作为基线")
|
||
await asyncio.to_thread(_alembic_stamp_head)
|
||
await asyncio.to_thread(_alembic_upgrade_head)
|
||
|
||
DEFAULT_PERMISSIONS = [
|
||
{"code": "view_dashboard", "name": "查看仪表盘", "module": "dashboard"},
|
||
{"code": "view_moldinsight", "name": "使用模具分析", "module": "moldinsight"},
|
||
{"code": "upload_file", "name": "上传文件", "module": "moldinsight"},
|
||
{"code": "view_history", "name": "查看历史记录", "module": "moldinsight"},
|
||
{"code": "view_inventory", "name": "查看库存", "module": "inventory"},
|
||
{"code": "manage_inventory", "name": "管理库存", "module": "inventory"},
|
||
{"code": "view_products", "name": "查看产品", "module": "inventory"},
|
||
{"code": "manage_products", "name": "管理产品", "module": "inventory"},
|
||
{"code": "view_suppliers", "name": "查看供应商", "module": "inventory"},
|
||
{"code": "manage_suppliers", "name": "管理供应商", "module": "inventory"},
|
||
{"code": "view_customers", "name": "查看客户", "module": "inventory"},
|
||
{"code": "manage_customers", "name": "管理客户", "module": "inventory"},
|
||
{"code": "view_finance", "name": "查看财务", "module": "finance"},
|
||
{"code": "manage_receipts", "name": "管理收款", "module": "finance"},
|
||
{"code": "manage_payments", "name": "管理付款", "module": "finance"},
|
||
{"code": "void_finance_transaction", "name": "作废财务单据", "module": "finance"},
|
||
{"code": "view_users", "name": "查看用户", "module": "admin"},
|
||
{"code": "manage_users", "name": "管理用户", "module": "admin"},
|
||
{"code": "manage_roles", "name": "管理角色", "module": "admin"},
|
||
# D17 Human-in-Loop：老师傅经验反馈
|
||
{"code": "view_experience_feedback", "name": "查看老师傅反馈", "module": "moldinsight"},
|
||
{"code": "feedback_experience_hint", "name": "提交方案级反馈", "module": "moldinsight"},
|
||
{"code": "manage_experience_feedback", "name": "管理老师傅反馈", "module": "moldinsight"},
|
||
]
|
||
|
||
DEFAULT_ROLES = [
|
||
{"code": "admin", "name": "管理员", "description": "系统管理员，拥有所有权限", "is_system": True, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_inventory", "manage_inventory", "view_products", "manage_products", "view_suppliers", "manage_suppliers", "view_customers", "manage_customers", "view_finance", "manage_receipts", "manage_payments", "void_finance_transaction", "view_users", "manage_users", "manage_roles", "view_experience_feedback", "feedback_experience_hint", "manage_experience_feedback"]},
|
||
{"code": "user", "name": "普通用户", "description": "普通用户，可使用模具分析和查看库存", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_inventory", "view_products", "view_suppliers", "view_customers", "view_finance", "manage_receipts", "manage_payments"]},
|
||
{"code": "viewer", "name": "只读用户", "description": "只读用户，只能查看数据", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "view_history", "view_inventory", "view_products", "view_suppliers", "view_customers", "view_finance"]},
|
||
# D17 Human-in-Loop：工艺工程师角色——可查看 + 提交老师傅反馈
|
||
{"code": "process_engineer", "name": "工艺工程师", "description": "工艺工程师，可查看 + 提交老师傅经验反馈", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_experience_feedback", "feedback_experience_hint"]},
|
||
]
|
||
|
||
|
||
async def init_permissions(session):
|
||
"""初始化权限（按 code 补登：已存在跳过，缺失新增）
|
||
|
||
设计要点（D17 修复）：
|
||
- 旧实现 `if existing_perms: return` 会让既存 DB 启动期漏掉新增权限码
|
||
- 改为按 code 比对：已存在的 permission 保留 id（避免 FK 引用失效），
|
||
缺失的新增；这样后续 DEFAULT_PERMISSIONS 追加的项也能在升级时落到既存 DB
|
||
"""
|
||
result = await session.execute(select(Permission))
|
||
existing_perms = {p.code: p for p in result.scalars().all()}
|
||
|
||
perm_map = {p.code: p.id for p in existing_perms.values()}
|
||
new_count = 0
|
||
for perm_data in DEFAULT_PERMISSIONS:
|
||
if perm_data["code"] in existing_perms:
|
||
continue
|
||
perm = Permission(**perm_data)
|
||
session.add(perm)
|
||
await session.flush()
|
||
perm_map[perm.code] = perm.id
|
||
new_count += 1
|
||
|
||
if not existing_perms:
|
||
logger.info(f"创建了 {len(DEFAULT_PERMISSIONS)} 个权限")
|
||
elif new_count:
|
||
logger.info(f"补登了 {new_count} 个权限（既有 DB 升级）")
|
||
else:
|
||
logger.info("权限已初始化（无新增）")
|
||
return perm_map
|
||
|
||
|
||
async def init_roles(session, perm_map):
|
||
"""初始化角色（按 code 补登：已存在跳过，缺失新建 + 完整绑定 permissions）
|
||
|
||
设计要点（D17 修复）：
|
||
- 旧实现 `if existing_roles: return` 会让既存 DB 启动期漏掉新角色
|
||
- 改为按 code 比对：已存在的角色不重置其 RolePermission 绑定
|
||
（避免重建关联破坏 user / role 关系），缺失的角色按 DEFAULT_ROLES 完整创建
|
||
"""
|
||
result = await session.execute(select(Role))
|
||
existing_roles = {r.code: r for r in result.scalars().all()}
|
||
|
||
new_count = 0
|
||
for role_data in DEFAULT_ROLES:
|
||
code = role_data["code"]
|
||
if code in existing_roles:
|
||
continue
|
||
perm_codes = role_data.pop("permissions")
|
||
role = Role(**role_data)
|
||
session.add(role)
|
||
await session.flush()
|
||
for perm_code in perm_codes:
|
||
perm_id = perm_map.get(perm_code)
|
||
if perm_id is None:
|
||
continue
|
||
rp = RolePermission(role_id=role.id, permission_id=perm_id)
|
||
session.add(rp)
|
||
new_count += 1
|
||
|
||
if not existing_roles:
|
||
logger.info(f"创建了 {len(DEFAULT_ROLES)} 个角色")
|
||
elif new_count:
|
||
logger.info(f"补登了 {new_count} 个角色（既有 DB 升级）")
|
||
else:
|
||
logger.info("角色已初始化（无新增）")
|
||
|
||
|
||
async def create_admin_user(session):
|
||
"""创建默认管理员"""
|
||
# compose 不再给 ADMIN_PASSWORD 弱默认（D14）：缺失时显式失败，
|
||
# 而不是静默创建空口令管理员
|
||
if not settings.ADMIN_PASSWORD:
|
||
raise RuntimeError(
|
||
"ADMIN_PASSWORD 未配置：请在 .env 中设置管理员初始密码后重启"
|
||
)
|
||
|
||
result = await session.execute(select(User).where(User.username == settings.ADMIN_USERNAME))
|
||
existing_admin = result.scalar_one_or_none()
|
||
|
||
if existing_admin:
|
||
logger.info("管理员账户已存在")
|
||
return
|
||
|
||
admin = User(
|
||
username=settings.ADMIN_USERNAME,
|
||
email=settings.ADMIN_EMAIL,
|
||
hashed_password=get_password_hash(settings.ADMIN_PASSWORD),
|
||
full_name=settings.ADMIN_FULL_NAME,
|
||
is_active=True
|
||
)
|
||
session.add(admin)
|
||
await session.flush()
|
||
|
||
result = await session.execute(select(Role).where(Role.code == "admin"))
|
||
admin_role = result.scalar_one_or_none()
|
||
|
||
if admin_role:
|
||
user_role = UserRole(user_id=admin.id, role_id=admin_role.id)
|
||
session.add(user_role)
|
||
|
||
await session.commit()
|
||
logger.info(f"创建了管理员账户: {settings.ADMIN_USERNAME}")
|
||
|
||
|
||
async def _verify_schema_coverage() -> None:
|
||
"""启动期 schema 校验：模型声明的表/列必须真实存在于 DB。
|
||
|
||
背景（2026-09-26 事故）：AUTO_MIGRATE 启动迁移曾静默失败，prod 库缺
|
||
stp_files.product_id 等列，上传全挂两天无启动日志线索。此校验保证
|
||
『迁移被跳过 / stamp 基线掩盖未应用增量』这类漂移在启动时即被点名。
|
||
|
||
只查缺、不查多：DB 里的遗留列/表（如 users.is_superuser）是历史产物，不管。
|
||
发现缺失只记 error 不抛——服务照常起，但 docker logs 必有醒目线索。
|
||
"""
|
||
import moldinsight.models # noqa: F401
|
||
import inventory.models # noqa: F401
|
||
from shared.models.base import Base
|
||
|
||
async with db_manager.engine.connect() as conn:
|
||
rows = (await conn.execute(text(
|
||
"SELECT table_name, column_name FROM information_schema.columns "
|
||
"WHERE table_schema='public'"
|
||
))).all()
|
||
db_cols: dict = {}
|
||
for tn, cn in rows:
|
||
db_cols.setdefault(tn, set()).add(cn)
|
||
|
||
missing = []
|
||
for tname, table in Base.metadata.tables.items():
|
||
present = db_cols.get(tname)
|
||
if present is None:
|
||
missing.append(f"整表缺失: {tname}")
|
||
continue
|
||
absent = {c.name for c in table.columns} - present
|
||
if absent:
|
||
missing.append(f"{tname} 缺列: {sorted(absent)}")
|
||
|
||
if missing:
|
||
logger.error(
|
||
"schema 校验失败：模型声明了 %d 处 DB 缺失（迁移链与实际 schema 不一致，"
|
||
"请人工执行 alembic upgrade head 或核对基线）：%s",
|
||
len(missing), "; ".join(missing),
|
||
)
|
||
else:
|
||
logger.info(
|
||
"schema 校验通过：模型 %d 张表的全部列均存在于 DB", len(Base.metadata.tables)
|
||
)
|
||
|
||
|
||
async def init_database(keep_connected: bool = True):
|
||
"""初始化数据库"""
|
||
try:
|
||
await db_manager.connect()
|
||
if settings.AUTO_MIGRATE:
|
||
await _run_alembic_migrations()
|
||
else:
|
||
logger.info(
|
||
"AUTO_MIGRATE=false：跳过启动期 alembic 迁移，"
|
||
"schema 由部署流程单点执行（alembic CLI 或 python -m shared.database.init_db）"
|
||
)
|
||
|
||
# 迁移后校验（只读、不抛）：迁移链与实际 schema 脱节时在启动日志直接点名
|
||
try:
|
||
await _verify_schema_coverage()
|
||
except Exception:
|
||
logger.exception("schema 校验自身异常（不影响启动流程）")
|
||
|
||
async with db_manager.session() as session:
|
||
perm_map = await init_permissions(session)
|
||
if perm_map is None:
|
||
# Permissions already existed, fetch them from database
|
||
result = await session.execute(select(Permission))
|
||
perms = result.scalars().all()
|
||
perm_map = {perm.code: perm.id for perm in perms}
|
||
await init_roles(session, perm_map)
|
||
await create_admin_user(session)
|
||
|
||
|
||
logger.info("数据库初始化完成")
|
||
print("=" * 60)
|
||
print("数据库初始化成功！")
|
||
print("=" * 60)
|
||
print(f"管理员用户名: {settings.ADMIN_USERNAME}")
|
||
print(f"管理员邮箱: {settings.ADMIN_EMAIL}")
|
||
print("=" * 60)
|
||
print("可以在 .env 文件中修改管理员配置:")
|
||
print(" ADMIN_USERNAME")
|
||
print(" ADMIN_EMAIL")
|
||
print(" ADMIN_FULL_NAME")
|
||
print("=" * 60)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
# logger.exception 而非 error(f"{e}")：吞掉 traceback 曾让启动迁移静默
|
||
# 失败两天无从排查（2026-09-26 事故）
|
||
logger.exception(f"数据库初始化失败: {e}")
|
||
print(f"数据库初始化失败: {e}")
|
||
return False
|
||
finally:
|
||
if not keep_connected:
|
||
await db_manager.disconnect()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(init_database(keep_connected=False))
|