mirror of
https://github.com/Noettore/lagomareGateKeeperBot.git
synced 2025-10-14 19:16:40 +02:00
100 lines
4.6 KiB
Python
100 lines
4.6 KiB
Python
from telegram import Update
|
|
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes
|
|
from config import BOT_TOKEN
|
|
from access_control import can_open_gate, grant_access, get_grantor
|
|
from gates import get_name, open_gate
|
|
from users import get_role, set_credentials, get_credentials, get_grantor_credentials, get_admins, update_user
|
|
|
|
async def updateuser(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
user_id = str(update.effective_user.id)
|
|
username = update.effective_user.username
|
|
fullname = update.effective_user.full_name
|
|
update_user(user_id, username, fullname)
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
await update.message.reply_text("Welcome to GatekeeperBot! Use `/setcredentials` to configure your access.")
|
|
|
|
async def setcredentials(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
user_id = str(update.effective_user.id)
|
|
args = context.args
|
|
if len(args) != 2:
|
|
return await update.message.reply_text("Usage: `/setcredentials <username> <password>`")
|
|
role = get_role(user_id)
|
|
if role not in ("admin", "member"):
|
|
return await update.message.reply_text("Only members or admins can set credentials.")
|
|
set_credentials(user_id, args[0], args[1])
|
|
await update.message.reply_text("Credentials saved.")
|
|
|
|
async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
user_id = str(update.effective_user.id)
|
|
args = context.args
|
|
if not args:
|
|
return await update.message.reply_text("Usage: `/opengate <gate>`")
|
|
gate = args[0]
|
|
role = get_role(user_id)
|
|
if role in ("admin", "member"):
|
|
creds = get_credentials(user_id)
|
|
if not creds:
|
|
return await update.message.reply_text("Please set your credentials with `/setcredentials` first.")
|
|
elif role == "guest" and can_open_gate(user_id, gate):
|
|
creds = get_grantor_credentials(user_id, gate)
|
|
if not creds:
|
|
return await update.message.reply_text("No valid grantor credentials available.")
|
|
grantor = get_grantor(user_id, gate)
|
|
if grantor:
|
|
try:
|
|
await context.bot.send_message(chat_id=grantor, text=f"Guest {user_id} opened {get_name(gate)}")
|
|
except Exception as e:
|
|
print(f"Failed to notify {grantor} that guest {user_id} opened {get_name(gate)}: {e}")
|
|
else:
|
|
return await update.message.reply_text("Access denied.")
|
|
|
|
if open_gate(gate, creds):
|
|
return await update.message.reply_text(f"Gate {get_name(gate)} opened!")
|
|
await update.message.reply_text(f"ERROR: Cannot open gate {get_name(gate)}")
|
|
|
|
async def requestaccess(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
user_id = str(update.effective_user.id)
|
|
role = get_role(user_id)
|
|
if role not in ("guest"):
|
|
return await update.message.reply_text("Only guests can request access.")
|
|
if not context.args:
|
|
return await update.message.reply_text("Usage: `/requestaccess <gate>`")
|
|
gate = context.args[0]
|
|
requester = update.effective_user.username or update.effective_user.full_name
|
|
text = (f"Access request: @{requester} ({user_id}) requests access to *{gate}*.\nUse `/approve {user_id} {gate} 60` to grant access.")
|
|
await update.message.reply_text("Your request has been submitted.")
|
|
admins = get_admins()
|
|
for admin_id in admins:
|
|
try:
|
|
await context.bot.send_message(chat_id=admin_id, text=text, parse_mode="Markdown")
|
|
except:
|
|
pass
|
|
|
|
async def approve(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
approver_id = str(update.effective_user.id)
|
|
if get_role(approver_id) != "admin":
|
|
return await update.message.reply_text("Only admins can approve access.")
|
|
try:
|
|
user_id = context.args[0]
|
|
gate = context.args[1]
|
|
expires_at = context.args[2]
|
|
grant_access(user_id, gate, expires_at, grantor_id=approver_id)
|
|
await update.message.reply_text(f"Access to {gate} granted for user {user_id}.")
|
|
#TODO: notify guest of granted access
|
|
except:
|
|
await update.message.reply_text("Usage: `/approve <user_id> <gate|all> <expires_at:YYYY-MM-DDTHH:MM:SSZ>`")
|
|
|
|
def main():
|
|
app = Application.builder().token(BOT_TOKEN).build()
|
|
app.add_handler(MessageHandler(None, updateuser), group=1)
|
|
app.add_handler(CommandHandler("start", start))
|
|
app.add_handler(CommandHandler("setcredentials", setcredentials))
|
|
app.add_handler(CommandHandler("opengate", opengate))
|
|
app.add_handler(CommandHandler("requestaccess", requestaccess))
|
|
app.add_handler(CommandHandler("approve", approve))
|
|
app.run_polling()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|