ad3laid3/disco-reaper
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

generate alias for migrated users

Browse source
This commit is contained in:
rambros 2026-03-19 21:30:32 +05:30
parent d9b50eeb89
commit 3a86487fc2
6 changed files with 1324 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -37,7 +37,6 @@ config.yaml
# Logs and State
*.log
*.json
# Temporary Test Scripts
tmp/

168
src/core/database.py
View file

@ -1,5 +1,7 @@
import sqlite3
import logging
import json
import random
from pathlib import Path
from typing import Optional, Dict, Any
import threading
@ -83,9 +85,9 @@ class MigrationDatabase:
except sqlite3.OperationalError:
pass # Already exists
# Table for entity mappings (channels, roles, etc.)
# Table for server entity mappings (channels, roles, categories)
cursor.execute("""
CREATE TABLE IF NOT EXISTS entity_mappings (
CREATE TABLE IF NOT EXISTS server_mappings (
category TEXT,
source_id TEXT,
target_id TEXT,
@ -93,6 +95,36 @@ class MigrationDatabase:
)
""")
# Table for asset mappings (emojis, stickers)
cursor.execute("""
CREATE TABLE IF NOT EXISTS asset_mappings (
category TEXT,
source_id TEXT,
target_id TEXT,
PRIMARY KEY (category, source_id)
)
""")
# Migrate old entity_mappings if it exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='entity_mappings'")
if cursor.fetchone():
# Copy channels, categories, roles to server_mappings
cursor.execute("""
INSERT OR IGNORE INTO server_mappings (category, source_id, target_id)
SELECT category, source_id, target_id FROM entity_mappings
WHERE category IN ('channel', 'category', 'role')
""")
# Copy emojis, stickers to asset_mappings
cursor.execute("""
INSERT OR IGNORE INTO asset_mappings (category, source_id, target_id)
SELECT category, source_id, target_id FROM entity_mappings
WHERE category IN ('emoji', 'sticker')
""")
# Drop old table
cursor.execute("DROP TABLE entity_mappings")
# Table for general metadata
cursor.execute("""
CREATE TABLE IF NOT EXISTS metadata (
@ -100,6 +132,14 @@ class MigrationDatabase:
value TEXT
)
""")
# Table for auto-generated user aliases (user_id -> alias)
cursor.execute("""
CREATE TABLE IF NOT EXISTS user_alias (
user_id TEXT PRIMARY KEY,
alias TEXT UNIQUE
)
""")
# Indexes for fast lookup by source message ID
cursor.execute("CREATE INDEX IF NOT EXISTS idx_message_mappings_source ON message_mappings (source_msg_id)")
@ -124,46 +164,146 @@ class MigrationDatabase:
).fetchone()
return row["target_msg_id"] if row else None
# --- New Entity Mapping Methods ---
# --- User Alias Methods ---
def set_entity_mapping(self, category: str, source_id: str, target_id: str):
def _generate_alias(self) -> str:
"""Generates a unique alias in the format {Adjective}{Name} from random_users.json."""
json_path = Path(__file__).parent.parent / "random_users.json"
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
names = data.get("names", [])
adjectives = data.get("adjectives", [])
conn = self._get_conn()
# Try random combinations until unique
for _ in range(10000):
alias = f"{random.choice(adjectives)}{random.choice(names)}"
# Check if this alias is already taken
row = conn.execute("SELECT user_id FROM user_alias WHERE alias = ?", (alias,)).fetchone()
if not row:
return alias
# Fallback: append a number just in case collision rate is too high
import time
return f"{random.choice(adjectives)}{random.choice(names)}{int(time.time()) % 10000}"
def get_or_create_user_alias(self, user_id: str) -> str:
"""Gets the existing alias for a user or generates and saves a new one."""
conn = self._get_conn()
# Check for existing alias
row = conn.execute("SELECT alias FROM user_alias WHERE user_id = ?", (str(user_id),)).fetchone()
if row:
return row["alias"]
# Generate new, uniquely constrained alias
# Using a simplistic retry loop in case of race-conditions, though lock-less SQLite handles this with errors
try:
new_alias = self._generate_alias()
conn.execute(
"INSERT INTO user_alias (user_id, alias) VALUES (?, ?)",
(str(user_id), new_alias)
)
conn.commit()
return new_alias
except sqlite3.IntegrityError:
# Race condition: someone else inserted a conflicting alias or this user ID
# Re-read or re-try
row = conn.execute("SELECT alias FROM user_alias WHERE user_id = ?", (str(user_id),)).fetchone()
if row: return row["alias"]
# Otherwise uniquely retry
new_alias = self._generate_alias()
conn.execute(
"INSERT OR REPLACE INTO user_alias (user_id, alias) VALUES (?, ?)",
(str(user_id), new_alias)
)
conn.commit()
return new_alias
# --- Server Mapping Methods ---
def set_server_mapping(self, category: str, source_id: str, target_id: str):
conn = self._get_conn()
conn.execute(
"INSERT OR REPLACE INTO entity_mappings (category, source_id, target_id) VALUES (?, ?, ?)",
"INSERT OR REPLACE INTO server_mappings (category, source_id, target_id) VALUES (?, ?, ?)",
(category, str(source_id), str(target_id))
)
conn.commit()
def get_entity_mapping(self, category: str, source_id: str) -> Optional[str]:
def get_server_mapping(self, category: str, source_id: str) -> Optional[str]:
conn = self._get_conn()
row = conn.execute(
"SELECT target_id FROM entity_mappings WHERE category = ? AND source_id = ?",
"SELECT target_id FROM server_mappings WHERE category = ? AND source_id = ?",
(category, str(source_id))
).fetchone()
return row["target_id"] if row else None
def get_all_entity_mappings(self, category: str) -> Dict[str, str]:
def get_all_server_mappings(self, category: str) -> Dict[str, str]:
conn = self._get_conn()
rows = conn.execute(
"SELECT source_id, target_id FROM entity_mappings WHERE category = ?",
"SELECT source_id, target_id FROM server_mappings WHERE category = ?",
(category,)
).fetchall()
return {row["source_id"]: row["target_id"] for row in rows}
def delete_entity_mapping(self, category: str, source_id: str):
def delete_server_mapping(self, category: str, source_id: str):
conn = self._get_conn()
conn.execute(
"DELETE FROM entity_mappings WHERE category = ? AND source_id = ?",
"DELETE FROM server_mappings WHERE category = ? AND source_id = ?",
(category, str(source_id))
)
conn.commit()
def clear_entities(self, category: str = None):
def clear_server_mappings(self, category: str = None):
conn = self._get_conn()
if category:
conn.execute("DELETE FROM entity_mappings WHERE category = ?", (category,))
conn.execute("DELETE FROM server_mappings WHERE category = ?", (category,))
else:
conn.execute("DELETE FROM entity_mappings")
conn.execute("DELETE FROM server_mappings")
conn.commit()
# --- Asset Mapping Methods ---
def set_asset_mapping(self, category: str, source_id: str, target_id: str):
conn = self._get_conn()
conn.execute(
"INSERT OR REPLACE INTO asset_mappings (category, source_id, target_id) VALUES (?, ?, ?)",
(category, str(source_id), str(target_id))
)
conn.commit()
def get_asset_mapping(self, category: str, source_id: str) -> Optional[str]:
conn = self._get_conn()
row = conn.execute(
"SELECT target_id FROM asset_mappings WHERE category = ? AND source_id = ?",
(category, str(source_id))
).fetchone()
return row["target_id"] if row else None
def get_all_asset_mappings(self, category: str) -> Dict[str, str]:
conn = self._get_conn()
rows = conn.execute(
"SELECT source_id, target_id FROM asset_mappings WHERE category = ?",
(category,)
).fetchall()
return {row["source_id"]: row["target_id"] for row in rows}
def delete_asset_mapping(self, category: str, source_id: str):
conn = self._get_conn()
conn.execute(
"DELETE FROM asset_mappings WHERE category = ? AND source_id = ?",
(category, str(source_id))
)
conn.commit()
def clear_asset_mappings(self, category: str = None):
conn = self._get_conn()
if category:
conn.execute("DELETE FROM asset_mappings WHERE category = ?", (category,))
else:
conn.execute("DELETE FROM asset_mappings")
conn.commit()
# --- Metadata Methods ---

153
src/core/state.py
View file

@ -20,88 +20,123 @@ class MigrationState:
return False
return True
# --- Type Specific Getters/Setters (Database Backed) ---
def set_channel_mapping(self, discord_id: str, target_id: str):
if self._ensure_db():
self.db.set_entity_mapping("channel", str(discord_id), str(target_id))
def get_target_channel_id(self, discord_id: str) -> str | None:
if self._ensure_db():
return self.db.get_entity_mapping("channel", str(discord_id))
# --- Type Specific Getters/Setters # --- Channel Mapping ---
def set_channel_mapping(self, discord_id: int | str, target_id: str):
"""Maps an original text/voice/forum channel ID to a minted server channel ID."""
if self.db:
self.db.set_server_mapping("channel", str(discord_id), str(target_id))
def get_target_channel_id(self, discord_id: int | str) -> str | None:
if self.db:
return self.db.get_server_mapping("channel", str(discord_id))
return None
def remove_channel_mapping(self, discord_id: int | str):
if self.db:
self.db.delete_server_mapping("channel", str(discord_id))
get_fluxer_channel_id = get_target_channel_id
set_target_channel_mapping = set_channel_mapping
def set_category_mapping(self, discord_id: str, target_id: str):
if self._ensure_db():
self.db.set_entity_mapping("category", str(discord_id), str(target_id))
def get_target_category_id(self, discord_id: str) -> str | None:
if self._ensure_db():
return self.db.get_entity_mapping("category", str(discord_id))
# --- Category Mapping ---
def set_category_mapping(self, discord_id: int | str, target_id: str):
"""Maps an original discord category ID to a Stoat category Group ID."""
if self.db:
self.db.set_server_mapping("category", str(discord_id), str(target_id))
def get_category_mapping(self, discord_id: int | str) -> str | None:
"""Returns the Stoat Group ID for a previously migrated category."""
if self.db:
return self.db.get_server_mapping("category", str(discord_id))
return None
def remove_category_mapping(self, discord_id: int | str):
if self.db:
self.db.delete_server_mapping("category", str(discord_id))
get_fluxer_category_id = get_target_category_id
get_fluxer_category_id = get_category_mapping
get_target_category_id = get_category_mapping
set_target_category_mapping = set_category_mapping
def set_role_mapping(self, discord_id: str, target_id: str):
if self._ensure_db():
self.db.set_entity_mapping("role", str(discord_id), str(target_id))
# --- Role Mapping ---
def set_role_mapping(self, discord_id: int | str, target_id: str):
"""Maps an original discord Role ID to a Stoat Role ID."""
if self.db:
self.db.set_server_mapping("role", str(discord_id), str(target_id))
def get_target_role_id(self, discord_id: str) -> str | None:
if self._ensure_db():
return self.db.get_entity_mapping("role", str(discord_id))
def get_role_mapping(self, discord_id: int | str) -> str | None:
"""Returns the target Role ID for a previously migrated Role."""
if self.db:
return self.db.get_server_mapping("role", str(discord_id))
return None
def remove_role_mapping(self, discord_id: int | str):
if self.db:
self.db.delete_server_mapping("role", str(discord_id))
get_fluxer_role_id = get_target_role_id
get_fluxer_role_id = get_role_mapping
get_target_role_id = get_role_mapping
set_target_role_mapping = set_role_mapping
def set_emoji_mapping(self, discord_id: str, target_id: str):
if self._ensure_db():
self.db.set_entity_mapping("emoji", str(discord_id), str(target_id))
def get_target_emoji_id(self, discord_id: str) -> str | None:
if self._ensure_db():
return self.db.get_entity_mapping("emoji", str(discord_id))
# --- Emoji Mapping ---
def set_emoji_mapping(self, discord_id: int | str, target_id: str):
"""Maps an original discord Custom Emoji ID to a minted Emoji ID/URL."""
if self.db:
self.db.set_asset_mapping("emoji", str(discord_id), str(target_id))
def get_emoji_mapping(self, discord_id: int | str) -> str | None:
if self.db:
return self.db.get_asset_mapping("emoji", str(discord_id))
return None
def remove_emoji_mapping(self, discord_id: int | str):
if self.db:
self.db.delete_asset_mapping("emoji", str(discord_id))
get_fluxer_emoji_id = get_target_emoji_id
get_fluxer_emoji_id = get_emoji_mapping
get_target_emoji_id = get_emoji_mapping
set_target_emoji_mapping = set_emoji_mapping
def set_sticker_mapping(self, discord_id: str, target_id: str):
if self._ensure_db():
self.db.set_entity_mapping("sticker", str(discord_id), str(target_id))
def get_target_sticker_id(self, discord_id: str) -> str | None:
if self._ensure_db():
return self.db.get_entity_mapping("sticker", str(discord_id))
# --- Sticker Mapping ---
def set_sticker_mapping(self, discord_id: int | str, target_id: str):
"""Maps an original discord Custom Sticker ID to a target URL or ID."""
if self.db:
self.db.set_asset_mapping("sticker", str(discord_id), str(target_id))
def get_sticker_mapping(self, discord_id: int | str) -> str | None:
if self.db:
return self.db.get_asset_mapping("sticker", str(discord_id))
return None
def remove_sticker_mapping(self, discord_id: int | str):
if self.db:
self.db.delete_asset_mapping("sticker", str(discord_id))
get_fluxer_sticker_id = get_target_sticker_id
get_fluxer_sticker_id = get_sticker_mapping
get_target_sticker_id = get_sticker_mapping
set_target_sticker_mapping = set_sticker_mapping
# --- Properties for backward compatibility ---
@property
def channel_map(self) -> Dict[str, str]:
return self.db.get_all_entity_mappings("channel") if self.db else {}
return self.db.get_all_server_mappings("channel") if self.db else {}
@property
def category_map(self) -> Dict[str, str]:
return self.db.get_all_entity_mappings("category") if self.db else {}
return self.db.get_all_server_mappings("category") if self.db else {}
@property
def role_map(self) -> Dict[str, str]:
return self.db.get_all_entity_mappings("role") if self.db else {}
return self.db.get_all_server_mappings("role") if self.db else {}
@property
def emoji_map(self) -> Dict[str, str]:
return self.db.get_all_entity_mappings("emoji") if self.db else {}
return self.db.get_all_asset_mappings("emoji") if self.db else {}
@property
def sticker_map(self) -> Dict[str, str]:
return self.db.get_all_entity_mappings("sticker") if self.db else {}
return self.db.get_all_asset_mappings("sticker") if self.db else {}
@property
def audit_log_channel(self) -> str | None:
@ -197,18 +232,18 @@ class MigrationState:
# --- Danger Zone Clearing ---
def clear_channel_mappings(self):
if self._ensure_db():
self.db.clear_entities("channel")
self.db.clear_entities("category")
if self.db:
self.db.clear_server_mappings("channel")
self.db.clear_server_mappings("category")
def clear_role_mappings(self):
if self._ensure_db():
self.db.clear_entities("role")
if self.db:
self.db.clear_server_mappings("role")
def clear_asset_mappings(self):
if self._ensure_db():
self.db.clear_entities("emoji")
self.db.clear_entities("sticker")
if self.db:
self.db.clear_asset_mappings("emoji")
self.db.clear_asset_mappings("sticker")
def clear_message_history(self):
if self.db:
@ -262,6 +297,12 @@ class MigrationState:
self.db = MigrationDatabase(db_path)
logger.info(f"Initialized SQLite database at {db_path}")
def get_user_alias(self, user_id: str) -> str | None:
"""Gets or generates a unique alias for a given user ID via the Migration Database."""
if self.db:
return self.db.get_or_create_user_alias(user_id)
return None
# No-op methods kept for compatibility with callers that might try to load/save JSON
def load(self): pass
def save_state(self): pass

3
src/fluxer/migrate_message.py
View file

@ -553,6 +553,9 @@ async def migrate_messages(
if avatar_url and not avatar_url.startswith("http"):
avatar_url = None
# Trigger alias generation/storage in DB without replacing the display name yet
context.state.get_user_alias(str(msg.author.id))
logger.debug(f"Fluxer: Calling send_message for Discord ID {msg.id}")
fluxer_msg_id = await context.fluxer_writer.send_message(
channel_id=target_channel_id,

1067
src/random_users.json Normal file
View file

File diff suppressed because it is too large Load diff

3
src/stoat/migrate_message.py
View file

@ -550,6 +550,9 @@ async def migrate_messages(
if avatar_url and not avatar_url.startswith("http"):
avatar_url = None
# Trigger alias generation/storage in DB without replacing the display name yet
context.state.get_user_alias(str(msg.author.id))
stoat_msg_id = await context.stoat_writer.send_message(
channel_id=target_channel_id,
author_name=msg.author.display_name,