155 lines
5.5 KiB
Python
155 lines
5.5 KiB
Python
"""给指定运营账号授予推介材料生成权限。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import text
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from app.core.contracts import RequestContext # noqa: E402
|
|
from app.infrastructure.db import SessionFactory, engine # noqa: E402
|
|
from app.service.identity_service import IdentityService # noqa: E402
|
|
|
|
PROMOTION_ROLE_CODE = "promotion_operator"
|
|
PROMOTION_ROLE_NAME = "推广材料运营"
|
|
DEFAULT_USER_ID = 9006
|
|
GRANTED_CODES = ("promotion:read", "promotion:write")
|
|
|
|
|
|
async def apply(user_id: int, *, dry_run: bool) -> int:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
assigned_at = now - timedelta(seconds=5)
|
|
async with SessionFactory() as session, session.begin():
|
|
user = (
|
|
await session.execute(
|
|
text(
|
|
"SELECT id, username, status FROM sys_user WHERE id=:user_id"
|
|
),
|
|
{"user_id": user_id},
|
|
)
|
|
).mappings().first()
|
|
if user is None:
|
|
raise SystemExit(f"未找到用户 id={user_id}")
|
|
if user["status"] != "正常":
|
|
raise SystemExit(f"用户 id={user_id} 未启用，拒绝授权")
|
|
|
|
role_id = await session.scalar(
|
|
text("SELECT id FROM sys_role WHERE role_code=:code"),
|
|
{"code": PROMOTION_ROLE_CODE},
|
|
)
|
|
permission_rows = (
|
|
await session.execute(
|
|
text(
|
|
"SELECT permission_code, id FROM sys_permission "
|
|
"WHERE permission_code IN ('promotion:read', 'promotion:write')"
|
|
)
|
|
)
|
|
).all()
|
|
permission_ids = {str(code): int(permission_id) for code, permission_id in permission_rows}
|
|
missing = [code for code in GRANTED_CODES if code not in permission_ids]
|
|
if missing:
|
|
raise SystemExit(f"数据库缺少权限码，请先运行权限种子：{missing}")
|
|
|
|
print(f"用户：id={user_id} username={user['username']}")
|
|
print(
|
|
f"角色：{PROMOTION_ROLE_CODE} "
|
|
f"{'已存在 id=' + str(role_id) if role_id else '将新建'}"
|
|
)
|
|
print(f"授权范围：{', '.join(GRANTED_CODES)}")
|
|
if dry_run:
|
|
print("[dry-run] 未写入任何数据")
|
|
return 0
|
|
|
|
if role_id is None:
|
|
await session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO sys_role
|
|
(role_code, role_name, status, created_at, updated_at)
|
|
VALUES (:code, :name, 'active', :now, :now)
|
|
"""
|
|
),
|
|
{"code": PROMOTION_ROLE_CODE, "name": PROMOTION_ROLE_NAME, "now": now},
|
|
)
|
|
role_id = await session.scalar(
|
|
text("SELECT id FROM sys_role WHERE role_code=:code"),
|
|
{"code": PROMOTION_ROLE_CODE},
|
|
)
|
|
role_id = int(role_id)
|
|
|
|
for code in GRANTED_CODES:
|
|
exists = await session.scalar(
|
|
text(
|
|
"""
|
|
SELECT COUNT(*) FROM sys_role_permission
|
|
WHERE role_id=:role_id AND permission_id=:permission_id
|
|
"""
|
|
),
|
|
{"role_id": role_id, "permission_id": permission_ids[code]},
|
|
)
|
|
if not exists:
|
|
await session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO sys_role_permission (role_id, permission_id, created_at)
|
|
VALUES (:role_id, :permission_id, :now)
|
|
"""
|
|
),
|
|
{
|
|
"role_id": role_id,
|
|
"permission_id": permission_ids[code],
|
|
"now": now,
|
|
},
|
|
)
|
|
|
|
await session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO sys_user_role (user_id, role_id, assigned_at)
|
|
VALUES (:user_id, :role_id, :assigned_at)
|
|
ON DUPLICATE KEY UPDATE assigned_at=:assigned_at
|
|
"""
|
|
),
|
|
{"user_id": user_id, "role_id": role_id, "assigned_at": assigned_at},
|
|
)
|
|
|
|
return await verify(user_id)
|
|
|
|
|
|
async def verify(user_id: int) -> int:
|
|
context = await IdentityService().resolve(
|
|
RequestContext(user_id=str(user_id), trace_id="grant-promotion-operator")
|
|
)
|
|
expected = set(GRANTED_CODES)
|
|
actual = set(context.permissions)
|
|
print(f"真实解析角色：{sorted(context.roles)}")
|
|
print(f"真实解析权限：{sorted(actual)}")
|
|
missing = expected - actual
|
|
if missing:
|
|
raise SystemExit(f"授权验证失败，缺少权限：{sorted(missing)}")
|
|
print("推介材料生成权限已生效：promotion:read、promotion:write")
|
|
return 0
|
|
|
|
|
|
async def main() -> int:
|
|
parser = argparse.ArgumentParser(description="授予运营账号推介材料生成权限")
|
|
parser.add_argument("--user-id", type=int, default=DEFAULT_USER_ID)
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
return await apply(args.user_id, dry_run=args.dry_run)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main()))
|