shoko/jigaido
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat: Replace SQLite with per-user JSON storage (fixes #2)' (#3) from fix/issue-2-json-storage into main

Browse Source
This commit was merged in pull request #3.
This commit is contained in:
shoko
2026-04-02 17:44:08 +02:00
parent 332d7fc60a 7c2bd09ada
commit 98a8c4d173
9 changed files with 450 additions and 1109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
SPEC.md
View File

@@ -6,23 +6,21 @@
## Overview
JIGAIDO is a Telegram bot that lets groups and individuals track bounties — tasks, obligations, and deadlines — with optional due dates and personal tracking/reminders.
JIGAIDO is a Telegram bot that lets groups and individuals track bounties — tasks, obligations, and deadlines — with optional due dates and personal tracking.
- **Group mode**: Each Telegram group has its own bounty list. Only group admins can add/update/delete bounties. Any member can track/untrack.
- **DM mode**: Personal bounty list. No admin restrictions — anyone can manage their own bounties.
- **Tracking**: Users can add any bounty (group or DM) to their personal tracking list.
- **Reminders**: Daily cron checks for due dates within 7 days and DMs the user.
- **Due dates**: Free-form text (`"april 15"`, `"in 3 days"`, `"tomorrow"`) parsed at add time, stored as Unix timestamp. If unparseable, stored as `NULL` — no reminder.
- **Links**: Optional. If provided, deduplicated per group (no two bounties in the same group can share the same link). Multiple links in one bounty: first link only, user can update later.
- **Informed by**: Every bounty stores the Telegram username of who posted/added it (not who created the record — the person whose message triggered the add).
- **Group mode**: Each Telegram group has its own bounty list. Anyone can add bounties. Only creator can edit/delete.
- **DM mode**: Personal bounty list. Anyone can manage their own bounties.
- **Tracking**: Users can track any bounty (group or personal) to their tracking list.
- **Due dates**: Free-form text (`"april 15"`, `"in 3 days"`, `"tomorrow"`) parsed at add time, stored as Unix timestamp. If unparseable, stored as `NULL`.
- **Links**: Optional. If provided, stored with the bounty.
- **Informed by**: Every bounty stores the user ID of who posted/added it.
---
## Architecture
### Stack
- **Bot**: `python-telegram-bot` (pure Python, no C extensions)
- **Database**: SQLite (zero-install, single file)
- **Storage**: Per-group JSON files (zero-setup, no DB server)
- **Date parsing**: `dateparser`
- **Runtime**: Python 3.10+
- **Deployment**: Any $5 VPS with Python 3.10+
@@ -30,82 +28,71 @@ JIGAIDO is a Telegram bot that lets groups and individuals track bounties — ta
### Directory Structure
```
jigaido/
├── apps/
│ └── telegram-bot/ # Telegram bot app
│ ├── bot.py # Entrypoint
│ ├── commands.py # Command handlers
│ ├── cron.py # Daily reminder job
│ ├── db.py # SQLite wrapper
│ ├── schema.sql # Database schema
│ ├── requirements.txt
│ └── .env.example
├── SPEC.md # This document
├── README.md
└── CONTRIBUTING.md
~/.jigaido/ # Data root (~/.jigaido/)
├── {group_id}/
│ ├── group.json # Group bounties
│ └── {user_id}.json # User tracking within this group
└── {user_id}/
└── user.json # User's personal bounties (DM mode)
```
---
**Note:** Data directory is at `~/.jigaido/` (home directory), NOT inside the repository or app directory.
## Database Schema
### Storage Design
```sql
CREATE TABLE groups (
id INTEGER PRIMARY KEY AUTOINCREMENT,
telegram_chat_id INTEGER UNIQUE NOT NULL,
creator_user_id INTEGER NOT NULL,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
**File: `data/{group_id}/group.json`**
CREATE TABLE group_admins (
group_id INTEGER REFERENCES groups(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL,
PRIMARY KEY (group_id, user_id)
);
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
telegram_user_id INTEGER UNIQUE NOT NULL,
username TEXT,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
CREATE TABLE bounties (
id INTEGER PRIMARY KEY AUTOINCREMENT,
group_id INTEGER REFERENCES groups(id) ON DELETE CASCADE,
created_by_user_id INTEGER REFERENCES users(id),
informed_by_username TEXT NOT NULL,
text TEXT,
link TEXT,
due_date_ts INTEGER,
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(group_id, link)
);
CREATE TABLE user_bounty_tracking (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
bounty_id INTEGER NOT NULL REFERENCES bounties(id) ON DELETE CASCADE,
added_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(user_id, bounty_id)
);
CREATE TABLE reminder_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
bounty_id INTEGER NOT NULL REFERENCES bounties(id) ON DELETE CASCADE,
reminded_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(user_id, bounty_id)
);
```json
{
"bounties": [
{
"id": 1,
"created_by_user_id": 123456,
"text": "Fix login bug",
"link": "https://github.com/example/repo/issues/1",
"due_date_ts": 1735689600,
"created_at": 1735603200
}
],
"next_id": 2
}
```
### Notes
- `group_id = NULL` means a personal/DM bounty.
- `UNIQUE(group_id, link)` — only enforced when `link IS NOT NULL` (SQLite treats NULL as distinct).
- `reminder_log` dedup ensures a user only gets one reminder per bounty.
**File: `data/{group_id}/{user_id}.json`**
```json
{
"tracked": [
{"bounty_id": 1, "created_at": 1735600000}
]
}
```
**File: `data/{user_id}/user.json`**
```json
{
"bounties": [
{
"id": 1,
"text": "Personal task",
"link": null,
"due_date_ts": 1735689600,
"created_at": 1735603200
}
],
"next_id": 2
}
```
**Key design decisions:**
1. **Group-isolated storage** — Each group has its own directory. No cross-group access.
2. **Bounty IDs are sequential per group.json** — Not global. Each group's file has its own ID counter.
3. **Atomic writes** — Uses `tempfile` + `rename` for safe writes.
4. **No reminders in v1** — Dropped for simplicity.
---
## Commands
### In Group
@@ -114,24 +101,22 @@ CREATE TABLE reminder_log (
|---|---|---|
| `/bounty` | anyone | List all bounties in this group |
| `/my` | anyone | List bounties tracked by you in this group |
| `/add <text> [link] [due date]` | admin only | Add a new bounty to the group |
| `/update <bounty_id> [text] [link] [due_date]` | admin only | Update an existing bounty |
| `/delete <bounty_id>` | admin only | Delete a bounty |
| `/track <bounty_id>` | anyone | Add a group bounty to your tracking |
| `/untrack <bounty_id>` | anyone | Remove a bounty from your tracking |
| `/admin_add <user>` | creator only | Promote a user to admin |
| `/admin_remove <user>` | creator only | Demote an admin |
| `/add <text> [link] [due date]` | anyone | Add a new bounty to the group |
| `/update <bounty_id> [text] [link] [due_date]` | creator only | Update an existing bounty |
| `/delete <bounty_id>` | creator only | Delete a bounty |
| `/track <bounty_id>` | anyone | Track a group bounty |
| `/untrack <bounty_id>` | anyone | Stop tracking a bounty |
### In DM (1:1 with bot)
| Command | Description |
|---|---|
| `/bounty` | List all your personal bounties |
| `/my` | List all your tracked personal bounties |
| `/my` | List all your personal bounties |
| `/add <text> [link] [due date]` | Add a personal bounty |
| `/update <bounty_id> [text] [link] [due_date]` | Update a personal bounty |
| `/delete <bounty_id>` | Delete a personal bounty (owner only) |
| `/track <bounty_id>` | Add a personal bounty to your tracking |
| `/delete <bounty_id>` | Delete a personal bounty |
| `/track <bounty_id>` | Track a personal bounty |
### Add/Update Syntax
@@ -143,19 +128,6 @@ CREATE TABLE reminder_log (
- `link` is optional
- `due_date` is optional, free-form
- If link already exists in group → rejected with error
### Tracking
- `/track <bounty_id>` — works in both group and DM. In group: tracks a group bounty. In DM: tracks a personal bounty.
- Users can track any bounty regardless of who created it.
- A bounty can be tracked by multiple users.
---
## Informed By
When a user triggers `/add`, the bot captures `message.from_user.username` and stores it in `informed_by_username`. This is displayed on bounty listings so the group/DM knows who posted or requested the bounty.
---
@@ -168,40 +140,26 @@ Uses `dateparser` library. Examples:
- `"2026-04-15"`
- `"next friday"`
If parsing fails → `due_date_ts = NULL`. No error is shown to user, reminder just won't fire.
If parsing fails → `due_date_ts = NULL`. No error is shown to user.
Stored as Unix timestamp. User-facing display can be localized/converted to any timezone at render time.
---
## Reminders (Cron)
Runs daily (e.g., 09:00 local). For each user:
1. Find tracked bounties where `due_date_ts - now() < 7 days`
2. Exclude any already in `reminder_log` for that user
3. Send DM: `"Bounty '{title}' is due in {N} days."`
4. Insert into `reminder_log`
Does not re-remind. If a bounty is 2 days away today, you get one message. Tomorrow you don't get another.
---
## Admin Management
- **Creator**: The user who first added the bot to the group. Stored as `creator_user_id` in `groups`. Only the creator can run `/admin_add` and `/admin_remove`.
- **Admins**: Added via `/admin_add <username>`. Can add/update/delete any bounty in the group. Regular members can only track/untrack.
- First admin assignment is automatic when the bot detects a new group.
---
## Error Handling
- Unknown command → help text with available commands
- `/add` with duplicate link in same group → rejection message
- `/update`/`/delete` by non-admin → "Admin only" message
- `/admin_add`/`/admin_remove` by non-creator → "Creator only" message
- `/track` already tracked → "Already tracking" (idempotent, no error)
- `/untrack` not tracked → "Not tracking" (idempotent, no error)
- `/update`/`/delete` by non-creator → "⛔ Only the creator can edit/delete this bounty."
- `/track` already tracked → "Already tracking" (idempotent)
- `/untrack` not tracked → "Not tracking" (idempotent)
- Bounty not found → "Bounty not found"
- User not found → "User not found"
---
## When to Revert to SQLite
- Multiple concurrent users with write conflicts
- Complex queries across users
- Reminder system with proper dedup
- Scale > 1,000 users
- Need ACID guarantees on concurrent writes

16
apps/telegram-bot/bot.py
View File

@@ -12,7 +12,6 @@ from telegram.ext import (
filters,
)
import db
import commands
logging.basicConfig(
@@ -21,14 +20,12 @@ logging.basicConfig(
)
log = logging.getLogger(__name__)
# Token from environment or config
BOT_TOKEN = os.environ.get("JIGAIDO_BOT_TOKEN", "")
def build_app() -> Application:
app = Application.builder().token(BOT_TOKEN).build()
# Core commands
app.add_handler(CommandHandler("start", commands.cmd_start))
app.add_handler(CommandHandler("help", commands.cmd_help))
app.add_handler(CommandHandler("bounty", commands.cmd_bounty))
@@ -38,25 +35,23 @@ def build_app() -> Application:
app.add_handler(CommandHandler("delete", commands.cmd_delete))
app.add_handler(CommandHandler("track", commands.cmd_track))
app.add_handler(CommandHandler("untrack", commands.cmd_untrack))
app.add_handler(CommandHandler("admin_add", commands.cmd_admin_add))
app.add_handler(CommandHandler("admin_remove", commands.cmd_admin_remove))
# Fallback: unknown commands
app.add_handler(MessageHandler(filters.COMMAND, commands.cmd_help))
return app
async def post_init(app: Application) -> None:
# Set bot commands in menu
await app.bot.set_my_commands([
await app.bot.set_my_commands(
[
("bounty", "List bounties"),
("my", "Your tracked bounties"),
("add", "Add a bounty"),
("track", "Track a bounty"),
("untrack", "Stop tracking"),
("help", "Show help"),
])
]
)
def main() -> None:
@@ -64,9 +59,6 @@ def main() -> None:
log.error("JIGAIDO_BOT_TOKEN environment variable not set.")
sys.exit(1)
db.init_db()
log.info("Database initialized.")
app = build_app()
app.post_init = post_init

329
apps/telegram-bot/commands.py
View File

@@ -1,33 +1,29 @@
"""Telegram command handlers for JIGAIDO."""
import json
import os
import re
import time
from functools import wraps
from typing import Optional
import dateparser
from telegram import Update
from telegram.ext import ContextTypes
import db
import storage
TELEGRAM_BOT_USERNAME = "your_bot_username" # Set via set_bot_commands / config
TELEGRAM_BOT_USERNAME = "your_bot_username"
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)."""
def parse_args(args: list[str]) -> tuple[Optional[str], Optional[str], Optional[int]]:
text = None
link = None
due_date_ts = None
@@ -53,11 +49,11 @@ def format_bounty(b: dict, show_id: bool = True) -> str:
parts = []
if show_id:
parts.append(f"[#{b['id']}]")
if b["text"]:
if b.get("text"):
parts.append(b["text"])
if b["link"]:
if b.get("link"):
parts.append(f"🔗 {b['link']}")
if b["due_date_ts"]:
if b.get("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:
@@ -66,7 +62,8 @@ def format_bounty(b: dict, show_id: bool = True) -> str:
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'}")
if b.get("created_by_user_id"):
parts.append(f"by {b['created_by_user_id']}")
return " | ".join(parts)
@@ -74,67 +71,21 @@ 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 get_group_id(update: Update) -> int:
return update.effective_chat.id
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 get_user_id(update: Update) -> int:
return update.effective_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"])
data = storage.load_group_bounties(get_group_id(update))
bounties = data.get("bounties", [])
else:
user_id = ensure_user(update)
bounties = db.get_user_personal_bounties(user_id)
data = storage.load_user_personal(get_user_id(update))
bounties = data.get("bounties", [])
if not bounties:
await update.message.reply_text("No bounties yet.")
@@ -145,32 +96,51 @@ async def cmd_bounty(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
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)
user_id = get_user_id(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"])
group_id = get_group_id(update)
tracking = storage.load_user_tracking(group_id, user_id)
tracked = tracking.get("tracked", [])
else:
bounties = db.get_user_tracked_bounties_personal(user_id)
data = storage.load_user_personal(user_id)
bounties = data.get("bounties", [])
lines = [format_bounty(dict(b), show_id=True) for b in bounties]
await update.message.reply_text(
"\n".join(lines) if lines else "No personal bounties.",
disable_web_page_preview=True,
)
return
if not bounties:
if not tracked:
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)
group_data = storage.load_group_bounties(group_id)
bounty_map = {b["id"]: b for b in group_data.get("bounties", [])}
bounty_lines = []
for t in tracked:
bounty = bounty_map.get(t["bounty_id"])
if bounty:
bounty_lines.append(format_bounty(bounty, show_id=True))
if not bounty_lines:
await update.message.reply_text("You are not tracking any bounties.")
return
await update.message.reply_text(
"\n".join(bounty_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")
await update.message.reply_text(
"Usage: /add <text> [link] [due_date]\n"
"Example: /add Fix the bug https://github.com/foo/bar tomorrow"
)
return
text, link, due_date_ts = parse_args(args)
@@ -178,37 +148,30 @@ async def cmd_add(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("A bounty needs at least text or a link.")
return
user_id = get_user_id(update)
if is_group(update):
group_id, creator_user_id = ensure_group(update)
created_by = creator_user_id
group_id = get_group_id(update)
bounty = storage.add_group_bounty(group_id, user_id, text, link, due_date_ts)
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
bounty = storage.add_personal_bounty(user_id, text, link, due_date_ts)
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}",
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]")
await update.message.reply_text(
"Usage: /update <bounty_id> [text] [link] [due_date]"
)
return
try:
@@ -222,37 +185,29 @@ async def cmd_update(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Nothing to update.")
return
# Verify bounty belongs to this group / user
bounty = db.get_bounty(bounty_id)
user_id = get_user_id(update)
if is_group(update):
group_id = get_group_id(update)
bounty = storage.get_group_bounty(group_id, 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.")
if bounty["created_by_user_id"] != user_id:
await update.message.reply_text("⛔ Only the creator can edit this bounty.")
return
storage.update_group_bounty(group_id, bounty_id, text, link, due_date_ts)
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}")
bounty = storage.get_personal_bounty(user_id, bounty_id)
if not bounty:
await update.message.reply_text("Bounty not found.")
return
storage.update_personal_bounty(user_id, bounty_id, text, link, due_date_ts)
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>")
@@ -264,27 +219,31 @@ async def cmd_delete(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Invalid bounty ID.")
return
bounty = db.get_bounty(bounty_id)
user_id = get_user_id(update)
if is_group(update):
group_id = get_group_id(update)
bounty = storage.get_group_bounty(group_id, 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.")
if bounty["created_by_user_id"] != user_id:
await update.message.reply_text(
"⛔ Only the creator can delete this bounty."
)
return
storage.delete_group_bounty(group_id, bounty_id)
else:
if bounty["group_id"] is not None:
await update.message.reply_text("This bounty belongs to a group.")
bounty = storage.get_personal_bounty(user_id, bounty_id)
if not bounty:
await update.message.reply_text("Bounty not found.")
return
storage.delete_personal_bounty(user_id, bounty_id)
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>")
@@ -296,31 +255,26 @@ async def cmd_track(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Invalid bounty ID.")
return
bounty = db.get_bounty(bounty_id)
user_id = get_user_id(update)
if is_group(update):
group_id = get_group_id(update)
bounty = storage.get_group_bounty(group_id, 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
if storage.track_bounty(group_id, user_id, bounty_id):
await update.message.reply_text(f"✅ Now tracking bounty #{bounty_id}.")
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"Already tracking bounty #{bounty_id}.")
else:
if storage.track_bounty(user_id, user_id, bounty_id):
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>")
@@ -332,83 +286,32 @@ async def cmd_untrack(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Invalid bounty ID.")
return
user_id = ensure_user(update)
removed = db.untrack_bounty(user_id, bounty_id)
if removed:
user_id = get_user_id(update)
if is_group(update):
group_id = get_group_id(update)
if storage.untrack_bounty(group_id, user_id, bounty_id):
await update.message.reply_text(f"✅ Untracked bounty #{bounty_id}.")
else:
await update.message.reply_text(f"Not tracking bounty #{bounty_id}.")
else:
if storage.untrack_bounty(user_id, user_id, bounty_id):
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."
"/bounty — list bounties\n"
"/add — create a bounty\n"
"/track — track a bounty\n"
"/my — your tracked bounties"
)
else:
ensure_user(update)
await update.message.reply_text(
"👻 JIGAIDO activated.\n\n"
"Personal bounty list ready.\n"
@@ -423,14 +326,12 @@ async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
"👻 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"
"/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\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"
"/start — re-initialize\n"
"/help — this message",
disable_web_page_preview=True,
)

77
apps/telegram-bot/cron.py
View File

@@ -1,77 +0,0 @@
"""Daily reminder cron job for JIGAIDO.
Run with: python -m cron
Or schedule via systemd timer / cron.
"""
import asyncio
import logging
import os
import sys
import time
# Add project root to path
sys.path.insert(0, os.path.dirname(__file__))
import db
logging.basicConfig(
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
level=logging.INFO,
)
log = logging.getLogger(__name__)
# Token from environment
BOT_TOKEN = os.environ.get("JIGAIDO_BOT_TOKEN", "")
REMINDER_WINDOW_DAYS = 7
async def send_reminder(user_telegram_id: int, bounty: dict, bot) -> None:
days_left = (bounty["due_date_ts"] - int(time.time())) // 86400
if days_left < 0:
urgency = "OVERDUE"
elif days_left == 0:
urgency = "TODAY"
else:
urgency = f"{days_left} days left"
due_str = time.strftime("%Y-%m-%d", time.localtime(bounty["due_date_ts"]))
text = f"⏰ Reminder: bounty #{bounty['id']}"
if bounty["text"]:
text += f" — {bounty['text']}"
text += f"\nDue: {due_str} ({urgency})"
try:
await bot.send_message(chat_id=user_telegram_id, text=text, disable_web_page_preview=True)
log.info(f"Reminder sent to {user_telegram_id} for bounty #{bounty['id']}")
except Exception as e:
log.error(f"Failed to send reminder to {user_telegram_id}: {e}")
async def run_reminders() -> None:
if not BOT_TOKEN:
log.error("JIGAIDO_BOT_TOKEN not set")
return
from telegram import Bot
bot = Bot(BOT_TOKEN)
user_ids = db.get_all_user_ids()
log.info(f"Running reminders for {len(user_ids)} users...")
for user_telegram_id in user_ids:
due_bounties = db.get_bounties_due_soon(user_telegram_id, REMINDER_WINDOW_DAYS)
for bounty in due_bounties:
await send_reminder(user_telegram_id, dict(bounty), bot)
db.log_reminder(user_telegram_id, bounty["id"])
log.info("Reminder run complete.")
def main() -> None:
asyncio.run(run_reminders())
if __name__ == "__main__":
main()

306
apps/telegram-bot/db.py
View File

@@ -1,306 +0,0 @@
"""SQLite database wrapper for JIGAIDO."""
import sqlite3
import time
from pathlib import Path
from typing import Optional
DB_PATH = Path(__file__).parent / "jigaido.db"
def get_conn() -> sqlite3.Connection:
# isolation_level=None enables autocommit mode.
# row_factory disables SQLite Python's implicit transaction management,
# so we need explicit autocommit to make writes work correctly.
conn = sqlite3.connect(DB_PATH, detect_types=sqlite3.PARSE_DECLTYPES)
conn.isolation_level = None
conn.execute("PRAGMA foreign_keys = ON")
conn.row_factory = sqlite3.Row
return conn
def _row_to_dict(row: sqlite3.Row) -> dict:
return dict(row)
def init_db() -> None:
schema = (Path(__file__).parent / "schema.sql").read_text()
with get_conn() as conn:
conn.executescript(schema)
# ── Users ──────────────────────────────────────────────────────────────────
def upsert_user(telegram_user_id: int, username: str | None) -> int:
with get_conn() as conn:
cur = conn.execute(
"""INSERT INTO users (telegram_user_id, username)
VALUES (?, ?)
ON CONFLICT (telegram_user_id) DO UPDATE SET username = excluded.username
RETURNING id""",
(telegram_user_id, username),
)
return cur.fetchone()["id"]
def get_user_by_telegram_id(telegram_user_id: int) -> Optional[dict]:
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM users WHERE telegram_user_id = ?",
(telegram_user_id,),
).fetchone()
return _row_to_dict(row) if row else None
# ── Groups ─────────────────────────────────────────────────────────────────
def upsert_group(telegram_chat_id: int, creator_user_id: int) -> int:
"""Insert group if not exists. Returns group id."""
with get_conn() as conn:
cur = conn.execute(
"""INSERT INTO groups (telegram_chat_id, creator_user_id)
VALUES (?, ?)
ON CONFLICT (telegram_chat_id) DO UPDATE SET creator_user_id = excluded.creator_user_id
WHERE groups.creator_user_id IS NULL OR groups.creator_user_id = excluded.creator_user_id
RETURNING id""",
(telegram_chat_id, creator_user_id),
)
return cur.fetchone()["id"]
def get_group(telegram_chat_id: int) -> Optional[dict]:
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM groups WHERE telegram_chat_id = ?",
(telegram_chat_id,),
).fetchone()
return _row_to_dict(row) if row else None
def get_group_creator_user_id(group_id: int) -> Optional[int]:
with get_conn() as conn:
row = conn.execute(
"SELECT creator_user_id FROM groups WHERE id = ?",
(group_id,),
).fetchone()
return row["creator_user_id"] if row else None
# ── Group Admins ────────────────────────────────────────────────────────────
def add_group_admin(group_id: int, user_id: int) -> bool:
"""Add user as admin. Returns True if newly added, False if already admin."""
with get_conn() as conn:
try:
conn.execute(
"INSERT INTO group_admins (group_id, user_id) VALUES (?, ?)",
(group_id, user_id),
)
return True
except sqlite3.IntegrityError:
return False
def remove_group_admin(group_id: int, user_id: int) -> bool:
"""Remove user from admins. Returns True if removed, False if not an admin."""
with get_conn() as conn:
cur = conn.execute(
"DELETE FROM group_admins WHERE group_id = ? AND user_id = ?",
(group_id, user_id),
)
return cur.rowcount > 0
def is_group_admin(group_id: int, user_id: int) -> bool:
with get_conn() as conn:
row = conn.execute(
"SELECT 1 FROM group_admins WHERE group_id = ? AND user_id = ?",
(group_id, user_id),
).fetchone()
return row is not None
def is_group_creator(group_id: int, user_id: int) -> bool:
return get_group_creator_user_id(group_id) == user_id
def get_user_by_username(username: str) -> Optional[dict]:
"""Look up user by username (without @)."""
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM users WHERE username = ?",
(username,),
).fetchone()
return _row_to_dict(row) if row else None
# ── Bounties ────────────────────────────────────────────────────────────────
def add_bounty(
group_id: int | None,
created_by_user_id: int,
informed_by_username: str,
text: str | None,
link: str | None,
due_date_ts: int | None,
) -> int:
"""Add a bounty. Returns bounty id. Raises ValueError on duplicate link."""
with get_conn() as conn:
try:
cur = conn.execute(
"""INSERT INTO bounties
(group_id, created_by_user_id, informed_by_username, text, link, due_date_ts)
VALUES (?, ?, ?, ?, ?, ?)
RETURNING id""",
(group_id, created_by_user_id, informed_by_username, text, link, due_date_ts),
)
return cur.fetchone()["id"]
except sqlite3.IntegrityError as e:
if "UNIQUE" in str(e) and "link" in str(e):
raise ValueError(f"Link already exists in this group: {link}")
raise
def get_bounty(bounty_id: int) -> Optional[dict]:
with get_conn() as conn:
row = conn.execute("SELECT * FROM bounties WHERE id = ?", (bounty_id,)).fetchone()
return _row_to_dict(row) if row else None
def get_group_bounties(group_id: int) -> list[dict]:
with get_conn() as conn:
return [_row_to_dict(r) for r in conn.execute(
"SELECT * FROM bounties WHERE group_id = ? ORDER BY created_at DESC",
(group_id,),
)]
def get_user_personal_bounties(user_id: int) -> list[dict]:
"""Bounties created by user in DM (group_id IS NULL)."""
with get_conn() as conn:
return [_row_to_dict(r) for r in conn.execute(
"SELECT * FROM bounties WHERE group_id IS NULL AND created_by_user_id = ? ORDER BY created_at DESC",
(user_id,),
)]
def update_bounty(
bounty_id: int,
text: str | None,
link: str | None,
due_date_ts: int | None,
) -> bool:
"""Update bounty fields. Returns True if updated. Raises ValueError on duplicate link."""
with get_conn() as conn:
try:
cur = conn.execute(
"""UPDATE bounties
SET text = COALESCE(?, text),
link = COALESCE(?, link),
due_date_ts = COALESCE(?, due_date_ts)
WHERE id = ?""",
(text, link, due_date_ts, bounty_id),
)
return cur.rowcount > 0
except sqlite3.IntegrityError as e:
if "UNIQUE" in str(e) and "link" in str(e):
raise ValueError(f"Link already exists in this group: {link}")
raise
def delete_bounty(bounty_id: int) -> bool:
with get_conn() as conn:
cur = conn.execute("DELETE FROM bounties WHERE id = ?", (bounty_id,))
return cur.rowcount > 0
# ── Tracking ────────────────────────────────────────────────────────────────
def track_bounty(user_id: int, bounty_id: int) -> bool:
"""Add bounty to user's tracking. Returns True if newly tracked, False if already tracking."""
with get_conn() as conn:
try:
conn.execute(
"INSERT INTO user_bounty_tracking (user_id, bounty_id) VALUES (?, ?)",
(user_id, bounty_id),
)
return True
except sqlite3.IntegrityError:
return False
def untrack_bounty(user_id: int, bounty_id: int) -> bool:
with get_conn() as conn:
cur = conn.execute(
"DELETE FROM user_bounty_tracking WHERE user_id = ? AND bounty_id = ?",
(user_id, bounty_id),
)
return cur.rowcount > 0
def is_tracking(user_id: int, bounty_id: int) -> bool:
with get_conn() as conn:
row = conn.execute(
"SELECT 1 FROM user_bounty_tracking WHERE user_id = ? AND bounty_id = ?",
(user_id, bounty_id),
).fetchone()
return row is not None
def get_user_tracked_bounties_in_group(user_id: int, group_id: int) -> list[dict]:
with get_conn() as conn:
return [_row_to_dict(r) for r in conn.execute(
"""SELECT b.* FROM bounties b
JOIN user_bounty_tracking t ON t.bounty_id = b.id
WHERE t.user_id = ? AND b.group_id = ?
ORDER BY b.created_at DESC""",
(user_id, group_id),
)]
def get_user_tracked_bounties_personal(user_id: int) -> list[dict]:
"""Tracked bounties where group_id IS NULL (personal)."""
with get_conn() as conn:
return [_row_to_dict(r) for r in conn.execute(
"""SELECT b.* FROM bounties b
JOIN user_bounty_tracking t ON t.bounty_id = b.id
WHERE t.user_id = ? AND b.group_id IS NULL
ORDER BY b.created_at DESC""",
(user_id,),
)]
# ── Reminders ───────────────────────────────────────────────────────────────
def get_bounties_due_soon(user_id: int, days: int = 7) -> list[dict]:
"""Get tracked bounties with due_date within `days` that haven't been reminded yet."""
now = int(time.time())
deadline = now + days * 86400
with get_conn() as conn:
return [_row_to_dict(r) for r in conn.execute(
"""SELECT b.*, u.username, u.telegram_user_id FROM bounties b
JOIN user_bounty_tracking t ON t.bounty_id = b.id
JOIN users u ON u.id = b.created_by_user_id
WHERE t.user_id = ?
AND b.due_date_ts IS NOT NULL
AND b.due_date_ts <= ?
AND b.due_date_ts >= ?
AND b.id NOT IN (
SELECT bounty_id FROM reminder_log WHERE user_id = ?
)
ORDER BY b.due_date_ts ASC""",
(user_id, deadline, now, user_id),
)]
def log_reminder(user_id: int, bounty_id: int) -> None:
with get_conn() as conn:
conn.execute(
"INSERT OR IGNORE INTO reminder_log (user_id, bounty_id) VALUES (?, ?)",
(user_id, bounty_id),
)
def get_all_user_ids() -> list[int]:
with get_conn() as conn:
return [row["telegram_user_id"] for row in conn.execute("SELECT telegram_user_id FROM users")]

49
apps/telegram-bot/schema.sql
View File

@@ -1,49 +0,0 @@
-- JIGAIDO Database Schema
CREATE TABLE IF NOT EXISTS groups (
id INTEGER PRIMARY KEY AUTOINCREMENT,
telegram_chat_id INTEGER UNIQUE NOT NULL,
creator_user_id INTEGER NOT NULL,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
CREATE TABLE IF NOT EXISTS group_admins (
group_id INTEGER REFERENCES groups(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL,
PRIMARY KEY (group_id, user_id)
);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
telegram_user_id INTEGER UNIQUE NOT NULL,
username TEXT,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
CREATE TABLE IF NOT EXISTS bounties (
id INTEGER PRIMARY KEY AUTOINCREMENT,
group_id INTEGER REFERENCES groups(id) ON DELETE CASCADE,
created_by_user_id INTEGER REFERENCES users(id),
informed_by_username TEXT NOT NULL,
text TEXT,
link TEXT,
due_date_ts INTEGER,
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(group_id, link)
);
CREATE TABLE IF NOT EXISTS user_bounty_tracking (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
bounty_id INTEGER NOT NULL REFERENCES bounties(id) ON DELETE CASCADE,
added_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(user_id, bounty_id)
);
CREATE TABLE IF NOT EXISTS reminder_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
bounty_id INTEGER NOT NULL REFERENCES bounties(id) ON DELETE CASCADE,
reminded_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE(user_id, bounty_id)
);

216
apps/telegram-bot/storage.py Normal file
View File

@@ -0,0 +1,216 @@
"""Per-group JSON file storage for JIGAIDO."""
import json
import os
import tempfile
from pathlib import Path
from typing import Optional
DATA_DIR = Path.home() / ".jigaido"
def _ensure_dirs() -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
def _group_dir(group_id: int) -> Path:
return DATA_DIR / str(group_id)
def _user_personal_dir(user_id: int) -> Path:
return DATA_DIR / str(user_id)
def _group_file(group_id: int) -> Path:
return _group_dir(group_id) / "group.json"
def _user_tracking_file(group_id: int, user_id: int) -> Path:
return _group_dir(group_id) / f"{user_id}.json"
def _user_personal_file(user_id: int) -> Path:
return _user_personal_dir(user_id) / "user.json"
def _atomic_write(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as tmp:
json.dump(data, tmp, indent=2)
tmp_path = tmp.name
os.rename(tmp_path, path)
def load_group_bounties(group_id: int) -> dict:
_ensure_dirs()
path = _group_file(group_id)
if not path.exists():
return {"bounties": [], "next_id": 1}
with open(path) as f:
return json.load(f)
def save_group_bounties(group_id: int, data: dict) -> None:
_atomic_write(_group_file(group_id), data)
def add_group_bounty(
group_id: int,
created_by_user_id: int,
text: Optional[str],
link: Optional[str],
due_date_ts: Optional[int],
) -> dict:
data = load_group_bounties(group_id)
bounty = {
"id": data["next_id"],
"created_by_user_id": created_by_user_id,
"text": text,
"link": link,
"due_date_ts": due_date_ts,
"created_at": int(os.path.getctime(_group_file(group_id)))
if _group_file(group_id).exists()
else 0,
}
data["bounties"].append(bounty)
data["next_id"] += 1
save_group_bounties(group_id, data)
return bounty
def update_group_bounty(
group_id: int,
bounty_id: int,
text: Optional[str],
link: Optional[str],
due_date_ts: Optional[int],
) -> bool:
data = load_group_bounties(group_id)
for bounty in data["bounties"]:
if bounty["id"] == bounty_id:
if text is not None:
bounty["text"] = text
if link is not None:
bounty["link"] = link
if due_date_ts is not None:
bounty["due_date_ts"] = due_date_ts
save_group_bounties(group_id, data)
return True
return False
def delete_group_bounty(group_id: int, bounty_id: int) -> bool:
data = load_group_bounties(group_id)
for i, bounty in enumerate(data["bounties"]):
if bounty["id"] == bounty_id:
data["bounties"].pop(i)
save_group_bounties(group_id, data)
return True
return False
def get_group_bounty(group_id: int, bounty_id: int) -> Optional[dict]:
data = load_group_bounties(group_id)
for bounty in data["bounties"]:
if bounty["id"] == bounty_id:
return bounty
return None
def load_user_tracking(group_id: int, user_id: int) -> dict:
path = _user_tracking_file(group_id, user_id)
if not path.exists():
return {"tracked": []}
with open(path) as f:
return json.load(f)
def save_user_tracking(group_id: int, user_id: int, data: dict) -> None:
_atomic_write(_user_tracking_file(group_id, user_id), data)
def track_bounty(group_id: int, user_id: int, bounty_id: int) -> bool:
data = load_user_tracking(group_id, user_id)
if any(t["bounty_id"] == bounty_id for t in data["tracked"]):
return False
data["tracked"].append({"bounty_id": bounty_id})
save_user_tracking(group_id, user_id, data)
return True
def untrack_bounty(group_id: int, user_id: int, bounty_id: int) -> bool:
data = load_user_tracking(group_id, user_id)
for i, t in enumerate(data["tracked"]):
if t["bounty_id"] == bounty_id:
data["tracked"].pop(i)
save_user_tracking(group_id, user_id, data)
return True
return False
def load_user_personal(user_id: int) -> dict:
path = _user_personal_file(user_id)
if not path.exists():
return {"bounties": [], "next_id": 1}
with open(path) as f:
return json.load(f)
def save_user_personal(user_id: int, data: dict) -> None:
_atomic_write(_user_personal_file(user_id), data)
def add_personal_bounty(
user_id: int, text: Optional[str], link: Optional[str], due_date_ts: Optional[int]
) -> dict:
data = load_user_personal(user_id)
bounty = {
"id": data["next_id"],
"text": text,
"link": link,
"due_date_ts": due_date_ts,
"created_at": 0,
}
data["bounties"].append(bounty)
data["next_id"] += 1
save_user_personal(user_id, data)
return bounty
def update_personal_bounty(
user_id: int,
bounty_id: int,
text: Optional[str],
link: Optional[str],
due_date_ts: Optional[int],
) -> bool:
data = load_user_personal(user_id)
for bounty in data["bounties"]:
if bounty["id"] == bounty_id:
if text is not None:
bounty["text"] = text
if link is not None:
bounty["link"] = link
if due_date_ts is not None:
bounty["due_date_ts"] = due_date_ts
save_user_personal(user_id, data)
return True
return False
def delete_personal_bounty(user_id: int, bounty_id: int) -> bool:
data = load_user_personal(user_id)
for i, bounty in enumerate(data["bounties"]):
if bounty["id"] == bounty_id:
data["bounties"].pop(i)
save_user_personal(user_id, data)
return True
return False
def get_personal_bounty(user_id: int, bounty_id: int) -> Optional[dict]:
data = load_user_personal(user_id)
for bounty in data["bounties"]:
if bounty["id"] == bounty_id:
return bounty
return None

28
apps/telegram-bot/tests/test_commands.py
View File

@@ -101,12 +101,21 @@ class TestParseArgs:
class TestFormatBounty:
def _row(self, id=1, text="Test bounty", link="https://example.com",
due_date_ts=None, informed_by_username="alice"):
def _row(
self,
id=1,
text="Test bounty",
link="https://example.com",
due_date_ts=None,
created_by_user_id=123456,
):
row = MagicMock()
row.__getitem__ = lambda s, k: {
"id": id, "text": text, "link": link,
"due_date_ts": due_date_ts, "informed_by_username": informed_by_username
"id": id,
"text": text,
"link": link,
"due_date_ts": due_date_ts,
"created_by_user_id": created_by_user_id,
}[k]
return row
@@ -155,12 +164,7 @@ class TestFormatBounty:
out = format_bounty(b)
assert "OVERDUE" in out
def test_informed_by_shown(self):
b = self._row(informed_by_username="bob")
def test_created_by_shown(self):
b = self._row(created_by_user_id=999)
out = format_bounty(b)
assert "@bob" in out
def test_informed_by_unknown_fallback(self):
b = self._row(informed_by_username=None)
out = format_bounty(b)
assert "@unknown" in out
assert "999" in out

298
apps/telegram-bot/tests/test_db.py
View File

@@ -1,298 +0,0 @@
"""Tests for db.py"""
import time
import pytest
import db as _db
class TestUsers:
def test_upsert_user_creates_new(self):
uid = _db.upsert_user(123, "alice")
assert uid > 0
row = _db.get_user_by_telegram_id(123)
assert row is not None
assert row["telegram_user_id"] == 123
assert row["username"] == "alice"
def test_upsert_user_updates_username(self):
# Two upserts to the same telegram_user_id: second one updates the username.
# Returns the same id both times (idempotent).
uid1 = _db.upsert_user(123, "alice")
uid2 = _db.upsert_user(123, "alice_updated")
assert uid1 == uid2
row = _db.get_user_by_telegram_id(123)
assert row["username"] == "alice_updated"
def test_get_user_by_telegram_id_not_found(self):
row = _db.get_user_by_telegram_id(999999)
assert row is None
class TestGroups:
def test_upsert_group_creates_new(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
assert gid > 0
row = _db.get_group(-100123)
assert row is not None
assert row["telegram_chat_id"] == -100123
assert row["creator_user_id"] == uid
def test_upsert_group_idempotent(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid1 = _db.upsert_group(-100123, uid)
gid2 = _db.upsert_group(-100123, uid)
assert gid1 == gid2
def test_get_group_creator_user_id(self, fresh_db):
uid = _db.upsert_user(1, "creator")
_db.upsert_group(-100123, uid)
assert _db.get_group_creator_user_id(_db.get_group(-100123)["id"]) == uid
def test_get_group_not_found(self, fresh_db):
row = _db.get_group(-999999)
assert row is None
class TestGroupAdmins:
def test_add_remove_is_admin(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
assert not _db.is_group_admin(gid, uid)
added = _db.add_group_admin(gid, uid)
assert added is True
assert _db.is_group_admin(gid, uid) is True
# Adding again returns False (already admin)
assert _db.add_group_admin(gid, uid) is False
removed = _db.remove_group_admin(gid, uid)
assert removed is True
assert _db.is_group_admin(gid, uid) is False
def test_remove_nonexistent_admin_returns_false(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
assert _db.remove_group_admin(gid, uid) is False
def test_is_group_creator(self, fresh_db):
uid = _db.upsert_user(1, "creator")
other = _db.upsert_user(2, "other")
gid = _db.upsert_group(-100123, uid)
_db.add_group_admin(gid, uid)
assert _db.is_group_creator(gid, uid) is True
assert _db.is_group_creator(gid, other) is False
def test_get_user_by_username(self, fresh_db):
uid = _db.upsert_user(1, "alice")
row = _db.get_user_by_username("alice")
assert row is not None
assert row["id"] == uid
assert _db.get_user_by_username("nobody") is None
class TestBounties:
def test_add_bounty_group(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
_db.add_group_admin(gid, uid)
bid = _db.add_bounty(
group_id=gid,
created_by_user_id=uid,
informed_by_username="bob",
text="Fix bug",
link="https://github.com/bob/repo",
due_date_ts=int(time.time()) + 86400,
)
assert bid > 0
bounty = _db.get_bounty(bid)
assert bounty["text"] == "Fix bug"
assert bounty["link"] == "https://github.com/bob/repo"
assert bounty["informed_by_username"] == "bob"
assert bounty["group_id"] == gid
def test_add_bounty_personal(self, fresh_db):
uid = _db.upsert_user(1, "alice")
bid = _db.add_bounty(
group_id=None,
created_by_user_id=uid,
informed_by_username="alice",
text="Personal reminder",
link=None,
due_date_ts=None,
)
assert bid > 0
bounty = _db.get_bounty(bid)
assert bounty["group_id"] is None
assert bounty["text"] == "Personal reminder"
def test_add_bounty_duplicate_link_rejected(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
_db.add_bounty(gid, uid, "user1", "text1", "https://example.com", None)
with pytest.raises(ValueError, match="Link already exists"):
_db.add_bounty(gid, uid, "user2", "text2", "https://example.com", None)
def test_add_bounty_null_link_allows_multiples(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
bid1 = _db.add_bounty(gid, uid, "user1", "text only 1", None, None)
bid2 = _db.add_bounty(gid, uid, "user2", "text only 2", None, None)
assert bid1 != bid2
def test_get_group_bounties(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
_db.add_group_admin(gid, uid)
_db.add_bounty(gid, uid, "user", "bounty1", None, None)
_db.add_bounty(gid, uid, "user", "bounty2", None, None)
bounties = _db.get_group_bounties(gid)
assert len(bounties) == 2
def test_get_user_personal_bounties(self, fresh_db):
uid = _db.upsert_user(1, "alice")
_db.add_bounty(None, uid, "alice", "personal1", None, None)
_db.add_bounty(None, uid, "alice", "personal2", None, None)
# Group bounty should not appear
other = _db.upsert_user(2, "bob")
gid = _db.upsert_group(-100, other)
_db.add_bounty(gid, other, "bob", "group bounty", None, None)
personal = _db.get_user_personal_bounties(uid)
assert len(personal) == 2
def test_update_bounty(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
bid = _db.add_bounty(gid, uid, "user", "old text", None, None)
_db.update_bounty(bid, "new text", None, None)
updated = _db.get_bounty(bid)
assert updated["text"] == "new text"
def test_update_bounty_duplicate_link_rejected(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
_db.add_bounty(gid, uid, "user1", "bounty1", "https://a.com", None)
bid2 = _db.add_bounty(gid, uid, "user2", "bounty2", None, None)
with pytest.raises(ValueError, match="Link already exists"):
_db.update_bounty(bid2, None, "https://a.com", None)
def test_delete_bounty(self, fresh_db):
uid = _db.upsert_user(1, "creator")
gid = _db.upsert_group(-100123, uid)
bid = _db.add_bounty(gid, uid, "user", "to delete", None, None)
assert _db.delete_bounty(bid) is True
assert _db.get_bounty(bid) is None
# Deleting again returns False
assert _db.delete_bounty(bid) is False
class TestTracking:
def test_track_untrack_is_tracking(self, fresh_db):
uid = _db.upsert_user(1, "alice")
uid2 = _db.upsert_user(2, "bob")
gid = _db.upsert_group(-100123, uid)
bid = _db.add_bounty(gid, uid, "alice", "task", None, None)
assert _db.track_bounty(uid, bid) is True
assert _db.is_tracking(uid, bid) is True
# Track again → False (already tracking)
assert _db.track_bounty(uid, bid) is False
# Other user tracking same bounty
assert _db.track_bounty(uid2, bid) is True
assert _db.is_tracking(uid2, bid) is True
# Untrack
assert _db.untrack_bounty(uid, bid) is True
assert _db.is_tracking(uid, bid) is False
assert _db.is_tracking(uid2, bid) is True # other user still tracking
# Untrack again → False
assert _db.untrack_bounty(uid, bid) is False
def test_get_user_tracked_bounties_in_group(self, fresh_db):
uid = _db.upsert_user(1, "alice")
gid = _db.upsert_group(-100123, uid)
bid1 = _db.add_bounty(gid, uid, "alice", "task1", None, None)
bid2 = _db.add_bounty(gid, uid, "alice", "task2", None, None)
# Different group bounty
other_gid = _db.upsert_group(-100124, uid)
bid3 = _db.add_bounty(other_gid, uid, "alice", "other group task", None, None)
_db.track_bounty(uid, bid1)
_db.track_bounty(uid, bid3)
tracked = _db.get_user_tracked_bounties_in_group(uid, gid)
assert len(tracked) == 1
assert tracked[0]["id"] == bid1
def test_get_user_tracked_bounties_personal(self, fresh_db):
uid = _db.upsert_user(1, "alice")
bid1 = _db.add_bounty(None, uid, "alice", "personal1", None, None)
bid2 = _db.add_bounty(None, uid, "alice", "personal2", None, None)
gid = _db.upsert_group(-100123, uid)
bid3 = _db.add_bounty(gid, uid, "alice", "group task", None, None)
_db.track_bounty(uid, bid1)
_db.track_bounty(uid, bid3)
tracked = _db.get_user_tracked_bounties_personal(uid)
assert len(tracked) == 1
assert tracked[0]["id"] == bid1
class TestReminders:
def test_get_bounties_due_soon(self, fresh_db):
uid = _db.upsert_user(1, "alice")
gid = _db.upsert_group(-100123, uid)
now = int(time.time())
# Due in 3 days (< 7 days)
bid_soon = _db.add_bounty(gid, uid, "alice", "soon", None, now + 3 * 86400)
# Due in 10 days (> 7 days)
_db.add_bounty(gid, uid, "alice", "later", None, now + 10 * 86400)
# No due date
bid_no_date = _db.add_bounty(gid, uid, "alice", "no date", None, None)
_db.track_bounty(uid, bid_soon)
_db.track_bounty(uid, bid_no_date)
due = _db.get_bounties_due_soon(uid, days=7)
assert len(due) == 1
assert due[0]["id"] == bid_soon
def test_reminder_log_prevents_duplicate_reminders(self, fresh_db):
uid = _db.upsert_user(1, "alice")
gid = _db.upsert_group(-100123, uid)
now = int(time.time())
bid = _db.add_bounty(gid, uid, "alice", "task", None, now + 2 * 86400)
_db.track_bounty(uid, bid)
due1 = _db.get_bounties_due_soon(uid, days=7)
assert len(due1) == 1
# Log that we reminded
_db.log_reminder(uid, bid)
# Should not appear again
due2 = _db.get_bounties_due_soon(uid, days=7)
assert len(due2) == 0
def test_get_all_user_ids(self, fresh_db):
_db.upsert_user(1, "alice")
_db.upsert_user(2, "bob")
ids = _db.get_all_user_ids()
assert sorted(ids) == [1, 2]