Compare commits

..

1 Commits

Author SHA1 Message Date
shokollm
f7503ecd14 feat: add /admin command to list room admins
Implement /admin command as specified in issue #50:
- Lists all admin user IDs for the current room
- Output format: 'Room Admins:\n- @user1\n- @user2'
- Shows 'No admins configured for this room.' if none exist
- Available to everyone (no permission check needed)

Changes:
- Add cmd_admin function to commands.py
- Register CommandHandler('admin', cmd_admin) in bot.py
- Add /admin to command menu in post_init
- Update /help to include /admin command

Closes #50
2026-04-04 06:52:23 +00:00
7 changed files with 86 additions and 854 deletions

View File

@@ -4,29 +4,17 @@ import logging
import os
import sys
sys.path.insert(0, "/home/shoko/repositories/jigaido")
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
CallbackQueryHandler,
filters,
)
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from commands import (
cmd_add,
cmd_admin,
cmd_bounty,
cmd_delete,
cmd_delete_message,
cmd_edit,
cmd_help,
cmd_my,
cmd_recover,
cmd_show,
cmd_start,
cmd_timezone,
cmd_track,
cmd_untrack,
cmd_update,
@@ -38,13 +26,7 @@ logging.basicConfig(
)
log = logging.getLogger(__name__)
from config import config
BOT_TOKEN = config.bot_token or ""
async def error_handler(update, context):
log.error(f"Error: {context.error}")
BOT_TOKEN = os.environ.get("JIGAIDO_BOT_TOKEN", "")
def build_app() -> Application:
@@ -60,14 +42,9 @@ def build_app() -> Application:
app.add_handler(CommandHandler("delete", cmd_delete))
app.add_handler(CommandHandler("track", cmd_track))
app.add_handler(CommandHandler("untrack", cmd_untrack))
app.add_handler(CommandHandler("show", cmd_show))
app.add_handler(CommandHandler("timezone", cmd_timezone))
app.add_handler(CommandHandler("admin", cmd_admin))
app.add_handler(CommandHandler("recover", cmd_recover))
app.add_handler(CallbackQueryHandler(cmd_delete_message))
app.add_error_handler(error_handler)
app.add_handler(MessageHandler(filters.COMMAND, cmd_help))
return app
@@ -81,18 +58,13 @@ async def post_init(app: Application) -> None:
("edit", "Edit a bounty"),
("track", "Track a bounty"),
("untrack", "Stop tracking"),
("show", "Show bounty details"),
("timezone", "Get/set room timezone"),
("admin", "Manage admins"),
("recover", "Recover deleted bounties"),
("admin", "List room admins"),
("help", "Show help"),
]
)
def main() -> None:
import asyncio
if not BOT_TOKEN:
log.error("JIGAIDO_BOT_TOKEN environment variable not set.")
sys.exit(1)
@@ -101,11 +73,6 @@ def main() -> None:
app.post_init = post_init
log.info("JIGAIDO starting...")
# Python 3.14 compatibility: ensure event loop exists
try:
asyncio.get_event_loop()
except RuntimeError:
asyncio.set_event_loop(asyncio.new_event_loop())
app.run_polling(drop_pending_updates=True)

View File

@@ -1,16 +1,12 @@
"""Telegram command handlers for JIGAIDO - Thin wrappers around core services."""
import time
from datetime import datetime, timezone
from functools import wraps
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import dateparser
from telegram import Update
from telegram.ext import ContextTypes
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ParseMode
from adapters.storage.json_file import JsonFileRoomStorage, JsonFileTrackingStorage
from core.services import BountyService, TrackingService
@@ -23,34 +19,6 @@ TRACKING_SERVICE = TrackingService(TRACKING_STORAGE, ROOM_STORAGE)
TELEGRAM_BOT_USERNAME = "your_bot_username"
def format_due_date(due_date_ts: int | None, timezone_str: str) -> str:
"""Format due date as human-readable with timezone.
Examples:
No due date: (none shown)
Date only: 4 April 2026
Date + time: 4 April 2026 14:30
With timezone: 4 April 2026 14:30 (Asia/Jakarta)
"""
if not due_date_ts:
return ""
try:
tz = ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
tz = ZoneInfo("UTC")
dt = datetime.fromtimestamp(due_date_ts, tz=tz)
date_str = dt.strftime("%-d %B %Y")
if dt.hour != 0 or dt.minute != 0:
date_str += f" {dt.strftime('%H:%M')}"
date_str += f" ({timezone_str})"
return date_str
def extract_args(text: str) -> list[str]:
if not text:
return []
@@ -58,176 +26,50 @@ def extract_args(text: str) -> list[str]:
return tokens[1:] if len(tokens) > 1 else []
def parse_args(
args: list[str],
timezone_str: str = "UTC",
) -> tuple[Optional[str], Optional[str], Optional[int], bool, bool]:
def parse_args(args: list[str]) -> tuple[Optional[str], Optional[str], Optional[int]]:
text = None
link = None
due_date_ts = None
clear_link = False
clear_date = False
try:
tz = ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
tz = ZoneInfo("UTC")
def is_url(s: str) -> bool:
if not s:
return False
if s.startswith("http://") or s.startswith("https://"):
return True
if "." in s and " " not in s:
return True
return False
def is_time(s: str) -> bool:
if not s or ":" not in s:
return False
parts = s.split(":")
if len(parts) != 2:
return False
try:
h, m = int(parts[0]), int(parts[1])
return 0 <= h <= 23 and 0 <= m <= 59
except ValueError:
return False
def parse_date_with_tz(date_str: str) -> int | None:
parsed = dateparser.parse(date_str)
if parsed:
localized = parsed.replace(tzinfo=tz)
return int(localized.timestamp())
return None
i = 0
while i < len(args):
arg = args[i]
if arg == "-link":
if i + 1 < len(args) and not args[i + 1].startswith("-"):
link = args[i + 1]
i += 2
else:
clear_link = True
i += 1
elif arg == "-date":
if i + 1 < len(args) and not args[i + 1].startswith("-"):
due_date_ts = parse_date_with_tz(args[i + 1])
if due_date_ts is not None:
i += 2
if i < len(args) and is_time(args[i]):
time_parts = args[i].split(":")
due_date_ts += (
int(time_parts[0]) * 3600 + int(time_parts[1]) * 60
)
i += 1
else:
clear_date = True
i += 1
else:
clear_date = True
i += 1
elif not link and is_url(arg):
remaining = []
for arg in args:
if not link and (arg.startswith("http://") or arg.startswith("https://")):
link = arg
i += 1
elif due_date_ts is None:
due_date_ts = parse_date_with_tz(arg)
if due_date_ts is not None:
i += 1
if i < len(args) and is_time(args[i]):
time_parts = args[i].split(":")
due_date_ts += int(time_parts[0]) * 3600 + int(time_parts[1]) * 60
i += 1
parsed = dateparser.parse(arg)
if parsed:
due_date_ts = int(parsed.timestamp())
else:
i += 1
if text is None:
text = arg
remaining.append(arg)
else:
text = text + " " + arg
else:
i += 1
if text is None:
text = arg
else:
text = text + " " + arg
remaining.append(arg)
return text, link, due_date_ts, clear_link, clear_date
text = " ".join(remaining) if remaining else None
return text, link, due_date_ts
def format_bounty(
b, show_id: bool = True, slice_length: int = 0, timezone_str: str = "UTC"
) -> str:
def format_bounty(b, show_id: bool = True) -> str:
parts = []
if show_id:
parts.append(f"[#{b.id}]")
if b.text:
text = b.text
if slice_length > 0 and len(text) > slice_length:
text = text[:slice_length] + "..."
parts.append(text)
parts.append(b.text)
if b.link:
parts.append(f"🔗 {b.link}")
if b.due_date_ts:
try:
tz = ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
tz = ZoneInfo("UTC")
now_utc = int(time.time())
dt_now = datetime.now(tz)
dt_due = datetime.fromtimestamp(b.due_date_ts, tz=tz)
seconds_left = int((dt_due - dt_now).total_seconds())
hours_left = seconds_left // 3600
days_left = seconds_left // 86400
due_str = dt_due.strftime("%d %b %Y %H:%M")
if seconds_left < 0:
due_str = time.strftime("%Y-%m-%d", time.localtime(b.due_date_ts))
days_left = (b.due_date_ts - int(time.time())) // 86400
if days_left < 0:
parts.append(f"{due_str} (OVERDUE)")
elif hours_left < 48:
if hours_left < 1:
minutes_left = seconds_left // 60
parts.append(f"{due_str} ({minutes_left}m)")
else:
parts.append(f"{due_str} ({hours_left}h)")
elif days_left == 0:
parts.append(f"{due_str} (TODAY)")
else:
parts.append(f"{due_str} ({days_left}d)")
if b.created_by_user_id:
parts.append(f"by {b.created_by_user_id}")
return " | ".join(parts)
async def cmd_delete_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
if not query:
return
data = query.data
if not data.startswith("del_msg:"):
return
parts = data.split(":")
if len(parts) != 2:
return
try:
expected_user_id = int(parts[1])
except ValueError:
return
user_id = get_user_id(update)
if user_id != expected_user_id:
await query.answer("You can't delete this message", show_alert=True)
return
try:
await query.message.delete()
await query.answer("Deleted")
except Exception as e:
await query.answer(f"Could not delete: {e}", show_alert=True)
def is_group(update: Update) -> bool:
return update.effective_chat.type != "private"
@@ -253,69 +95,14 @@ def get_room_id(update: Update) -> int:
async def cmd_bounty(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
room_id = get_room_id(update)
user_id = get_user_id(update)
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
args = extract_args(update.message.text)
bounties = BOUNTY_SERVICE.list_bounties(room_id)
show_all = "all" in args
args = [a for a in args if a != "all"]
try:
limit = int(args[0]) if args else 5
except (ValueError, IndexError):
limit = 5
now = int(time.time())
cutoff_24h = now - 86400
all_bounties = BOUNTY_SERVICE.list_bounties(room_id)
def is_expired(b) -> bool:
return b.due_date_ts is not None and b.due_date_ts < cutoff_24h
def sort_key(b):
if b.due_date_ts is not None:
return (0, b.due_date_ts)
return (1, b.created_at)
filtered_bounties = [b for b in all_bounties if not is_expired(b) or show_all]
filtered_bounties.sort(key=sort_key)
total_count = len(filtered_bounties)
displayed_bounties = filtered_bounties[:limit]
if not displayed_bounties:
if show_all:
if not bounties:
await update.message.reply_text("No bounties yet.")
else:
await update.message.reply_text(
"No active bounties. Use /bounty all to show expired."
)
return
lines = []
if limit < total_count:
lines.append(f"Showing {limit} of {total_count} bounties:")
slice_length = 40
elif show_all and total_count > limit:
lines.append(f"Showing {limit} of {total_count} bounties (including expired):")
slice_length = 40
else:
lines.append(f"Showing {total_count} bounties:")
slice_length = 0
for b in displayed_bounties:
lines.append(
format_bounty(
b, show_id=True, slice_length=slice_length, timezone_str=timezone_str
)
)
keyboard = [[InlineKeyboardButton("🗑️ Delete", callback_data=f"del_msg:{user_id}")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"\n".join(lines), disable_web_page_preview=True, reply_markup=reply_markup
)
lines = [format_bounty(b, show_id=True) for b in bounties]
await update.message.reply_text("\n".join(lines), disable_web_page_preview=True)
async def cmd_my(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
@@ -324,13 +111,10 @@ async def cmd_my(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
if is_group(update):
group_id = get_group_id(update)
bounties = TRACKING_SERVICE.get_tracked_bounties(group_id, user_id)
room_id = group_id
else:
room_id = get_room_id(update)
bounties = BOUNTY_SERVICE.list_bounties(room_id)
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
if not bounties:
msg = (
"You are not tracking any bounties."
@@ -340,9 +124,7 @@ async def cmd_my(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text(msg)
return
lines = [
format_bounty(b, show_id=True, timezone_str=timezone_str) for b in bounties
]
lines = [format_bounty(b, show_id=True) for b in bounties]
await update.message.reply_text("\n".join(lines), disable_web_page_preview=True)
@@ -355,38 +137,26 @@ async def cmd_add(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
)
return
user_id = get_user_id(update)
room_id = get_room_id(update)
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
effective_user = update.effective_user
username = effective_user.username or effective_user.first_name or None
text, link, due_date_ts, _, _ = parse_args(args, timezone_str)
text, link, due_date_ts = parse_args(args)
if not text and not link:
await update.message.reply_text("A bounty needs at least text or a link.")
return
try:
user_id = get_user_id(update)
room_id = get_room_id(update)
bounty = BOUNTY_SERVICE.add_bounty(
room_id=room_id,
user_id=user_id,
text=text,
link=link,
due_date_ts=due_date_ts,
created_by_username=username,
)
except PermissionError as e:
await update.message.reply_text(f"{e}")
return
except ValueError as e:
await update.message.reply_text(f"{e}")
return
due_str = ""
if due_date_ts:
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
due_str = f" | Due: {format_due_date(due_date_ts, timezone_str)}"
due_str = f" | Due: {time.strftime('%Y-%m-%d', time.localtime(due_date_ts))}"
await update.message.reply_text(
f"✅ Bounty added (#{bounty.id}){due_str}",
disable_web_page_preview=True,
@@ -397,14 +167,7 @@ async def cmd_update(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
if len(args) < 1:
await update.message.reply_text(
"Usage: /update <bounty_id> [text] [link] [due_date]\n"
" /update <bounty_id> -link [<url>] - clear or set link\n"
" /update <bounty_id> -date [<date>] - clear or set date\n"
"Examples:\n"
" /update 1 new text - update text only\n"
" /update 1 -link - clear link\n"
" /update 1 -link https://... - set link\n"
" /update 1 -link -date - clear both link and date"
"Usage: /update <bounty_id> [text] [link] [due_date]"
)
return
@@ -414,30 +177,13 @@ async def cmd_update(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Invalid bounty ID.")
return
user_id = get_user_id(update)
room_id = get_room_id(update)
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
try:
tz = ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
tz = ZoneInfo("UTC")
text, link, due_date_ts, clear_link, clear_date = parse_args(args[1:], timezone_str)
if (
not text
and not link
and due_date_ts is None
and not clear_link
and not clear_date
):
text, link, due_date_ts = parse_args(args[1:])
if not text and not link and due_date_ts is None:
await update.message.reply_text("Nothing to update.")
return
old_bounty = BOUNTY_SERVICE.get_bounty(room_id, bounty_id)
if not old_bounty:
await update.message.reply_text("Bounty not found.")
return
user_id = get_user_id(update)
room_id = get_room_id(update)
try:
success = BOUNTY_SERVICE.update_bounty(
@@ -447,48 +193,12 @@ async def cmd_update(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
text=text,
link=link,
due_date_ts=due_date_ts,
clear_link=clear_link,
clear_due=clear_date,
)
except PermissionError as e:
await update.message.reply_text(f"{e}")
return
except ValueError as e:
await update.message.reply_text(f"{e}")
return
if success:
changes = []
if text is not None:
old_text = old_bounty.text or "(none)"
changes.append(f"Text: {old_text}{text}")
if link is not None:
old_link = old_bounty.link or "(none)"
changes.append(f"Link: {old_link}{link}")
if due_date_ts is not None:
old_date = (
format_due_date(old_bounty.due_date_ts, timezone_str)
if old_bounty.due_date_ts
else "(none)"
)
new_date = format_due_date(due_date_ts, timezone_str)
changes.append(f"Date: {old_date}{new_date}")
if clear_link:
old_link = old_bounty.link or "(none)"
changes.append(f"Link: {old_link} → (cleared)")
if clear_date:
old_date = (
format_due_date(old_bounty.due_date_ts, timezone_str)
if old_bounty.due_date_ts
else "(none)"
)
changes.append(f"Date: {old_date} → (cleared)")
if changes:
await update.message.reply_text(
f"✅ Bounty #{bounty_id} updated:\n" + "\n".join(changes)
)
else:
await update.message.reply_text(f"✅ Bounty #{bounty_id} updated.")
else:
await update.message.reply_text("Bounty not found.")
@@ -500,11 +210,11 @@ cmd_edit = cmd_update
async def cmd_delete(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
if not args:
await update.message.reply_text("Usage: /delete <bounty_id> [bounty_id ...]")
await update.message.reply_text("Usage: /delete <bounty_id>")
return
try:
bounty_ids = [int(arg) for arg in args]
bounty_id = int(args[0])
except ValueError:
await update.message.reply_text("Invalid bounty ID.")
return
@@ -513,25 +223,19 @@ async def cmd_delete(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
room_id = get_room_id(update)
try:
results = BOUNTY_SERVICE.delete_bounties(
success = BOUNTY_SERVICE.delete_bounty(
room_id=room_id,
bounty_ids=bounty_ids,
bounty_id=bounty_id,
user_id=user_id,
)
except PermissionError as e:
await update.message.reply_text(f"{e}")
return
response_lines = []
for bounty_id, result in results.items():
if result == "deleted":
response_lines.append(f"Bounty #{bounty_id} deleted.")
elif result == "not_found":
response_lines.append(f"⛔ Bounty #{bounty_id} not found.")
elif result == "permission_denied":
response_lines.append(f"⛔ Bounty #{bounty_id} permission denied.")
await update.message.reply_text("\n".join(response_lines))
if success:
await update.message.reply_text(f"✅ Bounty #{bounty_id} deleted.")
else:
await update.message.reply_text("Bounty not found.")
async def cmd_track(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
@@ -588,29 +292,7 @@ async def cmd_untrack(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
user_id = get_user_id(update)
room_id = get_room_id(update)
if is_group(update):
try:
chat_member = await ctx.bot.get_chat_member(room_id, user_id)
if chat_member.status == "creator" and not BOUNTY_SERVICE.is_admin(
room_id, user_id
):
BOUNTY_SERVICE.add_admin(room_id, user_id, user_id)
await update.message.reply_text(
"👻 JIGAIDO is watching.\n\n"
"This group's bounty list is now active.\n"
"You are now an admin (group creator).\n"
"/bounty — list bounties\n"
"/add — create a bounty\n"
"/track — track a bounty\n"
"/my — your tracked bounties"
)
return
except Exception:
pass
await update.message.reply_text(
"👻 JIGAIDO is watching.\n\n"
"This group's bounty list is now active.\n"
@@ -620,9 +302,6 @@ async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
"/my — your tracked bounties"
)
else:
if not BOUNTY_SERVICE.is_admin(room_id, user_id):
BOUNTY_SERVICE.add_admin(room_id, user_id, user_id)
await update.message.reply_text(
"👻 JIGAIDO activated.\n\n"
"Personal bounty list ready.\n"
@@ -632,278 +311,33 @@ async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
)
async def cmd_show(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
if not args:
await update.message.reply_text("Usage: /show <bounty_id>")
return
try:
bounty_id = int(args[0])
except ValueError:
await update.message.reply_text("Invalid bounty ID.")
return
room_id = get_room_id(update)
bounty = BOUNTY_SERVICE.get_bounty(room_id, bounty_id)
if not bounty:
await update.message.reply_text("Bounty not found.")
return
timezone_str = BOUNTY_SERVICE.get_timezone(room_id)
try:
tz = ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
tz = ZoneInfo("UTC")
lines = []
title = bounty.text or "(no text)"
lines.append(f"[#{bounty.id}] {title}")
if bounty.link:
lines.append(f"🔗 {bounty.link}")
if bounty.due_date_ts:
dt_due = datetime.fromtimestamp(bounty.due_date_ts, tz=tz)
due_str = dt_due.strftime("%d %B %Y %H:%M")
lines.append(f"📅 {due_str} ({timezone_str})")
if bounty.created_by_username:
lines.append(
f'👤 <a href="tg://user?id={bounty.created_by_user_id}">{bounty.created_by_username}</a>'
)
dt_created = datetime.fromtimestamp(bounty.created_at, tz=tz)
created_str = dt_created.strftime("%Y-%m-%d %H:%M")
lines.append(f"📌 Created: {created_str}")
async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text(
"\n".join(lines), disable_web_page_preview=True, parse_mode=ParseMode.HTML
"👻 JIGAIDO Commands:\n\n"
"/bounty — list all bounties\n"
"/my — bounties you're tracking\n"
"/add <text> [link] [due] — add bounty\n"
"/update <id> [text> [link] [due] — update bounty\n"
"/delete <id> — delete bounty\n"
"/track <id> — track a bounty (groups only)\n"
"/untrack <id> — stop tracking (groups only)\n"
"/admin — list room admins\n"
"/start — re-initialize\n"
"/help — this message",
disable_web_page_preview=True,
)
async def cmd_timezone(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
room_id = get_room_id(update)
user_id = get_user_id(update)
if not args:
current_tz = BOUNTY_SERVICE.get_timezone(room_id)
await update.message.reply_text(f"Current timezone: {current_tz}")
return
timezone_str = args[0]
try:
ZoneInfo(timezone_str)
except (KeyError, ZoneInfoNotFoundError):
await update.message.reply_text(
"⛔ Invalid timezone. Use IANA format (e.g., Asia/Jakarta)"
)
return
try:
BOUNTY_SERVICE.set_timezone(room_id, timezone_str, user_id)
except PermissionError as e:
await update.message.reply_text(f"{e}")
return
await update.message.reply_text(f"✅ Timezone set to {timezone_str}.")
async def cmd_recover(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
room_id = get_room_id(update)
user_id = get_user_id(update)
if not BOUNTY_SERVICE.is_admin(room_id, user_id):
await update.message.reply_text("⛔ Only admins can perform this action.")
return
if not args:
deleted_bounties = BOUNTY_SERVICE.list_deleted_bounties(room_id)
if not deleted_bounties:
await update.message.reply_text("No recoverable bounties.")
return
deleted_bounties.sort(key=lambda b: b.deleted_at or 0, reverse=True)
lines = ["Recoverable bounties:"]
for b in deleted_bounties[:10]:
text = (
b.text[:40] + "..."
if b.text and len(b.text) > 40
else (b.text or "(no text)")
)
link_str = f" | 🔗 {b.link}" if b.link else ""
deleted_str = (
time.strftime("%d %b %Y", time.localtime(b.deleted_at))
if b.deleted_at
else "unknown"
)
lines.append(f"[#{b.id}] {text}{link_str} | 🗑️ Deleted {deleted_str}")
await update.message.reply_text("\n".join(lines), disable_web_page_preview=True)
return
try:
bounty_ids = [int(arg) for arg in args]
except ValueError:
await update.message.reply_text("Invalid bounty ID.")
return
results = BOUNTY_SERVICE.recover_bounties(room_id, bounty_ids, user_id)
response_lines = []
for bounty_id, result in results.items():
if result == "recovered":
response_lines.append(f"✅ Recovered bounty #{bounty_id}.")
elif result == "not_found":
response_lines.append(f"⛔ Bounty #{bounty_id} not found.")
elif result == "not_deleted":
response_lines.append(f"⛔ Bounty #{bounty_id} is not deleted.")
elif result == "permission_denied":
response_lines.append(f"⛔ Bounty #{bounty_id} permission denied.")
await update.message.reply_text("\n".join(response_lines))
async def _find_user_id_by_username(
ctx: ContextTypes.DEFAULT_TYPE, username: str
) -> int | None:
"""Find user_id by username using Telegram API."""
import logging
log = logging.getLogger(__name__)
try:
chat = await ctx.bot.get_chat(f"@{username}")
log.info(f"Found user {username}: {chat.id}")
return chat.id
except Exception as e:
log.error(f"Failed to find user @{username}: {e}")
return None
def _find_username_by_user_id(room_id: int, user_id: int) -> str | None:
"""Find username by user_id from bounty creators in the room."""
bounties = BOUNTY_SERVICE.list_bounties(room_id)
for bounty in bounties:
if bounty.created_by_user_id == user_id:
return bounty.created_by_username
return None
async def cmd_admin(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
args = extract_args(update.message.text)
if not args:
admins = BOUNTY_SERVICE.list_admins(get_room_id(update))
if not admins:
await update.message.reply_text("No admins in this room.")
return
admin_mentions = []
for admin_id in admins:
username = _find_username_by_user_id(get_room_id(update), admin_id)
if username:
admin_mentions.append(
f'<a href="tg://user?id={admin_id}">@{username}</a>'
)
else:
admin_mentions.append(
f'<a href="tg://user?id={admin_id}">{admin_id}</a>'
)
await update.message.reply_text(
f"Room Admins:\n" + "\n".join(f"- {m}" for m in admin_mentions),
parse_mode=ParseMode.HTML,
)
return
if args[0] not in ("add", "remove"):
await update.message.reply_text(
"Usage:\n"
"/admin — list admins\n"
"/admin add @username — add admin\n"
"/admin remove @username — remove admin"
)
return
subcommand = args[0]
if len(args) < 2:
await update.message.reply_text(f"Usage: /admin {subcommand} @username")
return
raw_username = args[1]
if not raw_username.startswith("@"):
await update.message.reply_text(f"Usage: /admin {subcommand} @username")
return
username = raw_username[1:]
user_id = get_user_id(update)
room_id = get_room_id(update)
admin_ids = BOUNTY_SERVICE.list_admins(room_id)
if not BOUNTY_SERVICE.is_admin(room_id, user_id):
await update.message.reply_text("⛔ Only admins can perform this action.")
if not admin_ids:
await update.message.reply_text("No admins configured for this room.")
return
target_user_id = await _find_user_id_by_username(ctx, username)
lines = [f"Room Admins:"]
for admin_id in admin_ids:
lines.append(f"- @{admin_id}")
if target_user_id is None:
await update.message.reply_text(f"⛔ User @{username} not found.")
return
try:
if subcommand == "add":
BOUNTY_SERVICE.add_admin(room_id, target_user_id, user_id)
await update.message.reply_text(f"✅ @{username} is now an admin.")
elif subcommand == "remove":
BOUNTY_SERVICE.remove_admin(room_id, target_user_id, user_id)
await update.message.reply_text(f"✅ @{username} is no longer an admin.")
except PermissionError as e:
await update.message.reply_text(f"{e}")
async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
user_id = get_user_id(update)
room_id = get_room_id(update)
is_admin = BOUNTY_SERVICE.is_admin(room_id, user_id)
if is_admin:
lines = [
"👻 JIGAIDO Commands (Admin):\n",
"📋 Bounty Management:",
"/bounty — list bounties",
"/add — add bounty",
"/edit — edit bounty",
"/delete — delete bounty",
"/recover — recover deleted bounties",
"",
"🔗 Tracking:",
"/track — track bounty",
"/untrack — stop tracking",
"/my — your tracked bounties",
"/show — show bounty details",
"",
"⚙️ Room Management:",
"/admin — manage admins",
"/timezone — get/set timezone",
"",
"/start — re-initialize",
"/help — show this message",
]
else:
lines = [
"👻 JIGAIDO Commands:\n",
"/bounty — list bounties",
"/track — track bounty",
"/untrack — stop tracking",
"/my — your tracked bounties",
"/show — show bounty details",
"/start — re-initialize",
"/help — show this message",
]
await update.message.reply_text("\n".join(lines), disable_web_page_preview=True)
await update.message.reply_text("\n".join(lines))

View File

@@ -1,28 +0,0 @@
#!/usr/bin/env python3
import asyncio
import os
import sys
# Run from the telegram-bot directory so local imports work
os.chdir("/home/shoko/repositories/jigaido/apps/telegram-bot")
sys.path.insert(0, "/home/shoko/repositories/jigaido")
# Import main from the local bot module
import bot as bot_module
if __name__ == "__main__":
if not bot_module.BOT_TOKEN:
bot_module.log.error("JIGAIDO_BOT_TOKEN environment variable not set.")
sys.exit(1)
app = bot_module.build_app()
app.post_init = bot_module.post_init
bot_module.log.info("JIGAIDO starting...")
# PTB v20+ app.run_polling() is async - use asyncio.get_event_loop() + run_until_complete
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(app.run_polling(drop_pending_updates=True))
finally:
loop.close()

View File

@@ -13,21 +13,7 @@ class Config:
def __init__(self):
self.data_dir: Path = self._resolve_data_dir()
self.bot_token: Optional[str] = self._resolve_bot_token()
def _resolve_bot_token(self) -> Optional[str]:
env_token = os.environ.get("JIGAIDO_BOT_TOKEN")
if env_token:
return env_token
config_file = Path("~/.jigaido/config.json").expanduser()
if config_file.exists():
with open(config_file) as f:
config_data = json.load(f)
if "JIGAIDO_BOT_TOKEN" in config_data:
return config_data["JIGAIDO_BOT_TOKEN"]
return None
self.bot_token: Optional[str] = os.environ.get("JIGAIDO_BOT_TOKEN")
def _resolve_data_dir(self) -> Path:
env_dir = os.environ.get("JIGAIDO_DATA_DIR")
@@ -49,6 +35,3 @@ class Config:
config = Config()
config = Config()

View File

@@ -35,23 +35,18 @@ class BountyService:
def add_admin(
self, room_id: int, admin_user_id: int, requesting_user_id: int
) -> None:
"""Add an admin to a room. Requires admin permission, or self-promotion if first admin."""
room_data = self._storage.load(room_id)
has_no_admins = room_data is None or not room_data.admin_user_ids
is_self_promotion = requesting_user_id == admin_user_id
"""Add an admin to a room. Requires admin permission."""
if not self.is_admin(room_id, requesting_user_id):
if not (has_no_admins and is_self_promotion):
raise PermissionError("Only admins can add admins.")
if room_data is None or room_data.admin_user_ids is None:
room_data = self._storage.load(room_id)
if room_data is None:
room_data = RoomData(
room_id=room_id, bounties=[], next_id=1, admin_user_ids=[]
)
if admin_user_id not in (room_data.admin_user_ids or []):
room_data.admin_user_ids.append(admin_user_id)
(room_data.admin_user_ids or []).append(admin_user_id)
self._storage.save(room_data)
def remove_admin(
@@ -124,7 +119,6 @@ class BountyService:
text: Optional[str] = None,
link: Optional[str] = None,
due_date_ts: Optional[int] = None,
created_by_username: Optional[str] = None,
) -> Bounty:
"""Add a new bounty to the room. Requires admin permission."""
if not self.is_admin(room_id, user_id):
@@ -142,7 +136,6 @@ class BountyService:
bounty = Bounty(
id=room_data.next_id,
created_by_user_id=user_id,
created_by_username=created_by_username,
text=text,
link=link,
due_date_ts=due_date_ts,
@@ -160,44 +153,6 @@ class BountyService:
all_bounties = self._storage.list_all_bounties(room_id, include_deleted=True)
return [b for b in all_bounties if b.deleted_at is not None]
def get_deleted_bounty(self, room_id: int, bounty_id: int) -> Bounty | None:
"""Get a specific soft-deleted bounty by ID."""
all_bounties = self._storage.list_all_bounties(room_id, include_deleted=True)
for b in all_bounties:
if b.id == bounty_id and b.deleted_at is not None:
return b
return None
def recover_bounty(self, room_id: int, bounty_id: int, user_id: int) -> str:
"""Recover a soft-deleted bounty. Admin only.
Returns: 'recovered', 'not_found', 'not_deleted', 'permission_denied'
"""
if not self.is_admin(room_id, user_id):
return "permission_denied"
bounty = self.get_deleted_bounty(room_id, bounty_id)
if not bounty:
return "not_found"
if bounty.deleted_at is None:
return "not_deleted"
bounty.deleted_at = None
self._storage.update_bounty(room_id, bounty)
return "recovered"
def recover_bounties(
self, room_id: int, bounty_ids: list[int], user_id: int
) -> dict[int, str]:
"""Recover multiple soft-deleted bounties. Admin only.
Returns dict of bounty_id -> result ('recovered', 'not_found', 'not_deleted', 'permission_denied')
"""
results = {}
for bounty_id in bounty_ids:
results[bounty_id] = self.recover_bounty(room_id, bounty_id, user_id)
return results
def get_bounty(self, room_id: int, bounty_id: int) -> Bounty | None:
"""Get a specific bounty by ID. Excludes soft-deleted bounties."""
bounty = self._storage.get_bounty(room_id, bounty_id)
@@ -255,28 +210,6 @@ class BountyService:
self._storage.update_bounty(room_id, bounty)
return True
def delete_bounties(
self, room_id: int, bounty_ids: list[int], user_id: int
) -> dict[int, str]:
"""Soft delete multiple bounties. Returns dict of bounty_id -> result.
Results can be: 'deleted', 'not_found', 'permission_denied'
"""
results = {}
for bounty_id in bounty_ids:
bounty = self._storage.get_bounty(room_id, bounty_id)
if not bounty:
results[bounty_id] = "not_found"
continue
if not self.is_admin(room_id, user_id):
results[bounty_id] = "permission_denied"
continue
bounty.deleted_at = int(time.time())
self._storage.update_bounty(room_id, bounty)
results[bounty_id] = "deleted"
return results
class TrackingService:
"""Service for tracking bounty operations."""

View File

@@ -47,28 +47,11 @@ class TestConfigDataDir:
assert cfg.bot_token == "test_token_123"
def test_bot_token_none_when_not_set(self):
"""Test that bot_token is None when JIGAIDO_BOT_TOKEN not set and no config file."""
"""Test that bot_token is None when JIGAIDO_BOT_TOKEN not set."""
with patch.dict(os.environ, {}, clear=True):
with patch("pathlib.Path.exists", return_value=False):
cfg = Config()
assert cfg.bot_token is None
def test_bot_token_from_config_file(self):
"""Test that bot_token is read from config file when env var not set."""
config_dir = Path.home() / ".jigaido"
config_file = config_dir / "config.json"
with patch.dict(os.environ, {}, clear=True):
with patch("pathlib.Path.expanduser", return_value=config_file):
with patch("pathlib.Path.exists", return_value=True):
with patch("builtins.open", create=True) as mock_open:
mock_open.return_value.__enter__ = lambda s: s
mock_open.return_value.__exit__ = lambda *a: None
mock_open.return_value.read = lambda: (
'{"JIGAIDO_BOT_TOKEN": "config_token"}'
)
cfg = Config()
assert cfg.bot_token == "config_token"
class TestConfigEnsureDataDir:
def test_ensure_data_dir_creates_directory(self, tmp_path):

View File

@@ -210,46 +210,6 @@ class TestBountyService:
result = self.service.delete_bounty(-1001, 999, self.admin_user_id)
assert result is False
def test_delete_bounties_multi_id_success(self):
"""Test delete_bounties returns individual results for multiple bounties."""
bounty1 = self.service.add_bounty(
room_id=-1001, user_id=self.admin_user_id, text="To delete 1"
)
bounty2 = self.service.add_bounty(
room_id=-1001, user_id=self.admin_user_id, text="To delete 2"
)
results = self.service.delete_bounties(
-1001, [bounty1.id, bounty2.id], self.admin_user_id
)
assert results == {bounty1.id: "deleted", bounty2.id: "deleted"}
# Verify both are soft deleted
assert self.service.get_bounty(-1001, bounty1.id) is None
assert self.service.get_bounty(-1001, bounty2.id) is None
def test_delete_bounties_mixed_results(self):
"""Test delete_bounties returns not_found for non-existent bounties."""
bounty = self.service.add_bounty(
room_id=-1001, user_id=self.admin_user_id, text="To delete"
)
results = self.service.delete_bounties(
-1001, [bounty.id, 999, 888], self.admin_user_id
)
assert results == {bounty.id: "deleted", 999: "not_found", 888: "not_found"}
def test_delete_bounties_permission_denied(self):
"""Test delete_bounties returns permission_denied for non-admin users."""
bounty = self.service.add_bounty(
room_id=-1001, user_id=self.admin_user_id, text="To delete"
)
results = self.service.delete_bounties(
-1001,
[bounty.id],
999, # non-admin user
)
assert results == {bounty.id: "permission_denied"}
# Verify bounty was NOT deleted
assert self.service.get_bounty(-1001, bounty.id) is not None
class TestTrackingService:
"""Unit tests for TrackingService."""