1
0

34 Commits
v1.1 ... master

Author SHA1 Message Date
aae03b51ee update start script, readme 2021-03-19 14:33:36 +00:00
c5358f4050 update requirements 2021-03-19 14:28:51 +00:00
140b4e799f fix remote_task_by_id creating tasks without IDs 2021-03-19 14:28:37 +00:00
190abc785c rearrange bot.py 2021-03-19 13:58:33 +00:00
13f64ee5d6 update src/db.py to use redis db instead of a json file lol 2021-03-19 13:55:08 +00:00
76313906bc add undo function 2021-02-27 15:22:08 +00:00
043cebe8b4 create function get_task_ids_from_context, use in functions delete and do 2021-02-27 15:16:51 +00:00
c08281b83a undo commit 669181239d, hide empty (deleted) tasks in ls function instead 2021-02-27 15:14:09 +00:00
5892e80e2e update .gitignore 2021-02-27 15:03:57 +00:00
f351fd8f90 add option to delete commands 2021-02-27 15:03:16 +00:00
a5ff687ade move fuzzy_get_task_id to src/db.py 2021-02-27 15:02:24 +00:00
669181239d don't return empty (deleted) tasks 2021-02-27 15:00:25 +00:00
0db78a6ab7 put pydo as git dependency in requirements.txt 2021-02-27 14:56:29 +00:00
452d4d2465 get bot token from commandline 2021-02-27 14:56:02 +00:00
Alvie Rahman
f3c3e671fa Merge pull request #1 from alvierahman90/dependabot/pip/cryptography-3.3.2
Bump cryptography from 3.3.1 to 3.3.2
2021-02-26 12:16:43 +00:00
18c4e1f193 create readme 2021-02-25 21:37:14 +00:00
dependabot[bot]
061c72a0fe Bump cryptography from 3.3.1 to 3.3.2
Bumps [cryptography](https://github.com/pyca/cryptography) from 3.3.1 to 3.3.2.
- [Release notes](https://github.com/pyca/cryptography/releases)
- [Changelog](https://github.com/pyca/cryptography/blob/main/CHANGELOG.rst)
- [Commits](https://github.com/pyca/cryptography/compare/3.3.1...3.3.2)

Signed-off-by: dependabot[bot] <support@github.com>
2021-02-25 21:35:05 +00:00
6212d92fc9 initial commit v2 !! 2021-02-25 21:34:33 +00:00
e9decc0ee7 clear repo 2021-02-25 21:30:45 +00:00
e1654bb023 Update help.md 2018-12-19 17:04:18 +00:00
14e942355b Update version number 2018-12-19 17:00:21 +00:00
ec07146c08 minor edits 2018-12-19 16:54:27 +00:00
343c03a6c1 merge rm, do, undo, and priority functions with fuzzy versions 2018-12-12 00:18:34 +00:00
09f55ea9b9 functions relating to a user all now have chat_id as first parameter 2018-12-11 23:05:56 +00:00
701ff42b22 Show output when runnign docker container 2018-12-11 22:54:19 +00:00
0b507b3cd3 move help.md into src/ 2018-09-16 12:59:47 +01:00
54e7af68f5 Change readme 2018-09-16 12:57:23 +01:00
31099219ec Update docker-compose example file 2018-09-16 12:47:41 +01:00
f671a3b198 config can be changed on the fly, moved to volume 2018-09-16 12:44:48 +01:00
a3ddc41118 Dockerize 2018-09-16 12:37:35 +01:00
6f16104524 Move Tasks module to different package 2018-09-16 12:16:14 +01:00
80bf201fa7 Add install instructions to readme, remove todo list from readme 2018-09-08 14:55:28 +01:00
0538f87706 Merge branch 'master' of github.com:alvierahman90/todo.txt_telegram 2018-09-08 14:26:40 +01:00
Alvie Rahman
eb666d34ce Create LICENSE 2018-09-04 16:40:12 +01:00
12 changed files with 209 additions and 813 deletions

6
.gitignore vendored
View File

@@ -1,2 +1,4 @@
config.json config
tasks.json env
__pycache__
*.swp

View File

@@ -1,16 +0,0 @@
# todo.txt as a Telegram bot
A bot to hold your todo.txt tasks
## commands
See [help.md](help.md)
## todo
- favourite commands
- reminders for tasks with due dates
- use a real database instead of json files
- multiple todo lists
- default filters
- ~~`Task` object generates new string on `__str__` method instead of relying on
accurate updating from functions~~
- ~~exporting all tasks~~

101
Task.py
View File

@@ -1,101 +0,0 @@
from re import match
from datetime import datetime as d
class Task:
def __init__(self, text):
# defaults
self.completion_date = None
self.creation_date = None
# for sorting, lower than all alphabet letters
self.priority = "{"
self.done = False
self.projects = []
self.contexts = []
self.specials = []
arguments = text.split(' ')
counter = 0
if arguments[counter] == 'x':
self.done = True
counter += 1
# try to get priority
priorities = match("\\([a-zA-Z]\\)", arguments[counter])
if priorities is not None:
self.priority = arguments[counter].split('(')[1].split(')')[0]
self.priority = self.priority.upper()
counter += 1
# try to get completion date if done
if self.done:
try:
self.completion_date = d.strptime(arguments[counter],
'%Y-%m-%d')
counter += 1
except ValueError as e:
pass
# try to get creation date
try:
self.creation_date = d.strptime(arguments[counter], '%Y-%m-%d')
counter += 1
except ValueError as e:
pass
# you cannot have a completion date w/o a creation date
if self.creation_date is None and self.completion_date is not None:
self.creation_date = self.completion_date
self.completion_date = None
# auto mark tasks with completion date as done
if self.completion_date is not None:
self.done = True
self.body = " ".join(arguments[counter:])
# the rest of the arguments may have projects, contexts, specials
for i in arguments[counter:]:
if len(i) < 1:
continue
if i[0] == '+':
self.projects.append(i.split('+')[1])
elif i[0] == '-':
self.contexts.append(i.split('-')[1])
elif ':' in i:
arguments = i.split(':')
key = arguments[0]
value = ":".join(arguments[1:])
special = {key: value}
self.specials.append(special)
def do(self):
if self.done:
return
self.done = True
def undo(self):
if not self.done:
return
self.done = False
def __str__(self):
text = ""
if self.done:
text += "x "
if self.priority != '{':
text += '(' + self.priority + ') '
if self.completion_date:
text += str(self.completion_date).split(' ')[0] + ' '
if self.creation_date:
text += str(self.creation_date).split(' ')[0] + ' '
text += self.body
return text

View File

@@ -1,148 +0,0 @@
#!/usr/bin/env python3
import unittest
import datetime
from Task import Task as T
class TaskTestCase(unittest.TestCase):
def test_basic(self):
text = "task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_basic_done(self):
text = "x basic test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, True)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_prioritized(self):
text = "(A) prioritized test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, 'A')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_prioritized_ignore_incorrect(self):
text = "(AA) prioritized test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_prioritized_done(self):
text = "x (A) prioritized test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, True)
self.assertEqual(task.priority, 'A')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_with_creation_date(self):
text = "2018-06-24 test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date,
datetime.datetime(2018, 6, 24, 0, 0))
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_with_creation_and_completion_date(self):
text = "x 2018-06-24 2018-05-24 test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, True)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date,
datetime.datetime(2018, 6, 24, 0, 0))
self.assertEqual(task.creation_date,
datetime.datetime(2018, 5, 24, 0, 0))
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_with_creation_and_completion_and_priority_date(self):
text = "x (B) 2018-06-24 2018-05-24 test task"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, True)
self.assertEqual(task.priority, 'B')
self.assertEqual(task.completion_date,
datetime.datetime(2018, 6, 24, 0, 0))
self.assertEqual(task.creation_date,
datetime.datetime(2018, 5, 24, 0, 0))
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
def test_special(self):
text = "special task special:value"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [{"special": "value"}])
def test_specials_with_colons(self):
text = "give muffin her pen back due:2028-07-10T14:28:15Z+0100"
task = T(text)
self.assertEqual(str(task), text)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, '{')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [{"due": "2028-07-10T14:28:15Z+0100"}])
def test_standardized_priority_case(self):
text = "(a) standard prioritization test"
text_standardized_priority = "(A) standard prioritization test"
task = T(text)
self.assertEqual(str(task), text_standardized_priority)
self.assertEqual(task.done, False)
self.assertEqual(task.priority, 'A')
self.assertEqual(task.completion_date, None)
self.assertEqual(task.creation_date, None)
self.assertEqual(task.projects, [])
self.assertEqual(task.contexts, [])
self.assertEqual(task.specials, [])
unittest.main()

View File

@@ -1 +0,0 @@
theme: jekyll-theme-minimal

515
bot.py
View File

@@ -1,515 +0,0 @@
#!/usr/bin/env python3
# coding=utf-8
"""
The bot script
Run this, I guess
"""
import re
import json
import time
import telepot
from telepot.loop import MessageLoop
from fuzzywuzzy import process
from fuzzywuzzy import fuzz
from Task import Task
VERSION = "v1.1"
PROPERTY_LAST_COMMAND = "last_command"
PROPERTY_LAST_ARGUMENTS = "last_arguments"
CONFIG_FILE = 'config.json'
ALPHABET = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
with open(CONFIG_FILE) as file:
CONFIG = json.loads(file.read())
BOT = telepot.Bot(CONFIG['token'])
def on_message(msg):
"""
The function which is run when MessageLoop receives an event
:param msg: The message object, passed by MessageLoop
"""
content_type, chat_type, chat_id = telepot.glance(msg)
print(content_type, chat_type, chat_id)
if content_type != 'text':
BOT.sendMessage(chat_id, "I can only understand text, sorry.")
return
text = msg['text']
command = text.split(' ')[0]
arguments = text.split(' ')[1:]
if command == '/last':
last_checks(chat_id)
command = get_property(PROPERTY_LAST_COMMAND, chat_id)
arguments = get_property(PROPERTY_LAST_ARGUMENTS, chat_id)
else:
set_property(PROPERTY_LAST_COMMAND, command, chat_id)
set_property(PROPERTY_LAST_ARGUMENTS, arguments, chat_id)
process_command(command, arguments, chat_id)
def process_command(command, arguments, chat_id):
"""
Processes the command sent by user
:param command: The command itself i.e /add
:param arguments: Anything else after it as a python list
:param chat_id: Telegram chat_id
"""
if command == '/start':
user_init(chat_id)
elif command == '/help':
user_help_info(chat_id)
elif command == '/add':
add_task(Task(" ".join(arguments)), chat_id)
elif command == '/rm':
rm_tasks(arguments, chat_id)
elif command == '/ls':
ls_tasks(arguments, chat_id)
elif command == '/do':
do_tasks(arguments, chat_id)
elif command == '/undo':
undo_tasks(arguments, chat_id)
elif command == '/export':
export_tasks(chat_id)
elif command == '/marco':
marco(chat_id)
elif command == '/delete_all_tasks':
delete_all_tasks(chat_id)
elif command == '/priority':
priority(chat_id, arguments)
elif command == '/fdo':
fuzzy_action(chat_id, ' '.join(arguments), do_tasks)
elif command == '/fundo':
fuzzy_action(chat_id, ' '.join(arguments), undo_tasks)
elif command == '/frm':
fuzzy_action(chat_id, ' '.join(arguments), rm_tasks)
elif command == '/fpriority':
fuzzy_priority(chat_id, arguments)
else:
set_property(PROPERTY_LAST_COMMAND, '/add', chat_id)
set_property(PROPERTY_LAST_ARGUMENTS, arguments, chat_id)
# command has to prefixed here since there is no actual command with a
# preceding slash
add_task(Task(command + " " + " ".join(arguments)), chat_id)
def add_task(task, chat_id):
"""
Adds a task
:param task: A Task object
:param chat_id: A numerical telegram chat_id
"""
tasks = get_tasks(chat_id)
tasks.append(task)
set_tasks(tasks, chat_id)
BOT.sendMessage(chat_id, "Added task: {0}".format(task))
def rm_tasks(task_ids, chat_id):
"""
Delete multiple tasks
:param task_ids: An iterable of IDs of task objects
:param chat_id: A numerical telegram chat_id
"""
tasks = get_tasks(chat_id)
for i in task_ids:
if not is_task_id_valid(chat_id, i):
continue
set_tasks([x for x in tasks if str(tasks[int(i)]) != str(x)], chat_id)
BOT.sendMessage(chat_id, "Removed task: {0}".format(tasks[int(i)]))
def get_property(property_name, chat_id):
"""
// TODO figure out what this does
:param property_name:
:param chat_id:
:return:
"""
with open(CONFIG['tasks_file']) as tasks_file:
info_dict = json.loads(tasks_file.read())
key = property_name + ":" + str(chat_id)
if key in info_dict.keys():
return info_dict[key]
return None
def set_property(property_name, value, chat_id):
"""
// TODO figure out what this does
:param property_name:
:param value:
:param chat_id:
"""
with open(CONFIG['tasks_file']) as tasks_file:
info_dict = json.loads(tasks_file.read())
key = property_name + ":" + str(chat_id)
info_dict[key] = value
with open(CONFIG['tasks_file'], 'w') as tasks_file:
info_dict = tasks_file.write(json.dumps(info_dict))
def get_tasks(chat_id, raw=False):
"""
Returns a list of tasks
:param chat_id: A numerical telegram chat_id, or None to get tasks for all users
:param raw: Defaults to False, raw returns the tasks as strings
:return: Returns a python list of tasks, or a python dict if raw is True
"""
with open(CONFIG['tasks_file']) as tasks_file:
tasks_dict = json.loads(tasks_file.read())
if chat_id is None:
return tasks_dict
chat_id = str(chat_id)
if chat_id not in tasks_dict:
tasks_dict[chat_id] = ""
if raw:
return tasks_dict[chat_id]
tasks = []
# even if the string is empty, split will return a list of one, which
# which creates a task that doesn't exist and without any content when user
# has no tasks
if tasks_dict[chat_id] == "":
return tasks
for i in tasks_dict[chat_id].split('\n'):
tasks.append(Task(i))
return tasks
def get_task(task_id, chat_id):
"""
Returns single task
:param task_id: ID of task
:param chat_id: Telegram chat_id
:return: Task object or none if task_id is invalid
"""
if not is_task_id_valid(chat_id, task_id):
return None
return get_tasks(chat_id)[int(task_id)]
def set_tasks(tasks, chat_id):
"""
Overwrite the existing tasks with a new list
:param tasks: Iterable of Task objects
:param chat_id: Telegram chat_id
"""
task_dict = get_tasks(None)
texts = []
for i in tasks:
texts.append(str(i))
plaintext = "\n".join(texts)
task_dict[chat_id] = plaintext
with open(CONFIG['tasks_file'], 'w+') as tasks_file:
tasks_file.write(json.dumps(task_dict))
def set_task(task_id, task, chat_id):
"""
Overwrite a single task by ID
:param task_id: ID of the task
:param task: Task object itself
:param chat_id: Telegram chat_id
"""
if not is_task_id_valid(chat_id, task_id):
return
tasks = get_tasks(chat_id)
tasks[task_id] = task
set_tasks(tasks, chat_id)
def ls_tasks(arguments, chat_id):
"""
Send a list of tasks to user
:param arguments: Iterable of strings
:param chat_id: Telegram chat_id
"""
tasks = get_tasks(chat_id)
if len(tasks) < 1:
BOT.sendMessage(chat_id, "You have no tasks.")
return
counter = 0
for i, value in enumerate(tasks, start=0):
tasks[i] = (counter, value)
counter += 1
tasks = sorted(tasks, key=lambda tup: tup[1].priority)
# create list of filters
filters = []
nfilters = [] # inverse filter
for i in arguments:
if re.match("^f:", i) is not None:
filters.append(i.split("f:")[1])
elif re.match("^filter:", i) is not None:
filters.append(i.split("filter:")[1])
elif re.match("^!f:", i) is not None:
nfilters.append(i.split("!f:")[1])
elif re.match("^!filter:", i) is not None:
nfilters.append(i.split("!filter:")[1])
text = "Tasks:\n"
for i in tasks:
task = i[1]
counter += 1
filter_pass = True
if not task.done and ":only-done" in arguments:
continue
elif task.done and ":only-done" in arguments:
pass
elif task.done and ":show-done" not in arguments:
continue
# filter checking
for j in filters:
if j not in str(task):
filter_pass = False
break
# needs continue statement after each filter list as filter_pass
# gets reset
if not filter_pass:
continue
for j in nfilters:
if j in str(task):
filter_pass = False
break
if not filter_pass:
continue
text += str(i[0]) + " " + str(i[1]) + "\n"
BOT.sendMessage(chat_id, text)
def do_tasks(task_ids, chat_id):
"""
Mark tasks by ID as done
:param task_ids: Iterable of task IDs
:param chat_id: Telegram chat_id
"""
for i in task_ids:
if not is_task_id_valid(chat_id, i):
continue
task = get_task(int(i), chat_id)
task.do()
set_task(int(i), task, chat_id)
BOT.sendMessage(chat_id, "Did task {1}: {0}".format(str(task), i))
def undo_tasks(task_ids, chat_id):
"""
Mark tasks as not done
:param task_ids: Iterable of task IDs
:param chat_id: Telegram chat_id
"""
for i in task_ids:
if not is_task_id_valid(chat_id, i):
continue
task = get_task(int(i), chat_id)
task.undo()
set_task(int(i), task, chat_id)
BOT.sendMessage(chat_id, "Undid task {1}: {0}".format(str(task), i))
def export_tasks(chat_id):
"""
Send all tasks to user as standard todo.txt format, to use in other apps
:param chat_id: Telegram chat_id
"""
text = get_tasks(chat_id, raw=True)
if text == "":
BOT.sendMessage(chat_id, "No tasks.")
return
BOT.sendMessage(chat_id, "RAW:")
BOT.sendMessage(chat_id, text)
def marco(chat_id):
"""
Sends the message "Polo" to user, tests if the bot is up
:param chat_id: Telegram chat_id
"""
BOT.sendMessage(chat_id, "Polo")
def last_checks(chat_id):
"""
Checks if the user has sent a command already
:param chat_id: Telegram chat_id
"""
if get_property(PROPERTY_LAST_ARGUMENTS, chat_id) is None or \
get_property(PROPERTY_LAST_COMMAND, chat_id) is None:
BOT.sendMessage(chat_id, "No recorded last command")
def user_init(chat_id):
"""
The function which is run to set up a new user
:param chat_id: Telegram chat_id
"""
BOT.sendMessage(chat_id, "Welcome to todo.txt but as a Telegram bot. Run"
" /help to get started")
def user_help_info(chat_id):
"""
The help text sent to user
:param chat_id: Telegram chat_id
"""
with open('help.md') as help_file:
text = help_file.read()
text += "\ntodo.txt bot for Telegram version {0}".format(VERSION)
text += "\n[View help on GitHub](alvierahman90.github.io/todo.txt_telegram/help.html)"
BOT.sendMessage(chat_id, text, parse_mode='Markdown')
def delete_all_tasks(chat_id):
"""
Deletes all the tasks for a user. Also exports the tasks in case the user
made a mistake.
:param chat_id: Telegram chat id
"""
export_tasks(chat_id)
set_tasks([], chat_id)
BOT.sendMessage(chat_id, "Deleted all tasks.")
def priority(chat_id, arguments):
"""
Changes the priority of a task
"""
if len(arguments) < 2:
BOT.sendMessage(chat_id, "Not enough arguments: /priority PRIORITY"
"ID-OF-TASK [ID-OF-TASK...]")
return
priorities = list(ALPHABET)
priorities.append('NONE')
if arguments[0].upper() not in priorities:
BOT.sendMessage(chat_id, "Priority (first argument) must be letter or"
"'none'")
return
new_priority = arguments[0].upper()
# This is what no priority is defined as for the sorting
if new_priority == 'NONE':
new_priority = '{'
del arguments[0]
tasks = get_tasks(chat_id)
for i in arguments:
if not is_task_id_valid(chat_id, i):
continue
i = int(i)
BOT.sendMessage(chat_id, "Setting priority of '{}'.".format(tasks[i]))
tasks[i].priority = new_priority
set_tasks(tasks, chat_id)
return
def is_task_id_valid(chat_id, task_id):
"""
Checks if task_id provided is an integer and in the list
:param chat_id: Telegram chat id
:param task_id: ID of the task to check
:return: the task_id as integer if valid, otherwise False
"""
def fail():
"""
Prints failure message
"""
BOT.sendMessage(chat_id, "Invalid task ID '{0}' - IDs are "
"integers and must actually exist (run /ls)"
"".format(str(task_id)))
return False
if isinstance(task_id, int):
real_task_id = int(task_id)
elif isinstance(task_id, str):
if task_id.isnumeric():
real_task_id = int(task_id)
else:
return fail()
else:
return fail()
print(real_task_id)
print(len(get_tasks(chat_id)))
if real_task_id < len(get_tasks(chat_id)):
return real_task_id
return fail()
def fuzzy_get_task_id_from_text(chat_id, text):
"""
Fuzzy searches for the closest match to the text string given
:param chat_id: Telegram chat_id
:param text: The string to fuzzy match
:return: task_id, matchness, task_text as a tuple
"""
tasks = [str(x) for x in get_tasks(chat_id)]
task_text, matchness = process.extractOne(text, tasks,
scorer=fuzz.token_sort_ratio)
task_id = tasks.index(task_text)
return task_id, matchness
def fuzzy_action(chat_id, text, function):
"""
Marks the task most similar to `text` as done
:param chat_id: Telegram chat_id
:param text: text to match to a task to perform function on it
:param function: the function with which to process the task_id
"""
task_id, matchness = fuzzy_get_task_id_from_text(chat_id, text)
return function([task_id], chat_id)
def fuzzy_priority(chat_id, arguments):
"""
Sets the priority of the closest matching task to text
:param chat_id: Telegram chat_id
:param text: text to match to a task to perform function on it
:param function: the function with which to process the task_id
"""
text = ' '.join(arguments[1:])
task_id, matchness = fuzzy_get_task_id_from_text(chat_id, text)
return priority(chat_id, [arguments[0], task_id])
if __name__ == "__main__":
MessageLoop(BOT, on_message).run_as_thread()
while True:
time.sleep(1)

30
help.md
View File

@@ -1,30 +0,0 @@
# commands
Anything sent without a command is assumed to be a new task to be added
## actions on tasks
- `/add <task-text>` - Add a new task
- `/do <id> [id [id [id]...]]` - Do task(s)
- `/priority <priority> <id> [id [id [id]...]]` - Set the priority of task(s)
- `/rm <id> [id [id [id]...]]` - Remove task(s)
- `/undo <id> [id [id [id]...]]` - undo task(s)
### fuzzy actions
- `/fdo <text to match>`
- `/fpriority <text to match>`
- `/frm <text to match>`
- `/fundo <text to match>`
## general
- `/export` - Send all tasks as plaintext
- `/help` - Show help information
- `/ls [filters]` - List tasks
- `/last` - Run the last command sent
- `/marco` - Test if bot is up
## /ls filters
- `f[ilter]:<text>` - Tasks must have this text in it
- `!f[ilter]:<text>` - Tasks must **not** have this text in it
- `:show-done` - Show and include done tasks
- `:only-done` - Show only done tasks

16
readme.md Normal file
View File

@@ -0,0 +1,16 @@
# todo.txt bot for telegram!
under heavy development
you must have a redis server installed
## running
1. `python3 -m venv env`
1. Create `config` file:
```bash
TOKEN="telegram bot token from bot father"
```
1. `bash start.sh`

14
requirements.txt Normal file
View File

@@ -0,0 +1,14 @@
APScheduler==3.6.3
certifi==2020.12.5
cffi==1.14.4
cryptography==3.3.2
fuzzywuzzy==0.18.0
pycparser==2.20
git+https://github.com/alvierahman90/pydo.git
python-Levenshtein==0.12.2
python-telegram-bot==13.2
pytz==2021.1
six==1.15.0
tornado==6.1
tzlocal==2.1
redis==3.5.3

102
src/bot.py Normal file
View File

@@ -0,0 +1,102 @@
import sys
if len(sys.argv) != 2 or sys.argv[1] == 'help':
printl("USAGE: " + sys.argv[0] + " BOT_TOKEN")
sys.exit(0)
from telegram.ext import Updater
from telegram.ext import CommandHandler
from telegram.ext import MessageHandler, Filters
import logging
import db
import pydo
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
updater = Updater(token=sys.argv[1], use_context=True)
dispatcher = updater.dispatcher
def new_task(update, context):
db.add_task(update.effective_user, pydo.Task(update.message.text))
context.bot.send_message(chat_id=update.effective_chat.id,
text=f"created task: {update.message.text}")
def delete(update, context):
task_ids = db.get_task_ids_from_context(update.effective_user, context)
for task_id in task_ids:
task = db.remove_task_by_id(update.effective_user, task_id)
if task is None:
context.bot.send_message(chat_id=update.effective_chat.id,
text="task not found :("
)
else:
context.bot.send_message(chat_id=update.effective_chat.id,
text=f"deleted task: {task}")
def do(update, context):
task_ids = db.get_task_ids_from_context(update.effective_user, context)
for task_id in task_ids:
task = db.get_task(update.effective_user, task_id)
task.do()
db.update_task(update.effective_user, task)
for id in task_ids:
context.bot.send_message(chat_id=update.effective_chat.id,
text='completed task: ' + str(db.get_task(update.effective_user, id))
)
def ls(update, context):
tasks = db.get_all_user_tasks(update.effective_user)
r = ""
for task in tasks:
if str(task) == "":
continue
if task.done is ('done' in update.message['text']) or ('all' in update.message['text']):
r+= f"{task.id} {str(task)}"
r+= '\n'
r = r if r != "" else "no tasks!"
context.bot.send_message(chat_id=update.effective_chat.id,
text=r)
def start(update, context):
context.bot.send_message(chat_id=update.effective_chat.id,
text="hiiiiiiiiii. what do you need to get done today?")
def undo(update, context):
task_ids = db.get_task_ids_from_context(update.effective_user, context)
for task_id in task_ids:
task = db.get_task(update.effective_user, task_id)
task.undo()
db.update_task(update.effective_user, task)
for id in task_ids:
context.bot.send_message(chat_id=update.effective_chat.id,
text='undone task: ' + str(db.get_task(update.effective_user, id))
)
dispatcher.add_handler(MessageHandler(Filters.text & (~Filters.command), new_task))
dispatcher.add_handler(CommandHandler('delete', delete))
dispatcher.add_handler(CommandHandler('do', do))
dispatcher.add_handler(CommandHandler('ls', ls))
dispatcher.add_handler(CommandHandler('rm', delete))
dispatcher.add_handler(CommandHandler('start', start))
dispatcher.add_handler(CommandHandler('undo', undo))
updater.start_polling()

66
src/db.py Normal file
View File

@@ -0,0 +1,66 @@
import pydo
import telegram
import redis
from fuzzywuzzy import process as fuzzyprocess
from fuzzywuzzy import utils as fuzzyutils
def _redis_user_tasks_key(user_id):
return f'user_tasks:{user_id}'
r = redis.Redis(host='localhost', port=6379,
charset='utf-8', decode_responses=True)
def get_all_user_tasks(user) -> "list of pydo.Task":
global r
task_list = r.lrange(_redis_user_tasks_key(user.id), 0, -1)
response = []
for id, task_str in enumerate(task_list):
task = pydo.Task(task_str)
task.id = id
response.append(task)
return response
def export_user_tasks(user: telegram.User) -> str:
return '\n'.join(r.lrange(_redis_user_tasks_key(user.id)), 0, -1)
def get_task(user: telegram.User, task_id: int) -> pydo.Task:
for task in get_all_user_tasks(user):
if int(task.id) == int(task_id):
return task
def fuzzy_get_task_id(user, text):
task_strs = [str(task) for task in get_all_user_tasks(user)]
return task_strs.index(fuzzyprocess.extractOne(text, task_strs)[0])
def get_task_ids_from_context(user, context):
for arg in context.args:
if not arg.isnumeric():
task_ids = [fuzzy_get_task_id(user, ' '.join(context.args))]
break
else:
task_ids = [int(x) for x in context.args]
return task_ids
def add_task(user: telegram.User, task: pydo.Task) -> pydo.Task:
r.rpush(_redis_user_tasks_key(user.id), str(task))
return task
def update_task(user: telegram.User, new_task: pydo.Task) -> pydo.Task:
r.lset(_redis_user_tasks_key(user.id), new_task.id, str(new_task))
return new_task
def remove_task_by_id(user: telegram.User, task_id: int) -> int:
# instead of removing an item, set it's text value to nothing, so that all other tasks' ids
# don't change
task = get_task(user, task_id)
if task is not None:
update_task(user, pydo.Task("", id=task_id))
return task
def remove_task(user: telegram.User, task: pydo.Task) -> pydo.Task:
return remove_task_by_id(user.id, task.id)

7
start.sh Executable file
View File

@@ -0,0 +1,7 @@
#!/usr/bin/env bash
source config
# source env if not in docker container
[[ `grep 'docker' /proc/self/cgroup` ]] && source env/bin/activate
python src/bot.py "$TOKEN"