tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: permission

Browse Source
This commit is contained in:
tao.chen
2026-09-02 10:10:41 +08:00
committed by tao.chen
parent 1c220490cf
commit dea6b4cfcf
11 changed files with 171 additions and 1037 deletions
Download Patch File Download Diff File Expand all files Collapse all files
migrations/data/README.md
-70
View File
@@ -1,70 +0,0 @@
# Data Migrations
该目录保存从旧版 JSON 状态文件迁移到 MySQL 的一次性工具。
数据迁移工具必须满足：
- 默认只预检，必须显式传入 `--apply` 才能写数据库；
- 单个事务提交，失败时完整回滚；
- 可以安全重复执行，不产生重复数据；
- 输出源文件摘要、源数据数量和实际变更数量；
- 不修改或删除旧版 JSON 源文件。
## system.json
```powershell
$env:DATABASE_URL = "mysql+asyncmy://<user>:<password>@<host>:3306/<database>?charset=utf8mb4"
python -m migrations.data.migrate_system_json --source "<path>/system.json"
python -m migrations.data.migrate_system_json --source "<path>/system.json" --apply
```
新迁移用户使用不可登录的占位密码哈希。后续接入认证时，必须通过密码初始化、
管理员重置或外部身份认证启用登录，不能把旧版无密码账号视为已有凭据。
## WORKSPACE_DEFINITIONS
旧版 Workspace 定义位于 `文件1/server.py` 的
`WORKSPACE_DEFINITIONS` 常量中。迁移工具通过 Python AST 仅读取这一静态常量，
不执行旧服务代码：
```powershell
python -m migrations.data.migrate_legacy_workspaces --source "<path>/server.py"
python -m migrations.data.migrate_legacy_workspaces --source "<path>/server.py" --apply
```
- 创建者取每个 Workspace 成员中的第一个管理员；
- 成员角色沿用第 9 步迁入的平台注册角色；
- 活跃目录统一为 `file:///workspace/workspaces/{workspace_code}`；
- 制品前缀统一为 `workspaces/{workspace_id}`；
- 旧审计记录仅在用户唯一属于一个 Workspace 时补齐归属。
## 第 9 小步执行记录
- 执行日期：2026-07-24
- 源文件：`文件1/platform_data/system.json`
- SHA-256：`ef4ee0e92f4a3679c17f23aff6066688fa9231db4477208e36fcbd343cd446a7`
- 迁移结果：2 个角色、13 个权限、21 条角色权限、4 个用户、26 条审计记录
- 角色权限：`admin=13`，`developer=8`
- 中文字段：UTF-8 校验通过，问号乱码记录为 0
- 幂等验证：第二次执行插入和更新均为 0，跳过已有审计记录 26 条
- 旧版源文件：未修改、未删除
## 第 10 小步执行记录
- 执行日期：2026-07-24
- 源文件及常量：`文件1/server.py` / `WORKSPACE_DEFINITIONS`
- SHA-256：`d7ec05f1ab7b2cdeb78f6293de40fbe23fa74b6a7fe4ccb03a9cc64e4cc1f464`
- 迁移结果：2 个 Workspace、4 条成员关系
- 成员分布：`model-dev` 2 人，`risk-validation` 2 人
- 审计归属：补齐 26 条，其中 `model-dev` 24 条、`risk-validation` 2 条
- 中文字段：UTF-8 校验通过，异常记录为 0
- 幂等验证：第二次执行及镜像内 dry-run 的变更数均为 0
- 镜像验证：仅挂载旧版 `server.py` 时，迁移工具可独立读取并完成校验
- 旧版源文件：未修改、未删除
## 第 11、12 小步数据处理决定
根据实施确认，第 11、12 小步不迁移旧版资源、脚本或稳定版本数据。新实现直接
使用 MySQL、Workspace 文件目录和对象存储（s3 模式连 S3-兼容服务，local 模式
走 `/data/storage` 共享卷），从空的 `data_resources`、`scripts` 及
`versions` 表开始运行。功能验收产生的临时对象和数据库记录均已清理。
migrations/data/migrate_legacy_workspaces.py
-365
View File
@@ -1,365 +0,0 @@
from __future__ import annotations
import argparse
import ast
import asyncio
import hashlib
import json
import os
from pathlib import Path
from typing import Any
from urllib.parse import quote
from common.db import create_database_engine, create_session_factory
from common.db.models import (
Roles,
Users,
WorkspaceMembers,
Workspaces,
)
from pydantic import BaseModel, ConfigDict, TypeAdapter
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from migrations.data.migrate_system_json import (
deterministic_legacy_ulid,
set_changed,
)
class LegacyWorkspace(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str
name: str
description: str | None = None
userIds: list[str]
portOffset: int
WORKSPACE_LIST = TypeAdapter(list[LegacyWorkspace])
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 extract_workspace_definitions(
source_path: Path,
) -> tuple[list[LegacyWorkspace], str]:
raw = source_path.read_bytes()
module = ast.parse(
raw.decode("utf-8-sig"),
filename=str(source_path),
)
definition: Any | None = None
for statement in module.body:
if not isinstance(statement, ast.Assign):
continue
if any(
isinstance(target, ast.Name)
and target.id == "WORKSPACE_DEFINITIONS"
for target in statement.targets
):
definition = ast.literal_eval(statement.value)
break
if definition is None:
raise ValueError("WORKSPACE_DEFINITIONS was not found")
workspaces = WORKSPACE_LIST.validate_python(definition)
require_unique([item.id for item in workspaces], "workspace ids")
for workspace in workspaces:
require_unique(
workspace.userIds,
f"members of workspace {workspace.id}",
)
if not workspace.userIds:
raise ValueError(
f"workspace {workspace.id!r} has no members"
)
return workspaces, hashlib.sha256(raw).hexdigest()
def new_stats() -> dict[str, int]:
return {
"workspaces_inserted": 0,
"workspaces_updated": 0,
"workspace_members_inserted": 0,
"workspace_members_updated": 0,
}
async def load_users_and_roles(
session: AsyncSession,
) -> tuple[dict[str, Users], dict[str, str]]:
users = {
item.username: item
for item in (
await session.scalars(select(Users).order_by(Users.username))
).all()
}
roles_by_id = {
item.role_id: item.role_code
for item in (
await session.scalars(select(Roles).order_by(Roles.role_code))
).all()
}
role_codes_by_username = {
username: roles_by_id[user.platform_role_id]
for username, user in users.items()
if user.platform_role_id in roles_by_id
}
return users, role_codes_by_username
def validate_members(
workspaces: list[LegacyWorkspace],
users: dict[str, Users],
role_codes_by_username: dict[str, str],
) -> None:
source_members = {
username
for workspace in workspaces
for username in workspace.userIds
}
missing_users = sorted(source_members - set(users))
if missing_users:
raise ValueError(
f"workspace members were not migrated as users: {missing_users}"
)
users_without_roles = sorted(
source_members - set(role_codes_by_username)
)
if users_without_roles:
raise ValueError(
"workspace members have no platform role: "
f"{users_without_roles}"
)
def workspace_creator(
workspace: LegacyWorkspace,
users: dict[str, Users],
role_codes_by_username: dict[str, str],
) -> Users:
admin_username = next(
(
username
for username in workspace.userIds
if role_codes_by_username[username] == "admin"
),
workspace.userIds[0],
)
return users[admin_username]
async def migrate_workspaces(
session: AsyncSession,
source: list[LegacyWorkspace],
users: dict[str, Users],
role_codes_by_username: dict[str, str],
stats: dict[str, int],
) -> dict[str, str]:
existing = {
item.workspace_code: item
for item in (
await session.scalars(
select(Workspaces).order_by(Workspaces.workspace_code)
)
).all()
}
workspace_ids: dict[str, str] = {}
for legacy in source:
workspace = existing.get(legacy.id)
if workspace is None:
workspace_id = deterministic_legacy_ulid(
"workspace", legacy.id
)
creator = workspace_creator(
legacy,
users,
role_codes_by_username,
)
workspace = Workspaces(
workspace_id=workspace_id,
workspace_code=legacy.id,
workspace_name=legacy.name,
description=legacy.description,
active_root_uri=(
"file:///workspace/workspaces/"
f"{quote(legacy.id, safe='')}"
),
quota_bytes=0,
used_bytes=0,
status="active",
created_by=creator.user_id,
artifact_bucket="model-platform",
artifact_prefix=f"workspaces/{workspace_id}",
)
session.add(workspace)
stats["workspaces_inserted"] += 1
else:
values = {
"workspace_name": legacy.name,
"description": legacy.description,
"active_root_uri": (
"file:///workspace/workspaces/"
f"{quote(legacy.id, safe='')}"
),
"artifact_bucket": "model-platform",
"artifact_prefix": (
f"workspaces/{workspace.workspace_id}"
),
}
if set_changed(workspace, values):
stats["workspaces_updated"] += 1
workspace_ids[legacy.id] = workspace.workspace_id
return workspace_ids
async def migrate_members(
session: AsyncSession,
source: list[LegacyWorkspace],
workspace_ids: dict[str, str],
users: dict[str, Users],
stats: dict[str, int],
) -> None:
existing = {
(item.workspace_id, item.user_id): item
for item in (
await session.scalars(select(WorkspaceMembers))
).all()
}
for legacy in source:
workspace_id = workspace_ids[legacy.id]
for username in legacy.userIds:
user = users[username]
pair = (workspace_id, user.user_id)
member = existing.get(pair)
values = {
"role_id": user.platform_role_id,
"member_status": "active",
}
if member is None:
member = WorkspaceMembers(
workspace_id=workspace_id,
user_id=user.user_id,
**values,
)
session.add(member)
existing[pair] = member
stats["workspace_members_inserted"] += 1
elif set_changed(member, values):
stats["workspace_members_updated"] += 1
async def run_migration(
database_url: str,
source: list[LegacyWorkspace],
*,
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:
users, role_codes_by_username = (
await load_users_and_roles(session)
)
validate_members(
source,
users,
role_codes_by_username,
)
workspace_ids = await migrate_workspaces(
session,
source,
users,
role_codes_by_username,
stats,
)
await migrate_members(
session,
source,
workspace_ids,
users,
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 WORKSPACE_DEFINITIONS into MySQL."
)
)
parser.add_argument(
"--source",
required=True,
type=Path,
help="Path to the legacy server.py 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)
workspaces, source_sha256 = extract_workspace_definitions(source_path)
stats = await run_migration(
os.environ["DATABASE_URL"],
workspaces,
apply_changes=args.apply,
)
result = {
"mode": "apply" if args.apply else "dry-run",
"source": str(source_path),
"source_symbol": "WORKSPACE_DEFINITIONS",
"source_sha256": source_sha256,
"source_counts": {
"workspaces": len(workspaces),
"workspace_members": sum(
len(workspace.userIds) for workspace in workspaces
),
},
"changes": stats,
}
print(json.dumps(result, ensure_ascii=False, indent=2))
def main() -> None:
asyncio.run(async_main())
if __name__ == "__main__":
main()
migrations/data/migrate_system_json.py
-406
View File
@@ -1,406 +0,0 @@
from __future__ import annotations
import argparse
import asyncio
import hashlib
import json
import os
from pathlib import Path
from typing import Any
from urllib.parse import quote
from common.db import create_database_engine, create_session_factory
from common.db.models import (
Permissions,
RolePermissions,
Roles,
Users,
)
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
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": "管理系统配置",
}
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 LegacySystem(BaseModel):
"""Legacy platform_data/system.json shape.
``extra='ignore'`` so legacy files that include other top-level
keys (e.g. historical audit-log payloads) can still be parsed —
only the fields this script actually consumes are listed below.
"""
model_config = ConfigDict(extra="ignore")
users: list[LegacyUser]
roles: list[LegacyRole]
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()
).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]
require_unique(role_codes, "role keys")
require_unique(user_codes, "user 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}")
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,
}
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
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,
)
await migrate_users(
session, source, role_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),
},
"changes": stats,
}
print(json.dumps(result, ensure_ascii=False, indent=2))
def main() -> None:
asyncio.run(async_main())
if __name__ == "__main__":
main()