Refactor to apps/ structure
JIGAIDO is now a platform with apps/ as the container. All telegram-bot files moved to apps/telegram-bot/: - bot.py, commands.py, cron.py, db.py, schema.sql - requirements.txt, .env.example, README.md - Root now holds SPEC.md, README.md, CONTRIBUTING.md only. Structure: jigaido/ ├── apps/ │ └── telegram-bot/ └── SPEC.md, README.md, CONTRIBUTING.md
This commit is contained in:
shokollm
436
apps/telegram-bot/commands.py
Normal file
436
apps/telegram-bot/commands.py
Normal file
@@ -0,0 +1,436 @@
|
||||
"""Telegram command handlers for JIGAIDO."""
|
||||
|
||||
import re
|
||||
import time
|
||||
from functools import wraps
|
||||
|
||||
import dateparser
|
||||
from telegram import Update
|
||||
from telegram.ext import ContextTypes
|
||||
|
||||
import db
|
||||
|
||||
TELEGRAM_BOT_USERNAME = "your_bot_username" # Set via set_bot_commands / config
|
||||
|
||||
REMINDER_WINDOW_DAYS = 7
|
||||
|
||||
|
||||
# ── Helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def extract_args(text: str) -> list[str]:
|
||||
"""Split command text into tokens, preserving URLs as single tokens."""
|
||||
if not text:
|
||||
return []
|
||||
tokens = text.strip().split()
|
||||
# First token is the command itself (e.g. /add), rest is args
|
||||
return tokens[1:] if len(tokens) > 1 else []
|
||||
|
||||
|
||||
def parse_args(args: list[str]) -> tuple[str | None, str | None, int | None]:
|
||||
"""Parse /add args into (text, link, due_date_ts)."""
|
||||
text = None
|
||||
link = None
|
||||
due_date_ts = None
|
||||
|
||||
remaining = []
|
||||
for arg in args:
|
||||
if not link and (arg.startswith("http://") or arg.startswith("https://")):
|
||||
link = arg
|
||||
elif due_date_ts is None:
|
||||
parsed = dateparser.parse(arg)
|
||||
if parsed:
|
||||
due_date_ts = int(parsed.timestamp())
|
||||
else:
|
||||
remaining.append(arg)
|
||||
else:
|
||||
remaining.append(arg)
|
||||
|
||||
text = " ".join(remaining) if remaining else None
|
||||
return text, link, due_date_ts
|
||||
|
||||
|
||||
def format_bounty(b: dict, show_id: bool = True) -> str:
|
||||
parts = []
|
||||
if show_id:
|
||||
parts.append(f"[#{b['id']}]")
|
||||
if b["text"]:
|
||||
parts.append(b["text"])
|
||||
if b["link"]:
|
||||
parts.append(f"🔗 {b['link']}")
|
||||
if b["due_date_ts"]:
|
||||
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 days_left == 0:
|
||||
parts.append(f"⏰ {due_str} (TODAY)")
|
||||
else:
|
||||
parts.append(f"⏰ {due_str} ({days_left}d)")
|
||||
parts.append(f"by @{b['informed_by_username'] or 'unknown'}")
|
||||
return " | ".join(parts)
|
||||
|
||||
|
||||
def is_group(update: Update) -> bool:
|
||||
return update.effective_chat.type != "private"
|
||||
|
||||
|
||||
def ensure_user(update: Update) -> int:
|
||||
user = update.effective_user
|
||||
username = user.username
|
||||
return db.upsert_user(user.id, username)
|
||||
|
||||
|
||||
def ensure_group(update: Update) -> tuple[int, int]:
|
||||
"""Ensure group and admin-creator exist. Returns (group_id, creator_user_id)."""
|
||||
user_id = ensure_user(update)
|
||||
creator_user_id = db.upsert_user(update.effective_user.id, update.effective_user.username)
|
||||
group_id = db.upsert_group(update.effective_chat.id, creator_user_id)
|
||||
# Ensure creator is also an admin
|
||||
db.add_group_admin(group_id, creator_user_id)
|
||||
return group_id, creator_user_id
|
||||
|
||||
|
||||
def admin_only(func):
|
||||
@wraps(func)
|
||||
async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if not group:
|
||||
await update.message.reply_text("Group not found. Try /start in the group first.")
|
||||
return
|
||||
user_id = ensure_user(update)
|
||||
if not db.is_group_admin(group["id"], user_id):
|
||||
await update.message.reply_text("⛔ Admin only.")
|
||||
return
|
||||
return await func(update, ctx)
|
||||
return wrapper
|
||||
|
||||
|
||||
def creator_only(func):
|
||||
@wraps(func)
|
||||
async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if not group:
|
||||
await update.message.reply_text("Group not found.")
|
||||
return
|
||||
user_id = ensure_user(update)
|
||||
if not db.is_group_creator(group["id"], user_id):
|
||||
await update.message.reply_text("⛔ Group creator only.")
|
||||
return
|
||||
return await func(update, ctx)
|
||||
return wrapper
|
||||
|
||||
|
||||
# ── Commands ─────────────────────────────────────────────────────────────────
|
||||
|
||||
async def cmd_bounty(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""List all bounties. Group: group bounties. DM: user's personal bounties."""
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if not group:
|
||||
await update.message.reply_text("Group not initialized. Try /start.")
|
||||
return
|
||||
bounties = db.get_group_bounties(group["id"])
|
||||
else:
|
||||
user_id = ensure_user(update)
|
||||
bounties = db.get_user_personal_bounties(user_id)
|
||||
|
||||
if not bounties:
|
||||
await update.message.reply_text("No bounties yet.")
|
||||
return
|
||||
|
||||
lines = [format_bounty(dict(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:
|
||||
"""List bounties tracked by the user. Group: tracked group bounties. DM: tracked personal bounties."""
|
||||
user_id = ensure_user(update)
|
||||
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if not group:
|
||||
await update.message.reply_text("Group not found.")
|
||||
return
|
||||
bounties = db.get_user_tracked_bounties_in_group(user_id, group["id"])
|
||||
else:
|
||||
bounties = db.get_user_tracked_bounties_personal(user_id)
|
||||
|
||||
if not bounties:
|
||||
await update.message.reply_text("You are not tracking any bounties.")
|
||||
return
|
||||
|
||||
lines = [format_bounty(dict(b), show_id=True) for b in bounties]
|
||||
await update.message.reply_text("\n".join(lines), disable_web_page_preview=True)
|
||||
|
||||
|
||||
@admin_only
|
||||
async def cmd_add(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Add a bounty. Args: [text] [link] [due_date]."""
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /add <text> [link] [due_date]\nExample: /add Fix the bug https://github.com/foo/bar tomorrow")
|
||||
return
|
||||
|
||||
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
|
||||
|
||||
if is_group(update):
|
||||
group_id, creator_user_id = ensure_group(update)
|
||||
created_by = creator_user_id
|
||||
else:
|
||||
group_id = None
|
||||
created_by = ensure_user(update)
|
||||
|
||||
informed_by = update.effective_user.username or str(update.effective_user.id)
|
||||
|
||||
try:
|
||||
bounty_id = db.add_bounty(group_id, created_by, informed_by, text, link, due_date_ts)
|
||||
except ValueError as e:
|
||||
await update.message.reply_text(f"⛔ {e}")
|
||||
return
|
||||
|
||||
due_str = ""
|
||||
if due_date_ts:
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
@admin_only
|
||||
async def cmd_update(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Update a bounty. Args: <bounty_id> [text] [link] [due_date]."""
|
||||
args = extract_args(update.message.text)
|
||||
if len(args) < 1:
|
||||
await update.message.reply_text("Usage: /update <bounty_id> [text] [link] [due_date]")
|
||||
return
|
||||
|
||||
try:
|
||||
bounty_id = int(args[0])
|
||||
except ValueError:
|
||||
await update.message.reply_text("Invalid bounty ID.")
|
||||
return
|
||||
|
||||
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
|
||||
|
||||
# Verify bounty belongs to this group / user
|
||||
bounty = db.get_bounty(bounty_id)
|
||||
if not bounty:
|
||||
await update.message.reply_text("Bounty not found.")
|
||||
return
|
||||
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if bounty["group_id"] != group["id"]:
|
||||
await update.message.reply_text("Bounty not found in this group.")
|
||||
return
|
||||
else:
|
||||
if bounty["group_id"] is not None:
|
||||
await update.message.reply_text("This bounty belongs to a group, not your personal list.")
|
||||
return
|
||||
if bounty["created_by_user_id"] != ensure_user(update):
|
||||
await update.message.reply_text("You can only update your own bounties.")
|
||||
return
|
||||
|
||||
try:
|
||||
db.update_bounty(bounty_id, text, link, due_date_ts)
|
||||
except ValueError as e:
|
||||
await update.message.reply_text(f"⛔ {e}")
|
||||
return
|
||||
|
||||
await update.message.reply_text(f"✅ Bounty #{bounty_id} updated.")
|
||||
|
||||
|
||||
@admin_only
|
||||
async def cmd_delete(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Delete a bounty. Args: <bounty_id>."""
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /delete <bounty_id>")
|
||||
return
|
||||
|
||||
try:
|
||||
bounty_id = int(args[0])
|
||||
except ValueError:
|
||||
await update.message.reply_text("Invalid bounty ID.")
|
||||
return
|
||||
|
||||
bounty = db.get_bounty(bounty_id)
|
||||
if not bounty:
|
||||
await update.message.reply_text("Bounty not found.")
|
||||
return
|
||||
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if bounty["group_id"] != group["id"]:
|
||||
await update.message.reply_text("Bounty not found in this group.")
|
||||
return
|
||||
else:
|
||||
if bounty["group_id"] is not None:
|
||||
await update.message.reply_text("This bounty belongs to a group.")
|
||||
return
|
||||
|
||||
db.delete_bounty(bounty_id)
|
||||
await update.message.reply_text(f"✅ Bounty #{bounty_id} deleted.")
|
||||
|
||||
|
||||
async def cmd_track(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Track a bounty. Args: <bounty_id>."""
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /track <bounty_id>")
|
||||
return
|
||||
|
||||
try:
|
||||
bounty_id = int(args[0])
|
||||
except ValueError:
|
||||
await update.message.reply_text("Invalid bounty ID.")
|
||||
return
|
||||
|
||||
bounty = db.get_bounty(bounty_id)
|
||||
if not bounty:
|
||||
await update.message.reply_text("Bounty not found.")
|
||||
return
|
||||
|
||||
if is_group(update):
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
if bounty["group_id"] != group["id"]:
|
||||
await update.message.reply_text("Bounty not found in this group.")
|
||||
return
|
||||
else:
|
||||
if bounty["group_id"] is not None:
|
||||
await update.message.reply_text("Use /track from the group where the bounty belongs.")
|
||||
return
|
||||
|
||||
user_id = ensure_user(update)
|
||||
added = db.track_bounty(user_id, bounty_id)
|
||||
if added:
|
||||
await update.message.reply_text(f"✅ Now tracking bounty #{bounty_id}.")
|
||||
else:
|
||||
await update.message.reply_text(f"Already tracking bounty #{bounty_id}.")
|
||||
|
||||
|
||||
async def cmd_untrack(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Untrack a bounty. Args: <bounty_id>."""
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /untrack <bounty_id>")
|
||||
return
|
||||
|
||||
try:
|
||||
bounty_id = int(args[0])
|
||||
except ValueError:
|
||||
await update.message.reply_text("Invalid bounty ID.")
|
||||
return
|
||||
|
||||
user_id = ensure_user(update)
|
||||
removed = db.untrack_bounty(user_id, bounty_id)
|
||||
if removed:
|
||||
await update.message.reply_text(f"✅ Untracked bounty #{bounty_id}.")
|
||||
else:
|
||||
await update.message.reply_text(f"Not tracking bounty #{bounty_id}.")
|
||||
|
||||
|
||||
@creator_only
|
||||
async def cmd_admin_add(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Promote a user to admin. Args: <username>."""
|
||||
if not is_group(update):
|
||||
await update.message.reply_text("This command only works in groups.")
|
||||
return
|
||||
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /admin_add <username>")
|
||||
return
|
||||
|
||||
username = args[0].lstrip("@")
|
||||
user = db.get_user_by_username(username)
|
||||
if not user:
|
||||
await update.message.reply_text(f"User @{username} not found. They must interact with the bot first.")
|
||||
return
|
||||
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
added = db.add_group_admin(group["id"], user["id"])
|
||||
if added:
|
||||
await update.message.reply_text(f"✅ @{username} is now a group admin.")
|
||||
else:
|
||||
await update.message.reply_text(f"@{username} is already an admin.")
|
||||
|
||||
|
||||
@creator_only
|
||||
async def cmd_admin_remove(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Demote an admin. Args: <username>."""
|
||||
if not is_group(update):
|
||||
await update.message.reply_text("This command only works in groups.")
|
||||
return
|
||||
|
||||
args = extract_args(update.message.text)
|
||||
if not args:
|
||||
await update.message.reply_text("Usage: /admin_remove <username>")
|
||||
return
|
||||
|
||||
username = args[0].lstrip("@")
|
||||
user = db.get_user_by_username(username)
|
||||
if not user:
|
||||
await update.message.reply_text(f"User @{username} not found.")
|
||||
return
|
||||
|
||||
group = db.get_group(update.effective_chat.id)
|
||||
# Prevent removing the creator
|
||||
if db.is_group_creator(group["id"], user["id"]):
|
||||
await update.message.reply_text("Cannot remove the group creator.")
|
||||
return
|
||||
|
||||
removed = db.remove_group_admin(group["id"], user["id"])
|
||||
if removed:
|
||||
await update.message.reply_text(f"✅ @{username} is no longer a group admin.")
|
||||
else:
|
||||
await update.message.reply_text(f"@{username} was not an admin.")
|
||||
|
||||
|
||||
async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
if is_group(update):
|
||||
ensure_group(update)
|
||||
await update.message.reply_text(
|
||||
"👻 JIGAIDO is watching.\n\n"
|
||||
"This group's bounty list is now active.\n"
|
||||
"Only admins can add/update/delete bounties.\n"
|
||||
"Anyone can /track and /untrack.\n\n"
|
||||
"Try /bounty to see all bounties, /add to create one."
|
||||
)
|
||||
else:
|
||||
ensure_user(update)
|
||||
await update.message.reply_text(
|
||||
"👻 JIGAIDO activated.\n\n"
|
||||
"Personal bounty list ready.\n"
|
||||
"/bounty — list your bounties\n"
|
||||
"/add — create a bounty\n"
|
||||
"/my — your tracked bounties"
|
||||
)
|
||||
|
||||
|
||||
async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
await update.message.reply_text(
|
||||
"👻 JIGAIDO Commands:\n\n"
|
||||
"/bounty — list all bounties\n"
|
||||
"/my — bounties you're tracking\n"
|
||||
"/add <text> [link] [due] — add bounty (admin/DM only)\n"
|
||||
"/update <id> [text] [link] [due] — update bounty (admin/DM only)\n"
|
||||
"/delete <id> — delete bounty (admin/DM only)\n"
|
||||
"/track <id> — track a bounty\n"
|
||||
"/untrack <id> — stop tracking\n"
|
||||
"/admin_add <user> — promote to admin (creator only, group)\n"
|
||||
"/admin_remove <user> — demote admin (creator only, group)\n"
|
||||
"/start — re-initialize group\n"
|
||||
"/help — this message",
|
||||
disable_web_page_preview=True,
|
||||
)
|
||||
Reference in New Issue
Block a user