505 lines
15 KiB
Python
505 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.db import create_database_engine, create_session_factory
|
|
from common.db.models import (
|
|
AuditLogs,
|
|
Permissions,
|
|
RolePermissions,
|
|
Roles,
|
|
Users,
|
|
)
|
|
|
|
CROCKFORD_BASE32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
|
|
LEGACY_PASSWORD_HASH = "!legacy-account-without-password!"
|
|
|
|
PERMISSION_NAMES = {
|
|
"dashboard.view": "查看工作台",
|
|
"script.build": "构建脚本",
|
|
"script.public.manage": "管理公共脚本",
|
|
"schedule.own": "管理本人调度",
|
|
"schedule.all": "管理全部调度",
|
|
"experiment.own": "管理本人实验",
|
|
"experiment.all": "管理全部实验",
|
|
"resource.personal": "管理个人资源",
|
|
"resource.public.upload": "上传公共资源",
|
|
"resource.public.manage": "管理公共资源",
|
|
"system.view": "查看系统管理",
|
|
"system.manage": "管理系统配置",
|
|
"audit.view": "查看审计日志",
|
|
}
|
|
|
|
ACTION_CATALOG = {
|
|
"保存调度配置": ("schedule.save", "schedule"),
|
|
"删除脚本对象": ("script.delete", "script"),
|
|
"删除实验记录": ("experiment.delete", "experiment"),
|
|
"删除数据资源": ("data_resource.delete", "data_resource"),
|
|
"上传数据资源": ("data_resource.upload", "data_resource"),
|
|
"新建脚本对象": ("script.create", "script"),
|
|
"修改用户角色": ("user.role.update", "user"),
|
|
"运行 Python": ("script.run_python", "script"),
|
|
"运行调度": ("schedule.run", "schedule"),
|
|
}
|
|
|
|
|
|
class LegacyUser(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
id: str
|
|
name: str
|
|
role: str
|
|
roleKey: str
|
|
avatar: str | None = None
|
|
|
|
|
|
class LegacyRole(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
key: str
|
|
name: str
|
|
description: str | None = None
|
|
permissions: list[str]
|
|
|
|
|
|
class LegacyAuditLog(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
id: str
|
|
actorId: str
|
|
actorName: str
|
|
role: str
|
|
action: str
|
|
target: str
|
|
detail: str
|
|
status: str
|
|
createdAt: str
|
|
|
|
|
|
class LegacySystem(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
users: list[LegacyUser]
|
|
roles: list[LegacyRole]
|
|
audit_logs: list[LegacyAuditLog] = Field(alias="auditLogs")
|
|
|
|
|
|
def deterministic_legacy_ulid(entity_type: str, legacy_key: str) -> str:
|
|
"""Create a stable ULID-compatible ID with an epoch timestamp prefix."""
|
|
digest = hashlib.sha256(
|
|
f"model-platform-v1:{entity_type}:{legacy_key}".encode("utf-8")
|
|
).digest()
|
|
value = int.from_bytes(b"\x00" * 6 + digest[:10], byteorder="big")
|
|
encoded = ["0"] * 26
|
|
for index in range(25, -1, -1):
|
|
encoded[index] = CROCKFORD_BASE32[value & 31]
|
|
value >>= 5
|
|
return "".join(encoded)
|
|
|
|
|
|
def require_unique(values: list[str], label: str) -> None:
|
|
duplicates = sorted(
|
|
value for value in set(values) if values.count(value) > 1
|
|
)
|
|
if duplicates:
|
|
raise ValueError(f"duplicate {label}: {duplicates}")
|
|
|
|
|
|
def validate_source(source: LegacySystem) -> None:
|
|
role_codes = [role.key for role in source.roles]
|
|
user_codes = [user.id for user in source.users]
|
|
audit_ids = [item.id for item in source.audit_logs]
|
|
require_unique(role_codes, "role keys")
|
|
require_unique(user_codes, "user ids")
|
|
require_unique(audit_ids, "audit ids")
|
|
|
|
role_code_set = set(role_codes)
|
|
unknown_roles = sorted(
|
|
user.roleKey
|
|
for user in source.users
|
|
if user.roleKey not in role_code_set
|
|
)
|
|
if unknown_roles:
|
|
raise ValueError(f"users reference unknown roles: {unknown_roles}")
|
|
|
|
user_code_set = set(user_codes)
|
|
unknown_actors = sorted(
|
|
item.actorId
|
|
for item in source.audit_logs
|
|
if item.actorId not in user_code_set
|
|
)
|
|
if unknown_actors:
|
|
raise ValueError(f"audit logs reference unknown users: {unknown_actors}")
|
|
|
|
for role in source.roles:
|
|
require_unique(role.permissions, f"permissions of role {role.key}")
|
|
for permission_code in role.permissions:
|
|
if "." not in permission_code:
|
|
raise ValueError(
|
|
f"invalid permission code {permission_code!r}"
|
|
)
|
|
|
|
|
|
def load_source(path: Path) -> tuple[LegacySystem, str]:
|
|
raw = path.read_bytes()
|
|
source = LegacySystem.model_validate_json(raw)
|
|
validate_source(source)
|
|
return source, hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def set_changed(instance: Any, values: dict[str, Any]) -> bool:
|
|
changed = False
|
|
for attribute, value in values.items():
|
|
if getattr(instance, attribute) != value:
|
|
setattr(instance, attribute, value)
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def new_stats() -> dict[str, int]:
|
|
return {
|
|
"roles_inserted": 0,
|
|
"roles_updated": 0,
|
|
"permissions_inserted": 0,
|
|
"permissions_updated": 0,
|
|
"role_permissions_inserted": 0,
|
|
"users_inserted": 0,
|
|
"users_updated": 0,
|
|
"audit_logs_inserted": 0,
|
|
"audit_logs_skipped": 0,
|
|
}
|
|
|
|
|
|
async def migrate_roles(
|
|
session: AsyncSession,
|
|
source: LegacySystem,
|
|
stats: dict[str, int],
|
|
) -> dict[str, str]:
|
|
existing = {
|
|
item.role_code: item
|
|
for item in (
|
|
await session.scalars(select(Roles).order_by(Roles.role_code))
|
|
).all()
|
|
}
|
|
role_ids: dict[str, str] = {}
|
|
|
|
for legacy in source.roles:
|
|
values = {
|
|
"role_name": legacy.name,
|
|
"role_scope": "platform",
|
|
"is_builtin": 1,
|
|
"description": legacy.description,
|
|
}
|
|
role = existing.get(legacy.key)
|
|
if role is None:
|
|
role = Roles(
|
|
role_id=deterministic_legacy_ulid("role", legacy.key),
|
|
role_code=legacy.key,
|
|
**values,
|
|
)
|
|
session.add(role)
|
|
stats["roles_inserted"] += 1
|
|
elif set_changed(role, values):
|
|
stats["roles_updated"] += 1
|
|
role_ids[legacy.key] = role.role_id
|
|
|
|
return role_ids
|
|
|
|
|
|
async def migrate_permissions(
|
|
session: AsyncSession,
|
|
source: LegacySystem,
|
|
stats: dict[str, int],
|
|
) -> dict[str, str]:
|
|
permission_codes = sorted(
|
|
{
|
|
permission_code
|
|
for role in source.roles
|
|
for permission_code in role.permissions
|
|
}
|
|
)
|
|
existing = {
|
|
item.permission_code: item
|
|
for item in (
|
|
await session.scalars(
|
|
select(Permissions).order_by(Permissions.permission_code)
|
|
)
|
|
).all()
|
|
}
|
|
permission_ids: dict[str, str] = {}
|
|
|
|
for permission_code in permission_codes:
|
|
values = {
|
|
"permission_name": PERMISSION_NAMES.get(
|
|
permission_code, permission_code
|
|
),
|
|
"module_code": permission_code.split(".", 1)[0],
|
|
"description": f"由旧版 system.json 迁移：{permission_code}",
|
|
}
|
|
permission = existing.get(permission_code)
|
|
if permission is None:
|
|
permission = Permissions(
|
|
permission_id=deterministic_legacy_ulid(
|
|
"permission", permission_code
|
|
),
|
|
permission_code=permission_code,
|
|
**values,
|
|
)
|
|
session.add(permission)
|
|
stats["permissions_inserted"] += 1
|
|
elif set_changed(permission, values):
|
|
stats["permissions_updated"] += 1
|
|
permission_ids[permission_code] = permission.permission_id
|
|
|
|
return permission_ids
|
|
|
|
|
|
async def migrate_role_permissions(
|
|
session: AsyncSession,
|
|
source: LegacySystem,
|
|
role_ids: dict[str, str],
|
|
permission_ids: dict[str, str],
|
|
stats: dict[str, int],
|
|
) -> None:
|
|
existing = set(
|
|
(
|
|
await session.execute(
|
|
select(
|
|
RolePermissions.role_id,
|
|
RolePermissions.permission_id,
|
|
)
|
|
)
|
|
).all()
|
|
)
|
|
|
|
for role in source.roles:
|
|
for permission_code in role.permissions:
|
|
pair = (
|
|
role_ids[role.key],
|
|
permission_ids[permission_code],
|
|
)
|
|
if pair not in existing:
|
|
session.add(
|
|
RolePermissions(
|
|
role_id=pair[0],
|
|
permission_id=pair[1],
|
|
)
|
|
)
|
|
existing.add(pair)
|
|
stats["role_permissions_inserted"] += 1
|
|
|
|
|
|
async def migrate_users(
|
|
session: AsyncSession,
|
|
source: LegacySystem,
|
|
role_ids: dict[str, str],
|
|
stats: dict[str, int],
|
|
) -> dict[str, str]:
|
|
existing = {
|
|
item.username: item
|
|
for item in (
|
|
await session.scalars(select(Users).order_by(Users.username))
|
|
).all()
|
|
}
|
|
user_ids: dict[str, str] = {}
|
|
|
|
for legacy in source.users:
|
|
values = {
|
|
"display_name": legacy.name,
|
|
"platform_role_id": role_ids[legacy.roleKey],
|
|
"avatar_uri": (
|
|
f"initial://{quote(legacy.avatar)}"
|
|
if legacy.avatar
|
|
else None
|
|
),
|
|
}
|
|
user = existing.get(legacy.id)
|
|
if user is None:
|
|
user = Users(
|
|
user_id=deterministic_legacy_ulid("user", legacy.id),
|
|
username=legacy.id,
|
|
password_hash=LEGACY_PASSWORD_HASH,
|
|
status="active",
|
|
**values,
|
|
)
|
|
session.add(user)
|
|
stats["users_inserted"] += 1
|
|
elif set_changed(user, values):
|
|
stats["users_updated"] += 1
|
|
user_ids[legacy.id] = user.user_id
|
|
|
|
return user_ids
|
|
|
|
|
|
def audit_catalog(action: str) -> tuple[str, str]:
|
|
known = ACTION_CATALOG.get(action)
|
|
if known is not None:
|
|
return known
|
|
digest = hashlib.sha256(action.encode("utf-8")).hexdigest()[:16]
|
|
return (f"legacy.action.{digest}", "legacy")
|
|
|
|
|
|
async def migrate_audit_logs(
|
|
session: AsyncSession,
|
|
source: LegacySystem,
|
|
user_ids: dict[str, str],
|
|
stats: dict[str, int],
|
|
) -> None:
|
|
existing_payloads = (
|
|
await session.scalars(select(AuditLogs.detail_json))
|
|
).all()
|
|
existing_legacy_ids = {
|
|
payload.get("legacy_id")
|
|
for payload in existing_payloads
|
|
if isinstance(payload, dict) and payload.get("legacy_id")
|
|
}
|
|
|
|
for legacy in sorted(source.audit_logs, key=lambda item: item.createdAt):
|
|
if legacy.id in existing_legacy_ids:
|
|
stats["audit_logs_skipped"] += 1
|
|
continue
|
|
|
|
action_code, target_type = audit_catalog(legacy.action)
|
|
session.add(
|
|
AuditLogs(
|
|
actor_user_id=user_ids[legacy.actorId],
|
|
action_code=action_code,
|
|
target_type=target_type,
|
|
target_id=legacy.target[:128] or None,
|
|
operation_status=(
|
|
"success" if legacy.status == "成功" else "failed"
|
|
),
|
|
created_at=datetime.strptime(
|
|
legacy.createdAt, "%Y-%m-%d %H:%M:%S"
|
|
),
|
|
detail_json={
|
|
"legacy_id": legacy.id,
|
|
"legacy_actor_name": legacy.actorName,
|
|
"legacy_role": legacy.role,
|
|
"legacy_action": legacy.action,
|
|
"legacy_target": legacy.target,
|
|
"legacy_detail": legacy.detail,
|
|
"legacy_status": legacy.status,
|
|
"migration_source": "platform_data/system.json",
|
|
},
|
|
)
|
|
)
|
|
existing_legacy_ids.add(legacy.id)
|
|
stats["audit_logs_inserted"] += 1
|
|
|
|
|
|
async def run_migration(
|
|
database_url: str,
|
|
source: LegacySystem,
|
|
*,
|
|
apply_changes: bool,
|
|
) -> dict[str, int]:
|
|
engine = create_database_engine(database_url)
|
|
factory = create_session_factory(engine)
|
|
stats = new_stats()
|
|
|
|
try:
|
|
async with factory() as session:
|
|
try:
|
|
role_ids = await migrate_roles(session, source, stats)
|
|
permission_ids = await migrate_permissions(
|
|
session, source, stats
|
|
)
|
|
await migrate_role_permissions(
|
|
session,
|
|
source,
|
|
role_ids,
|
|
permission_ids,
|
|
stats,
|
|
)
|
|
user_ids = await migrate_users(
|
|
session, source, role_ids, stats
|
|
)
|
|
await migrate_audit_logs(
|
|
session, source, user_ids, stats
|
|
)
|
|
|
|
if apply_changes:
|
|
await session.commit()
|
|
else:
|
|
await session.rollback()
|
|
except Exception:
|
|
await session.rollback()
|
|
raise
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
return stats
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Migrate legacy system.json records into MySQL."
|
|
)
|
|
parser.add_argument(
|
|
"--source",
|
|
required=True,
|
|
type=Path,
|
|
help="Path to the legacy platform_data/system.json file.",
|
|
)
|
|
parser.add_argument(
|
|
"--apply",
|
|
action="store_true",
|
|
help="Commit changes. Without this flag the transaction is rolled back.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
async def async_main() -> None:
|
|
args = parse_args()
|
|
source_path = args.source.resolve(strict=True)
|
|
source, source_sha256 = load_source(source_path)
|
|
database_url = os.environ["DATABASE_URL"]
|
|
stats = await run_migration(
|
|
database_url,
|
|
source,
|
|
apply_changes=args.apply,
|
|
)
|
|
result = {
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"source": str(source_path),
|
|
"source_sha256": source_sha256,
|
|
"source_counts": {
|
|
"roles": len(source.roles),
|
|
"permissions": len(
|
|
{
|
|
permission
|
|
for role in source.roles
|
|
for permission in role.permissions
|
|
}
|
|
),
|
|
"role_permissions": sum(
|
|
len(role.permissions) for role in source.roles
|
|
),
|
|
"users": len(source.users),
|
|
"audit_logs": len(source.audit_logs),
|
|
},
|
|
"changes": stats,
|
|
}
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def main() -> None:
|
|
asyncio.run(async_main())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|