hyapp-server/scripts/ops/refresh_robot_appearance.py
2026-06-24 18:56:14 +08:00

862 lines
34 KiB
Python
Executable File

#!/usr/bin/env python3
"""Refresh robot avatar frames and VIP badges.
The script is intentionally dry-run by default. With --yes it:
1. Builds the robot user set from users.source=game_robot plus robot-service pools.
2. Picks avatar frames priced at or below 2,000,000 COIN, excluding known reserved frames.
3. Picks one VIP strip badge and one VIP tile badge.
4. Ensures wallet entitlements exist, replaces equipped avatar/badge rows, and rebuilds activity badge display slots.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import shlex
import subprocess
import time
import uuid
from pathlib import Path
from typing import Any
import pymysql
import yaml
APP_CODE_DEFAULT = "lalu"
ROBOT_SOURCE_DEFAULT = "game_robot"
GRANT_SOURCE = "game_robot_init"
RESOURCE_OUTBOX_ASSET = "RESOURCE"
OUTBOX_STATUS_PENDING = "pending"
ENTITLEMENT_STATUS_ACTIVE = "active"
GRANT_STATUS_DONE = "succeeded"
RESULT_ENTITLEMENT = "entitlement"
ROBOT_RESOURCE_DURATION_DAYS = 9999
ROBOT_RESOURCE_DURATION_MS = ROBOT_RESOURCE_DURATION_DAYS * 24 * 60 * 60 * 1000
ROBOT_AVATAR_FRAME_MAX_COIN_PRICE = 2_000_000
EXCLUDED_AVATAR_FRAME_IDS = {165, 175}
BADGE_DISPLAY_SLOTS = ("profile_strip", "profile_tile", "honor_wall")
def normalize_app_code(value: str) -> str:
return (value or "").strip().lower() or APP_CODE_DEFAULT
def stable_hash(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def prefixed_id(prefix: str, value: str) -> str:
return prefix + stable_hash(value)
def current_ms() -> int:
return int(time.time() * 1000)
def read_config_text(location: str) -> str:
location = location.strip()
if looks_like_ssh_location(location):
host, remote_path = location.split(":", 1)
return subprocess.check_output(
["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=5", host, "cat " + shlex.quote(remote_path)],
text=True,
)
return Path(location).expanduser().read_text(encoding="utf-8")
def looks_like_ssh_location(location: str) -> bool:
if location.startswith("/") or location.startswith("./") or location.startswith("../") or "://" in location:
return False
head, sep, tail = location.partition(":")
return bool(sep and tail and ("@" in head or "." in head or head.replace("-", "").isalnum()))
def load_yaml(location: str) -> dict[str, Any]:
payload = yaml.safe_load(read_config_text(location)) or {}
if not isinstance(payload, dict):
raise RuntimeError(f"config must be a YAML object: {location}")
return payload
def parse_go_mysql_dsn(dsn: str) -> dict[str, Any]:
prefix, rest = dsn.split("@tcp(", 1)
user, password = prefix.split(":", 1) if ":" in prefix else (prefix, "")
addr, rest = rest.split(")/", 1)
host, port = addr.rsplit(":", 1) if ":" in addr else (addr, "3306")
database = rest.split("?", 1)[0]
return {
"host": host,
"port": int(port),
"user": user,
"password": password,
"database": database,
"charset": "utf8mb4",
"autocommit": False,
"connect_timeout": 8,
"read_timeout": 120,
"write_timeout": 120,
"cursorclass": pymysql.cursors.DictCursor,
}
def connect_from_config(path: str, dsn_key: str = "mysql_dsn"):
cfg = load_yaml(path)
dsn = str(cfg.get(dsn_key) or "").strip()
if not dsn:
raise RuntimeError(f"{dsn_key} is required in {path}")
return pymysql.connect(**parse_go_mysql_dsn(dsn))
def table_exists(conn, table_name: str) -> bool:
with conn.cursor() as cursor:
cursor.execute(
"SELECT COUNT(*) AS count FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
(table_name,),
)
return int(cursor.fetchone()["count"]) > 0
def fetch_user_source_robots(conn, app_code: str, source: str) -> dict[int, dict[str, Any]]:
with conn.cursor() as cursor:
cursor.execute(
"""
SELECT user_id, status, COALESCE(source, '') AS source
FROM users
WHERE app_code = %s AND source = %s
ORDER BY user_id ASC
""",
(app_code, source),
)
return {
int(row["user_id"]): {"user_status": row["status"], "sources": {f"users.source:{row['source']}"}}
for row in cursor.fetchall()
}
def fetch_robot_pool_users(conn, app_code: str) -> dict[int, set[str]]:
result: dict[int, set[str]] = {}
for table, label_column in (("robot_game_robots", "game_id"), ("robot_room_robots", "room_scene")):
if not table_exists(conn, table):
continue
with conn.cursor() as cursor:
cursor.execute(
f"""
SELECT user_id, status, {label_column} AS label
FROM {table}
WHERE app_code = %s
ORDER BY user_id ASC
""",
(app_code,),
)
for row in cursor.fetchall():
user_id = int(row["user_id"])
result.setdefault(user_id, set()).add(f"{table}:{row['label']}:{row['status']}")
return result
def merge_robot_sources(
user_robots: dict[int, dict[str, Any]],
pool_robots: dict[int, set[str]],
) -> dict[int, dict[str, Any]]:
merged = dict(user_robots)
for user_id, sources in pool_robots.items():
item = merged.setdefault(user_id, {"user_status": "", "sources": set()})
item["sources"].update(sources)
return merged
def filter_existing_users(conn, app_code: str, robot_map: dict[int, dict[str, Any]]) -> dict[int, dict[str, Any]]:
user_ids = sorted(robot_map)
if not user_ids:
return {}
existing: dict[int, str] = {}
for chunk in chunks(user_ids, 500):
placeholders = ",".join(["%s"] * len(chunk))
with conn.cursor() as cursor:
cursor.execute(
f"SELECT user_id, status FROM users WHERE app_code = %s AND user_id IN ({placeholders})",
[app_code, *chunk],
)
for row in cursor.fetchall():
existing[int(row["user_id"])] = str(row["status"])
filtered: dict[int, dict[str, Any]] = {}
for user_id in user_ids:
if user_id not in existing:
continue
item = robot_map[user_id]
if not item.get("user_status"):
item["user_status"] = existing[user_id]
filtered[user_id] = item
return filtered
def fetch_resources(conn, app_code: str) -> list[dict[str, Any]]:
with conn.cursor() as cursor:
cursor.execute(
"""
SELECT app_code, resource_id, resource_code, resource_type, name, status,
grantable, manager_grant_enabled, grant_strategy, wallet_asset_type,
wallet_asset_amount, price_type, coin_price, gift_point_amount,
COALESCE(CAST(usage_scope_json AS CHAR), '[]') AS usage_scope_json,
asset_url, preview_url, animation_url,
COALESCE(CAST(metadata_json AS CHAR), '{}') AS metadata_json,
sort_order, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
FROM resources
WHERE app_code = %s AND status = 'active'
ORDER BY sort_order ASC, resource_id ASC
""",
(app_code,),
)
return list(cursor.fetchall())
def parse_json_value(raw: str, fallback: Any) -> Any:
try:
return json.loads(raw or "")
except Exception:
return fallback
def badge_form(resource: dict[str, Any]) -> str:
payload = parse_json_value(str(resource.get("metadata_json") or "{}"), {})
if not isinstance(payload, dict):
return ""
return str(payload.get("badge_form") or "").strip().lower()
def is_vip_badge(resource: dict[str, Any]) -> bool:
code = str(resource.get("resource_code") or "").strip().lower()
name = str(resource.get("name") or "").strip().lower()
return str(resource.get("resource_type") or "") == "badge" and (code.startswith("vip") or name.startswith("vip"))
def build_catalog(resources: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
avatar_frames: list[dict[str, Any]] = []
long_badges: list[dict[str, Any]] = []
short_badges: list[dict[str, Any]] = []
for resource in resources:
resource_type = str(resource.get("resource_type") or "").strip()
resource_id = int(resource.get("resource_id") or 0)
if resource_type == "avatar_frame":
if resource_id in EXCLUDED_AVATAR_FRAME_IDS:
continue
if int(resource.get("coin_price") or 0) > ROBOT_AVATAR_FRAME_MAX_COIN_PRICE:
continue
avatar_frames.append(resource)
continue
if not is_vip_badge(resource):
continue
form = badge_form(resource)
if form in ("strip", "long"):
long_badges.append(resource)
elif form in ("tile", "short"):
short_badges.append(resource)
return {
"avatar_frame": avatar_frames,
"long_badge": long_badges,
"short_badge": short_badges,
}
def assert_catalog(catalog: dict[str, list[dict[str, Any]]]) -> None:
missing = [key for key, value in catalog.items() if not value]
if missing:
raise RuntimeError(f"robot appearance resource pool is incomplete: {', '.join(missing)}")
def choose_resource(items: list[dict[str, Any]], app_code: str, user_id: int, slot: str) -> dict[str, Any]:
index = int(stable_hash(f"{app_code}|{user_id}|{slot}")[:16], 16) % len(items)
return items[index]
def resource_snapshot(resource: dict[str, Any]) -> dict[str, Any]:
scopes = parse_json_value(str(resource.get("usage_scope_json") or "[]"), [])
if not isinstance(scopes, list):
scopes = []
return {
"AppCode": str(resource.get("app_code") or ""),
"ResourceID": int(resource.get("resource_id") or 0),
"ResourceCode": str(resource.get("resource_code") or ""),
"ResourceType": str(resource.get("resource_type") or ""),
"Name": str(resource.get("name") or ""),
"Status": str(resource.get("status") or ""),
"Grantable": bool(resource.get("grantable")),
"GrantStrategy": str(resource.get("grant_strategy") or ""),
"WalletAssetType": str(resource.get("wallet_asset_type") or ""),
"WalletAssetAmount": int(resource.get("wallet_asset_amount") or 0),
"PriceType": str(resource.get("price_type") or ""),
"CoinPrice": int(resource.get("coin_price") or 0),
"GiftPointAmount": int(resource.get("gift_point_amount") or 0),
"UsageScopes": [str(item) for item in scopes],
"AssetURL": str(resource.get("asset_url") or ""),
"PreviewURL": str(resource.get("preview_url") or ""),
"AnimationURL": str(resource.get("animation_url") or ""),
"MetadataJSON": str(resource.get("metadata_json") or "{}"),
"SortOrder": int(resource.get("sort_order") or 0),
"CreatedByUserID": int(resource.get("created_by_user_id") or 0),
"UpdatedByUserID": int(resource.get("updated_by_user_id") or 0),
"CreatedAtMS": int(resource.get("created_at_ms") or 0),
"UpdatedAtMS": int(resource.get("updated_at_ms") or 0),
"ManagerGrantEnabled": bool(resource.get("manager_grant_enabled")),
}
def grant_request_hash(app_code: str, user_id: int, resource_id: int, reason: str, operator_user_id: int) -> str:
return stable_hash(
f"resource_grant|{normalize_app_code(app_code)}|{user_id}|{resource_id}|1|{ROBOT_RESOURCE_DURATION_MS}|{reason}|{operator_user_id}|{GRANT_SOURCE}"
)
def resource_outbox_event_id(event_type: str, command_id: str, user_id: int, resource_id: int) -> str:
return "wev_" + stable_hash(f"{event_type}|{command_id}|{user_id}|{resource_id}")
def insert_wallet_outbox(
cursor,
*,
app_code: str,
event_type: str,
command_id: str,
user_id: int,
resource_id: int,
payload: dict[str, Any],
now_ms: int,
) -> None:
cursor.execute(
"""
INSERT IGNORE INTO wallet_outbox (
app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
available_delta, frozen_delta, payload, status, retry_count, created_at_ms, updated_at_ms
) VALUES (%s, %s, %s, '', %s, %s, %s, 0, 0, %s, %s, 0, %s, %s)
""",
(
app_code,
resource_outbox_event_id(event_type, command_id, user_id, resource_id),
event_type,
command_id,
user_id,
RESOURCE_OUTBOX_ASSET,
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
OUTBOX_STATUS_PENDING,
now_ms,
now_ms,
),
)
def active_entitlement(cursor, app_code: str, user_id: int, resource_id: int, now_ms: int) -> str:
cursor.execute(
"""
SELECT entitlement_id
FROM user_resource_entitlements
WHERE app_code = %s AND user_id = %s AND resource_id = %s
AND status = 'active'
AND effective_at_ms <= %s
AND (expires_at_ms = 0 OR expires_at_ms > %s)
AND remaining_quantity > 0
ORDER BY expires_at_ms DESC, created_at_ms DESC
LIMIT 1
FOR UPDATE
""",
(app_code, user_id, resource_id, now_ms, now_ms),
)
row = cursor.fetchone()
return str(row["entitlement_id"]) if row else ""
def ensure_entitlement(
cursor,
*,
app_code: str,
user_id: int,
resource: dict[str, Any],
slot: str,
operator_user_id: int,
reason: str,
run_id: str,
now_ms: int,
) -> tuple[str, bool, str]:
resource_id = int(resource["resource_id"])
existing = active_entitlement(cursor, app_code, user_id, resource_id, now_ms)
if existing:
return existing, False, ""
command_id = f"robot_appearance_refresh:{run_id}:{user_id}:{slot}:{resource_id}"
grant_id = prefixed_id("rgr_", f"{normalize_app_code(app_code)}|{command_id}")
entitlement_id = prefixed_id("ent_", f"{normalize_app_code(app_code)}|{grant_id}|{resource_id}|{now_ms}")
snapshot = resource_snapshot(resource)
snapshot_json = json.dumps(snapshot, ensure_ascii=False, separators=(",", ":"))
expires_at_ms = now_ms + ROBOT_RESOURCE_DURATION_MS
cursor.execute(
"""
INSERT INTO resource_grants (
app_code, grant_id, command_id, target_user_id, grant_source, grant_subject_type,
grant_subject_id, status, request_hash, group_snapshot_json, reason,
operator_user_id, created_at_ms, updated_at_ms
) VALUES (%s, %s, %s, %s, %s, 'resource', %s, %s, %s, NULL, %s, %s, %s, %s)
""",
(
app_code,
grant_id,
command_id,
user_id,
GRANT_SOURCE,
str(resource_id),
GRANT_STATUS_DONE,
grant_request_hash(app_code, user_id, resource_id, reason, operator_user_id),
reason,
operator_user_id,
now_ms,
now_ms,
),
)
cursor.execute(
"""
INSERT INTO user_resource_entitlements (
app_code, entitlement_id, user_id, resource_id, status, quantity, remaining_quantity,
effective_at_ms, expires_at_ms, source_grant_id, created_at_ms, updated_at_ms
) VALUES (%s, %s, %s, %s, %s, 1, 1, %s, %s, %s, %s, %s)
""",
(app_code, entitlement_id, user_id, resource_id, ENTITLEMENT_STATUS_ACTIVE, now_ms, expires_at_ms, grant_id, now_ms, now_ms),
)
cursor.execute(
"""
INSERT INTO resource_grant_items (
app_code, grant_id, resource_id, resource_snapshot_json, quantity, duration_ms,
result_type, wallet_transaction_id, entitlement_id, created_at_ms
) VALUES (%s, %s, %s, %s, 1, %s, %s, '', %s, %s)
""",
(app_code, grant_id, resource_id, snapshot_json, ROBOT_RESOURCE_DURATION_MS, RESULT_ENTITLEMENT, entitlement_id, now_ms),
)
insert_wallet_outbox(
cursor,
app_code=app_code,
event_type="ResourceGranted",
command_id=command_id,
user_id=user_id,
resource_id=resource_id,
payload={"grant_id": grant_id, "resource": snapshot},
now_ms=now_ms,
)
return entitlement_id, True, grant_id
def entitlement_snapshot(
*,
app_code: str,
user_id: int,
entitlement_id: str,
resource: dict[str, Any],
grant_id: str,
now_ms: int,
) -> dict[str, Any]:
return {
"AppCode": app_code,
"EntitlementID": entitlement_id,
"UserID": user_id,
"ResourceID": int(resource["resource_id"]),
"Resource": resource_snapshot(resource),
"Status": ENTITLEMENT_STATUS_ACTIVE,
"Quantity": 1,
"RemainingQuantity": 1,
"EffectiveAtMS": now_ms,
"ExpiresAtMS": now_ms + ROBOT_RESOURCE_DURATION_MS,
"SourceGrantID": grant_id,
"CreatedAtMS": now_ms,
"UpdatedAtMS": now_ms,
"Equipped": True,
}
def replace_wallet_equipment(
conn,
*,
app_code: str,
user_id: int,
selected: dict[str, dict[str, Any]],
operator_user_id: int,
reason: str,
run_id: str,
) -> dict[str, Any]:
now_ms = current_ms()
created_grants = 0
equipped_rows = 0
cursor = conn.cursor()
try:
cursor.execute("START TRANSACTION")
cursor.execute(
"""
DELETE FROM user_resource_equipment
WHERE app_code = %s AND user_id = %s AND resource_type IN ('avatar_frame', 'badge')
""",
(app_code, user_id),
)
deleted_equipment = cursor.rowcount
for slot, resource in selected.items():
resource_type = str(resource["resource_type"])
resource_id = int(resource["resource_id"])
entitlement_id, created, grant_id = ensure_entitlement(
cursor,
app_code=app_code,
user_id=user_id,
resource=resource,
slot=slot,
operator_user_id=operator_user_id,
reason=reason,
run_id=run_id,
now_ms=now_ms,
)
if created:
created_grants += 1
cursor.execute(
"""
INSERT INTO user_resource_equipment (
app_code, user_id, resource_type, entitlement_id, resource_id, updated_at_ms
) VALUES (%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE resource_id = VALUES(resource_id), updated_at_ms = VALUES(updated_at_ms)
""",
(app_code, user_id, resource_type, entitlement_id, resource_id, now_ms),
)
equipped_rows += 1
event_command_id = f"robot_appearance_equip:{run_id}:{user_id}:{slot}:{resource_id}"
insert_wallet_outbox(
cursor,
app_code=app_code,
event_type="UserResourceChanged",
command_id=event_command_id,
user_id=user_id,
resource_id=resource_id,
payload={
"action": "equip",
"app_code": app_code,
"user_id": user_id,
"resource_id": resource_id,
"resource_type": resource_type,
"entitlement": entitlement_snapshot(
app_code=app_code,
user_id=user_id,
entitlement_id=entitlement_id,
resource=resource,
grant_id=grant_id,
now_ms=now_ms,
),
"updated_at_ms": now_ms,
"event_command": event_command_id,
"source": "robot_appearance_refresh",
},
now_ms=now_ms,
)
conn.commit()
return {"created_grants": created_grants, "equipped_rows": equipped_rows, "deleted_equipment": deleted_equipment}
except Exception:
conn.rollback()
raise
finally:
cursor.close()
def replace_activity_badge_slots(
conn,
*,
app_code: str,
user_id: int,
selected: dict[str, dict[str, Any]],
wallet_conn,
) -> dict[str, int]:
now_ms = current_ms()
long_badge = selected["long_badge"]
short_badge = selected["short_badge"]
long_entitlement = latest_equipped_entitlement(wallet_conn, app_code, user_id, int(long_badge["resource_id"]))
short_entitlement = latest_equipped_entitlement(wallet_conn, app_code, user_id, int(short_badge["resource_id"]))
cursor = conn.cursor()
try:
cursor.execute("START TRANSACTION")
cursor.execute(
f"""
DELETE FROM user_badge_display_slots
WHERE app_code = %s AND user_id = %s AND slot IN ({','.join(['%s'] * len(BADGE_DISPLAY_SLOTS))})
""",
(app_code, user_id, *BADGE_DISPLAY_SLOTS),
)
deleted = cursor.rowcount
inserts = [
("profile_strip", 1, "strip", int(long_badge["resource_id"]), long_entitlement, "robot_appearance_refresh:long"),
("profile_tile", 1, "tile", int(short_badge["resource_id"]), short_entitlement, "robot_appearance_refresh:tile"),
("honor_wall", 1, "tile", int(short_badge["resource_id"]), short_entitlement, "robot_appearance_refresh:honor"),
]
for slot, position, form, resource_id, entitlement_id, source_id in inserts:
cursor.execute(
"""
INSERT INTO user_badge_display_slots (
app_code, user_id, slot, position, badge_form, resource_id,
entitlement_id, source_type, source_id, pin_mode, updated_at_ms
) VALUES (%s, %s, %s, %s, %s, %s, %s, 'system_grant', %s, 'auto', %s)
ON DUPLICATE KEY UPDATE
badge_form = VALUES(badge_form),
resource_id = VALUES(resource_id),
entitlement_id = VALUES(entitlement_id),
source_type = VALUES(source_type),
source_id = VALUES(source_id),
pin_mode = VALUES(pin_mode),
updated_at_ms = VALUES(updated_at_ms)
""",
(app_code, user_id, slot, position, form, resource_id, entitlement_id, source_id, now_ms),
)
conn.commit()
return {"deleted_slots": deleted, "inserted_slots": len(inserts)}
except Exception:
conn.rollback()
raise
finally:
cursor.close()
def latest_equipped_entitlement(conn, app_code: str, user_id: int, resource_id: int) -> str:
with conn.cursor() as cursor:
cursor.execute(
"""
SELECT entitlement_id
FROM user_resource_equipment
WHERE app_code = %s AND user_id = %s AND resource_id = %s
ORDER BY updated_at_ms DESC
LIMIT 1
""",
(app_code, user_id, resource_id),
)
row = cursor.fetchone()
return str(row["entitlement_id"]) if row else ""
def equipment_summary(conn, app_code: str, user_ids: list[int], valid_frame_ids: set[int], valid_badge_ids: set[int]) -> dict[str, int]:
if not user_ids:
return {}
summary = {
"users_with_valid_avatar_frame": 0,
"invalid_avatar_frame_equipment_rows": 0,
"valid_badge_equipment_rows": 0,
"invalid_badge_equipment_rows": 0,
}
for chunk in chunks(user_ids, 500):
user_placeholders = ",".join(["%s"] * len(chunk))
frame_placeholders = ",".join(["%s"] * len(valid_frame_ids)) or "NULL"
badge_placeholders = ",".join(["%s"] * len(valid_badge_ids)) or "NULL"
with conn.cursor() as cursor:
cursor.execute(
f"""
SELECT
COUNT(DISTINCT CASE WHEN eq.resource_type = 'avatar_frame' AND eq.resource_id IN ({frame_placeholders}) THEN eq.user_id END) AS users_with_valid_avatar_frame,
SUM(CASE WHEN eq.resource_type = 'avatar_frame' AND eq.resource_id NOT IN ({frame_placeholders}) THEN 1 ELSE 0 END) AS invalid_avatar_frame_equipment_rows,
SUM(CASE WHEN eq.resource_type = 'badge' AND eq.resource_id IN ({badge_placeholders}) THEN 1 ELSE 0 END) AS valid_badge_equipment_rows,
SUM(CASE WHEN eq.resource_type = 'badge' AND eq.resource_id NOT IN ({badge_placeholders}) THEN 1 ELSE 0 END) AS invalid_badge_equipment_rows
FROM user_resource_equipment eq
WHERE eq.app_code = %s AND eq.user_id IN ({user_placeholders})
AND eq.resource_type IN ('avatar_frame', 'badge')
""",
[*valid_frame_ids, *valid_frame_ids, *valid_badge_ids, *valid_badge_ids, app_code, *chunk],
)
row = cursor.fetchone()
for key in summary:
summary[key] += int(row.get(key) or 0)
return summary
def activity_slot_summary(conn, app_code: str, user_ids: list[int], valid_long_ids: set[int], valid_short_ids: set[int]) -> dict[str, int]:
if not user_ids:
return {}
valid_by_slot = {
"profile_strip": valid_long_ids,
"profile_tile": valid_short_ids,
"honor_wall": valid_short_ids,
}
summary = {f"{slot}_valid_users": 0 for slot in BADGE_DISPLAY_SLOTS}
summary.update({f"{slot}_invalid_rows": 0 for slot in BADGE_DISPLAY_SLOTS})
for slot, valid_ids in valid_by_slot.items():
for chunk in chunks(user_ids, 500):
user_placeholders = ",".join(["%s"] * len(chunk))
valid_placeholders = ",".join(["%s"] * len(valid_ids)) or "NULL"
with conn.cursor() as cursor:
cursor.execute(
f"""
SELECT
COUNT(DISTINCT CASE WHEN resource_id IN ({valid_placeholders}) THEN user_id END) AS valid_users,
SUM(CASE WHEN resource_id NOT IN ({valid_placeholders}) THEN 1 ELSE 0 END) AS invalid_rows
FROM user_badge_display_slots
WHERE app_code = %s AND user_id IN ({user_placeholders}) AND slot = %s
""",
[*valid_ids, *valid_ids, app_code, *chunk, slot],
)
row = cursor.fetchone()
summary[f"{slot}_valid_users"] += int(row.get("valid_users") or 0)
summary[f"{slot}_invalid_rows"] += int(row.get("invalid_rows") or 0)
return summary
def chunks(values: list[int], size: int):
for index in range(0, len(values), size):
yield values[index : index + size]
def sample_assignments(app_code: str, user_ids: list[int], catalog: dict[str, list[dict[str, Any]]], limit: int = 10) -> list[dict[str, Any]]:
samples = []
for user_id in user_ids[:limit]:
selected = select_for_user(app_code, user_id, catalog)
samples.append(
{
"user_id": user_id,
"avatar_frame": compact_resource(selected["avatar_frame"]),
"long_badge": compact_resource(selected["long_badge"]),
"short_badge": compact_resource(selected["short_badge"]),
}
)
return samples
def select_for_user(app_code: str, user_id: int, catalog: dict[str, list[dict[str, Any]]]) -> dict[str, dict[str, Any]]:
return {
"avatar_frame": choose_resource(catalog["avatar_frame"], app_code, user_id, "avatar_frame"),
"long_badge": choose_resource(catalog["long_badge"], app_code, user_id, "long_badge"),
"short_badge": choose_resource(catalog["short_badge"], app_code, user_id, "short_badge"),
}
def compact_resource(resource: dict[str, Any]) -> dict[str, Any]:
return {
"resource_id": int(resource["resource_id"]),
"resource_code": resource["resource_code"],
"name": resource["name"],
"coin_price": int(resource.get("coin_price") or 0),
"badge_form": badge_form(resource),
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Refresh all robot avatar frames and VIP badges.")
parser.add_argument("--user-config", default="services/user-service/configs/config.yaml")
parser.add_argument("--wallet-config", default="services/wallet-service/configs/config.yaml")
parser.add_argument("--robot-config", default="services/robot-service/configs/config.yaml")
parser.add_argument("--activity-config", default="services/activity-service/configs/config.yaml")
parser.add_argument("--app-code", default=APP_CODE_DEFAULT)
parser.add_argument("--robot-source", default=ROBOT_SOURCE_DEFAULT)
parser.add_argument("--operator-user-id", type=int, default=0)
parser.add_argument("--reason", default="refresh robot avatar frames and vip badges")
parser.add_argument("--progress-every", type=int, default=50)
parser.add_argument("--skip-activity", action="store_true")
parser.add_argument("--yes", action="store_true", help="Actually write wallet/activity changes.")
return parser.parse_args()
def main() -> int:
args = parse_args()
app_code = normalize_app_code(args.app_code)
if args.yes and args.operator_user_id <= 0:
raise RuntimeError("--operator-user-id is required with --yes")
if args.yes and not args.reason.strip():
raise RuntimeError("--reason is required with --yes")
user_conn = connect_from_config(args.user_config)
wallet_conn = connect_from_config(args.wallet_config)
robot_conn = connect_from_config(args.robot_config) if args.robot_config.strip() else None
activity_conn = None if args.skip_activity else connect_from_config(args.activity_config)
try:
source_robots = fetch_user_source_robots(user_conn, app_code, args.robot_source.strip())
pool_robots = fetch_robot_pool_users(robot_conn, app_code) if robot_conn else {}
robots = filter_existing_users(user_conn, app_code, merge_robot_sources(source_robots, pool_robots))
user_ids = sorted(robots)
resources = fetch_resources(wallet_conn, app_code)
catalog = build_catalog(resources)
assert_catalog(catalog)
valid_frame_ids = {int(item["resource_id"]) for item in catalog["avatar_frame"]}
valid_long_ids = {int(item["resource_id"]) for item in catalog["long_badge"]}
valid_short_ids = {int(item["resource_id"]) for item in catalog["short_badge"]}
valid_badge_ids = valid_long_ids | valid_short_ids
before_wallet = equipment_summary(wallet_conn, app_code, user_ids, valid_frame_ids, valid_badge_ids)
before_activity = (
activity_slot_summary(activity_conn, app_code, user_ids, valid_long_ids, valid_short_ids) if activity_conn else {}
)
summary = {
"mode": "execute" if args.yes else "dry_run",
"app_code": app_code,
"robot_user_count": len(user_ids),
"source_user_count": len(source_robots),
"robot_pool_user_count": len(pool_robots),
"resource_pool": {key: [compact_resource(item) for item in value] for key, value in catalog.items()},
"before_wallet": before_wallet,
"before_activity": before_activity,
"sample_assignments": sample_assignments(app_code, user_ids, catalog),
}
print(json.dumps(summary, ensure_ascii=False, indent=2))
if not args.yes:
print("dry-run only; rerun with --yes --operator-user-id <id> to write changes")
return 0
run_id = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + "_" + uuid.uuid4().hex[:12]
changed = {
"run_id": run_id,
"processed": 0,
"created_grants": 0,
"equipped_rows": 0,
"deleted_equipment": 0,
"deleted_activity_slots": 0,
"inserted_activity_slots": 0,
}
for user_id in user_ids:
selected = select_for_user(app_code, user_id, catalog)
wallet_result = replace_wallet_equipment(
wallet_conn,
app_code=app_code,
user_id=user_id,
selected=selected,
operator_user_id=args.operator_user_id,
reason=args.reason.strip(),
run_id=run_id,
)
changed["created_grants"] += wallet_result["created_grants"]
changed["equipped_rows"] += wallet_result["equipped_rows"]
changed["deleted_equipment"] += wallet_result["deleted_equipment"]
if activity_conn:
activity_result = replace_activity_badge_slots(
activity_conn,
app_code=app_code,
user_id=user_id,
selected=selected,
wallet_conn=wallet_conn,
)
changed["deleted_activity_slots"] += activity_result["deleted_slots"]
changed["inserted_activity_slots"] += activity_result["inserted_slots"]
changed["processed"] += 1
if args.progress_every > 0 and (changed["processed"] % args.progress_every == 0 or changed["processed"] == len(user_ids)):
print(json.dumps(changed, ensure_ascii=False))
after_wallet = equipment_summary(wallet_conn, app_code, user_ids, valid_frame_ids, valid_badge_ids)
after_activity = (
activity_slot_summary(activity_conn, app_code, user_ids, valid_long_ids, valid_short_ids) if activity_conn else {}
)
print(
json.dumps(
{
"ok": True,
**changed,
"after_wallet": after_wallet,
"after_activity": after_activity,
},
ensure_ascii=False,
indent=2,
)
)
return 0
finally:
if robot_conn:
robot_conn.close()
if activity_conn:
activity_conn.close()
wallet_conn.close()
user_conn.close()
if __name__ == "__main__":
raise SystemExit(main())