371 lines
10 KiB
Python
371 lines
10 KiB
Python
# import sqlite3
|
|
import getpass
|
|
import datetime
|
|
import logging
|
|
import sys
|
|
|
|
from action import record_action
|
|
from action import TypeAction
|
|
from action import set_active
|
|
from args import get_string_arg
|
|
from args import arg_number
|
|
|
|
from utils import get_pits_path
|
|
|
|
import note
|
|
|
|
|
|
class Task:
|
|
def __init__(self, status=None, priority=None):
|
|
self.name = None
|
|
self.status = status
|
|
self.priority = priority
|
|
self.date = None
|
|
self.time = None
|
|
|
|
|
|
def handle_task(args, last_action, conn):
|
|
logging.info("> handle task")
|
|
logging.debug("args: %s", args)
|
|
|
|
if not args:
|
|
list_task(last_action.project_id, last_action.task_id, conn)
|
|
sys.exit(0)
|
|
|
|
if args[0].isdigit():
|
|
view_task_set_active(arg_number(args[0]), last_action, conn)
|
|
elif args[0] == "-c":
|
|
if len(args) == 1:
|
|
print("missing task name")
|
|
sys.exit(1)
|
|
|
|
task = Task("open", "normal")
|
|
task.name = get_string_arg(args[1:], "task name")
|
|
parse_task_args(task, args[2:])
|
|
create_task(task, last_action.project_id, conn)
|
|
elif args[0] == "-e":
|
|
if len(args) == 1:
|
|
print("nothing to update")
|
|
sys.exit(1)
|
|
|
|
task = Task()
|
|
|
|
if args[1].isdigit():
|
|
task_id = int(args[1])
|
|
parse_task_args(task, args[2:])
|
|
else:
|
|
task_id = (
|
|
last_action.task_id
|
|
) # FIXME If no active tast - use incremental args_i
|
|
parse_task_args(task, args[1:])
|
|
logging.debug("Task: (%s) %s", task_id, task)
|
|
|
|
edit_task(task, task_id, conn)
|
|
elif args[0] == "-d":
|
|
# FIXME Duplicate code
|
|
if len(args) > 1:
|
|
delete_task(arg_number(args[1]), conn)
|
|
else:
|
|
delete_task(last_action.task_id, conn)
|
|
elif args[0] == "-m":
|
|
# Try another logic
|
|
args_i = 1
|
|
if len(args) < 3:
|
|
print("missing project number")
|
|
sys.exit(1)
|
|
|
|
task_id = last_action.task_id
|
|
|
|
# Task id
|
|
if args[args_i].isdigit():
|
|
task_id = arg_number(args[args_i])
|
|
args_i += 1
|
|
|
|
if not task_id:
|
|
print("No active task to move.")
|
|
sys.exit(1)
|
|
|
|
# Project option - if not fail
|
|
if args[args_i] != "-p":
|
|
print("missing project number")
|
|
sys.exit(1)
|
|
else:
|
|
args_i += 1
|
|
|
|
moving_task(task_id, arg_number(args[args_i]), conn)
|
|
else:
|
|
print(f"Invalid task option: {args[0]}")
|
|
|
|
|
|
def parse_task_args(task, args):
|
|
if not args:
|
|
return
|
|
|
|
i = 0
|
|
while i < len(args):
|
|
logging.debug(args[i])
|
|
|
|
if args[i] == "-s":
|
|
i += 1
|
|
task.status = get_string_arg(args[i : i + 1], "task status")
|
|
elif args[i] == "-n":
|
|
i += 1
|
|
task.name = get_string_arg(args[i : i + 1], "task name")
|
|
elif args[i] == "-p":
|
|
i += 1
|
|
task.priority = get_string_arg(args[i : i + 1], "task priority")
|
|
elif args[i] == "-d":
|
|
i += 1
|
|
task.date = get_string_arg(args[i : i + 1], "task date")
|
|
elif args[i] == "-t":
|
|
i += 1
|
|
task.time = get_string_arg(args[i : i + 1], "task time")
|
|
else:
|
|
print(f"Invalid task option: {args[i]}")
|
|
sys.exit(1)
|
|
i += 1
|
|
|
|
|
|
def create_task(task, project_id, conn):
|
|
logging.info(">> Create task")
|
|
|
|
if project_id == 0:
|
|
print("No project selected")
|
|
sys.exit(1)
|
|
|
|
query = """
|
|
INSERT INTO
|
|
task (project_id, username, name, status, priority, date, time, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?);
|
|
"""
|
|
|
|
# TODO Date & time
|
|
date = ""
|
|
time = ""
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
query,
|
|
(
|
|
project_id,
|
|
getpass.getuser(),
|
|
task.name,
|
|
task.status,
|
|
task.priority,
|
|
date,
|
|
time,
|
|
datetime.datetime.now(),
|
|
),
|
|
)
|
|
|
|
task_id = cursor.lastrowid
|
|
|
|
logging.debug("CREATE ACTION " + str(project_id) + str(task_id))
|
|
|
|
task_details = [
|
|
item
|
|
for item in task.__dict__.items()
|
|
if item[1] is not None and item[0] is not "name"
|
|
]
|
|
log_args = ", ".join("%s: %s" % (k, v) for k, v in task_details)
|
|
|
|
action_message = "{} ({}, project: {})".format(task.name, log_args, project_id)
|
|
|
|
record_action(cursor, TypeAction.CREATE, action_message, project_id, task_id)
|
|
print("created task {}: {}".format(task_id, action_message))
|
|
|
|
|
|
def edit_task(task, task_id, conn):
|
|
logging.info(">> Edit task")
|
|
|
|
update_args = [item for item in task.__dict__.items() if item[1] is not None]
|
|
|
|
if not update_args:
|
|
print("nothing to update")
|
|
print(
|
|
"Tips: if you want to activate a task, just do '{} task <task_id>'".format(
|
|
get_pits_path()
|
|
)
|
|
)
|
|
sys.exit(1)
|
|
|
|
logging.debug("Task update args: %s", update_args)
|
|
|
|
# Retrieve name if not updated
|
|
task_name = None
|
|
if "name" not in update_args[0]:
|
|
task_name, _ = get_task_name_project_id(task_id, conn)
|
|
|
|
query = "UPDATE task SET {} WHERE id = ?"
|
|
query = query.format(", ".join("%s = '%s'" % (k, v) for k, v in update_args))
|
|
logging.debug("update task query: %s")
|
|
|
|
cursor = conn.cursor()
|
|
logging.debug("Do a task update")
|
|
cursor.execute(query, (task_id,))
|
|
|
|
log_args = ", ".join("%s: %s" % (k, v) for k, v in update_args)
|
|
|
|
if task_name:
|
|
print("updated task {}: {} ({})".format(task_id, task_name, log_args))
|
|
else:
|
|
print("updated task {}: ({})".format(task_id, log_args))
|
|
|
|
logging.debug("UPDATE TASK - ACTION MESSAGE : %s", log_args)
|
|
record_action(
|
|
cursor,
|
|
TypeAction.UPDATE,
|
|
"{}({})".format(task_name + " " if task_name else "", log_args),
|
|
task_id=task_id,
|
|
)
|
|
|
|
|
|
def moving_task(task_id, dest_project_id, conn):
|
|
logging.info(">> Moving task")
|
|
|
|
_, or_project_id = get_task_name_project_id(task_id, conn)
|
|
|
|
if or_project_id == dest_project_id:
|
|
print("Task already in project {} - nothing to do".format(dest_project_id))
|
|
sys.exit(0)
|
|
|
|
query = "UPDATE task SET project_id = ? WHERE id = ?;"
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, (dest_project_id, task_id))
|
|
|
|
if cursor.rowcount != 1:
|
|
print("Problem occur when moving task...")
|
|
sys.exit(1)
|
|
|
|
print(
|
|
"moved task {}: from project {} to project {}".format(
|
|
task_id, or_project_id, dest_project_id
|
|
)
|
|
)
|
|
record_action(
|
|
cursor,
|
|
TypeAction.MOVE,
|
|
"from project {} to project {}".format(or_project_id, dest_project_id),
|
|
project_id=dest_project_id,
|
|
task_id=task_id,
|
|
)
|
|
|
|
|
|
def delete_task(task_id, conn):
|
|
logging.info(">> Remove task")
|
|
|
|
task_name, project_id = get_task_name_project_id(task_id, conn)
|
|
|
|
cursor = conn.cursor()
|
|
|
|
# Cascade deleting: note
|
|
query = "SELECT id FROM note WHERE task_id = ?;"
|
|
cursor.execute(query, (task_id,))
|
|
|
|
deleted_note = 0
|
|
for row in cursor.fetchall():
|
|
note_id = row[0]
|
|
note.delete_note(note_id, conn)
|
|
deleted_note += 1
|
|
# FIXME Use a delete_note_for_task function in note module?
|
|
|
|
# Remove task
|
|
query = "DELETE FROM task WHERE id = ?;"
|
|
|
|
cursor.execute(query, (task_id,))
|
|
if cursor.rowcount != 1:
|
|
logging.error("DELETE FAILED")
|
|
print("could not find task {}".format(task_id))
|
|
sys.exit(1)
|
|
|
|
if deleted_note:
|
|
log_message = "{} with {} note{} (project: {})".format(
|
|
task_name, deleted_note, "s" if deleted_note > 1 else "", project_id
|
|
)
|
|
else:
|
|
log_message = "{} (project: {})".format(task_name, project_id)
|
|
|
|
print("deleted task {}: {}".format(task_id, log_message))
|
|
record_action(cursor, TypeAction.DELETE, log_message, project_id, task_id)
|
|
|
|
|
|
def list_task(active_project_id, active_task_id, conn):
|
|
logging.info(">> No arguments")
|
|
|
|
query = """
|
|
SELECT id, username, status, priority, name,
|
|
(SELECT count(*) FROM note WHERE note.task_id = task.id)
|
|
FROM task WHERE project_id = ?;
|
|
""" # TODO Date & time
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, (active_project_id,))
|
|
for row in cursor.fetchall():
|
|
task_id, username, status, priority, name, nb_note = row
|
|
print(
|
|
"{:1} {:2d}: ({:8}) | {} | {} | {} ({} notes)".format(
|
|
"*" if active_task_id == task_id else "",
|
|
task_id,
|
|
username,
|
|
status,
|
|
priority,
|
|
name,
|
|
nb_note,
|
|
)
|
|
)
|
|
|
|
|
|
def view_task_set_active(task_id, last_action, conn):
|
|
# FIXME duplicate with list_task / view_project_set_active
|
|
|
|
# Retrieve selected task
|
|
query = """
|
|
SELECT id, username, name, project_id, status, priority,
|
|
(SELECT count(*) FROM note WHERE note.task_id = task.id)
|
|
FROM task WHERE id = ?;
|
|
""" # TODO Date & time
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, (task_id,))
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
print("Could not find task {}".format(task_id))
|
|
sys.exit(1)
|
|
|
|
print("* {:d}: ({}) {} (project: {}, status: {}, priority: {}, {} note(s))".format(*row))
|
|
|
|
# FIXME duplicate with list_note
|
|
query = """
|
|
SELECT id, username, message FROM note WHERE task_id = ?
|
|
"""
|
|
cursor.execute(query, (task_id,))
|
|
for row in cursor.fetchall():
|
|
logging.debug("Note row: %s", row)
|
|
note_id, username, message = row
|
|
print(
|
|
" {} {:d}: ({}) {}".format(
|
|
"*" if last_action.note_id == row[0] else " ",
|
|
note_id,
|
|
username,
|
|
message,
|
|
)
|
|
)
|
|
set_active(cursor, task_id=task_id)
|
|
|
|
|
|
def get_task_name_project_id(task_id, conn):
|
|
""" Return a tuple with task name and project id of task"""
|
|
|
|
query = "SELECT name, project_id FROM task WHERE id = ?"
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, (task_id,))
|
|
return cursor.fetchone()
|
|
|
|
|
|
# $ pit t 11
|
|
# * 11: (budd) a task (project: 1, status: in progress, priority: high, date: Apr 26, 2018, time: 5:00, 2 notes)
|
|
# 6: (budd) bonjour le monde
|
|
# * 7: (budd) bonjour le monde
|