hamster-tools/hamstertools/__init__.py

719 lines
19 KiB
Python
Raw Normal View History

2019-12-27 10:28:25 +00:00
#!/usr/bin/env python3.7
2020-09-30 18:36:27 +00:00
import csv
from datetime import datetime
2023-07-08 12:13:07 +00:00
from itertools import chain
2019-12-27 10:28:25 +00:00
from pathlib import Path
import sys
import click
import requests
2020-09-30 18:36:27 +00:00
import sqlite3
2019-12-27 10:28:25 +00:00
# HAMSTER_DIR = Path.home() / '.local/share/hamster'
# HAMSTER_FILE = HAMSTER_DIR / 'hamster.db'
HAMSTER_FILE = 'hamster-testing.db'
2019-12-27 10:28:25 +00:00
conn = sqlite3.connect(HAMSTER_FILE)
c = conn.cursor()
def get_categories(ids=None, search=None):
sql = '''
SELECT
2019-12-27 10:28:25 +00:00
id, name
FROM
2019-12-27 10:28:25 +00:00
categories '''
args = []
if ids is not None:
sql = sql + 'WHERE id IN ({seq})'.format(
seq=','.join(['?'] * len(ids))
)
args = args + list(ids)
if search is not None:
sql = sql + " WHERE name LIKE ?"
search = '%{0}%'.format(search)
args.append(search)
results = c.execute(sql, args)
results = c.fetchall()
return results
2020-09-30 18:36:27 +00:00
def get_activities(ids=None, search=None, category_search=None):
2019-12-27 10:28:25 +00:00
sql = '''
SELECT
2019-12-27 10:28:25 +00:00
activities.id, activities.name, categories.name, categories.id
FROM
2019-12-27 10:28:25 +00:00
activities
LEFT JOIN
categories
ON
activities.category_id = categories.id '''
args = []
if ids is not None:
sql = sql + 'WHERE activities.id IN ({seq})'.format(
seq=','.join(['?'] * len(ids))
)
args = args + list(ids)
if search is not None:
sql = sql + " WHERE activities.name LIKE ?"
search = '%{0}%'.format(search)
args.append(search)
2020-09-30 18:36:27 +00:00
if category_search is not None:
sql = sql + " WHERE categories.name LIKE ?"
category_search = '%{0}%'.format(category_search)
args.append(category_search)
2019-12-27 10:28:25 +00:00
results = c.execute(sql, args)
results = c.fetchall()
return results
@click.group()
def cli():
pass
@cli.group()
def categories():
pass
2019-12-27 10:28:25 +00:00
@categories.command('list')
@click.option('--search', help='Search string')
def list_categories(search):
""" List / search categories """
2019-12-27 10:28:25 +00:00
results = get_categories(search=search)
for r in results:
2022-11-12 00:43:30 +00:00
click.echo('@{0[0]}: {0[1]}'.format(r))
2019-12-27 10:28:25 +00:00
@categories.command('delete')
2019-12-27 10:28:25 +00:00
@click.argument('ids', nargs=-1)
def delete_categories(ids):
""" Delete categories specified by IDS """
2019-12-27 10:28:25 +00:00
click.secho('Deleting:', fg='red')
results = get_categories(ids)
for r in results:
2022-11-12 01:18:23 +00:00
sql = 'select count(id) from activities where category_id = ?'
2019-12-27 10:28:25 +00:00
count = c.execute(sql, (r[0],)).fetchone()[0]
click.echo('@{0[0]}: {0[1]} ({1} activities)'.format(r, count))
click.confirm('Do you want to continue?', abort=True)
2021-09-07 21:53:36 +00:00
for r in results:
sql = 'DELETE FROM activities WHERE category_id = ?'
c.execute(sql, (r[0],))
2019-12-27 10:28:25 +00:00
sql = 'DELETE FROM categories '
sql = sql + 'WHERE id IN ({seq})'.format(
seq=','.join(['?'] * len(ids))
)
c.execute(sql, ids)
conn.commit()
click.secho('Deleted {0} categories'.format(len(ids)), fg='green')
2021-09-07 21:53:36 +00:00
@categories.command('rename')
2022-11-12 00:43:30 +00:00
@click.argument('id_', metavar='ID')
@click.argument('name')
2021-09-07 21:53:36 +00:00
def rename_category(id_, name):
""" Rename a category """
r = get_categories((id_,))[0]
click.echo('Renaming @{0[0]}: {0[1]} to "{1}"'.format(r, name))
sql = 'UPDATE categories SET name = ? WHERE id = ?'
c.execute(sql, (name, r[0]))
conn.commit()
2022-11-12 00:42:24 +00:00
2019-12-27 10:28:25 +00:00
@categories.command('activities')
@click.argument('ids', nargs=-1)
def list_category_activities(ids):
""" Show activities for categories specified by IDS """
2019-12-27 10:28:25 +00:00
sql = '''
SELECT
2019-12-27 10:28:25 +00:00
activities.id, activities.name, categories.name
FROM
2019-12-27 10:28:25 +00:00
activities
LEFT JOIN
categories
ON
activities.category_id = categories.id
WHERE
2019-12-27 10:28:25 +00:00
categories.id IN ({seq})
'''.format(
seq=','.join(['?'] * len(ids))
)
2022-11-12 00:43:30 +00:00
results = c.execute(sql, ids)
2019-12-27 10:28:25 +00:00
for r in results:
2022-11-12 00:43:30 +00:00
click.echo('@{0[0]}: {0[2]} » {0[1]}'.format(r))
2019-12-27 10:28:25 +00:00
2022-11-12 01:18:23 +00:00
@categories.command('tidy')
def tidy_categories():
""" Remove categories with no activities """
sql = 'SELECT categories.id, categories.name FROM categories LEFT JOIN activities ON categories.id = activities.category_id WHERE activities.id IS NULL'
categories = c.execute(sql).fetchall()
click.echo('Found {0} empty categories:'.format(len(categories)))
for cat in categories:
click.echo('@{0[0]}: {0[1]}'.format(cat))
click.confirm('Do you want to continue?', abort=True)
sql = 'DELETE FROM categories '
sql = sql + 'WHERE id IN ({seq})'.format(
seq=','.join(['?'] * len(categories))
)
c.execute(sql, [cat[0] for cat in categories])
conn.commit()
2019-12-27 10:28:25 +00:00
@cli.group()
def activities():
pass
@activities.command('list')
@click.option('--search', help='Search string')
2020-12-18 15:11:39 +00:00
@click.option('--csv/--no-csv', 'csv_output', default=False, help='CSV output')
def list_activities(search, csv_output):
""" List / search activities """
2019-12-27 10:28:25 +00:00
results = get_activities(search=search)
2023-01-13 00:34:51 +00:00
results.sort(key=lambda t: (t[2], t[1]))
2020-12-18 15:11:39 +00:00
if csv_output:
csv_writer = csv.writer(sys.stdout)
2019-12-27 10:28:25 +00:00
for r in results:
2020-12-18 15:11:39 +00:00
if csv_output:
csv_writer.writerow([r[3], r[2], r[0], r[1]])
else:
click.echo('@{0[3]}: {0[2]} » {0[0]}: {0[1]}'.format(r))
2019-12-27 10:28:25 +00:00
@activities.command('delete')
2019-12-27 10:28:25 +00:00
@click.argument('ids', nargs=-1)
def delete_activities(ids):
""" Delete activities specified by IDS """
2019-12-27 10:28:25 +00:00
results = get_activities(ids)
click.secho('Deleting:', fg='red')
for r in results:
sql = "SELECT COUNT(id) FROM facts WHERE activity_id = ?"
count = c.execute(sql, (r[0],)).fetchone()[0]
click.echo('@{0[0]}: {0[2]} » {0[1]} ({1} facts)'.format(r, count))
2019-12-27 10:28:25 +00:00
click.confirm('Do you want to continue?', abort=True)
sql = 'DELETE FROM activities '
sql = sql + 'WHERE id IN ({seq})'.format(
seq=','.join(['?'] * len(ids))
)
c.execute(sql, ids)
conn.commit()
click.secho('Deleted {0} activities'.format(len(ids)), fg='green')
2020-09-21 18:17:59 +00:00
2019-12-27 10:28:25 +00:00
@activities.command()
@click.argument('category_id')
@click.argument('ids', nargs=-1)
def move(category_id, ids):
""" Move activities to another category """
2019-12-27 10:28:25 +00:00
category = get_categories((category_id,))[0]
results = get_activities(ids)
click.secho('Moving to "@{0[0]}: {0[1]}":'.format(category), fg='green')
for r in results:
2022-11-12 00:43:30 +00:00
click.secho('@{0[3]}: {0[2]} » @{0[0]}: {0[1]}'.format(r), fg='blue')
2019-12-27 10:28:25 +00:00
click.confirm('Do you want to continue?', abort=True)
sql = '''
UPDATE
activities
SET
category_id = ?
'''
sql = sql + 'WHERE id IN ({seq})'.format(
seq=','.join(['?'] * len(ids))
)
c.execute(sql, (category[0], *ids))
conn.commit()
click.secho('Moved {0} activities'.format(len(ids)), fg='green')
@activities.command()
@click.argument('ids', nargs=-1)
def list_facts(ids):
""" Show facts for activities """
results = get_activities(ids)
for r in results:
click.secho(
'@{0[0]}: {0[1]}'.format(r), fg='green'
)
sql = '''
SELECT
start_time,
activities.name
FROM
facts
LEFT JOIN
activities
ON
facts.activity_id = activities.id
WHERE
activities.id = ?
'''
results = c.execute(sql, (r[0],))
for r in results:
2022-11-12 00:43:30 +00:00
click.secho('@{0[0]}, {0[1]}'.format(r), fg='blue')
@activities.command()
@click.argument('from_id')
@click.argument('to_id')
def move_facts(from_id, to_id):
""" Move facts from one activity to another """
from_activity = get_activities((from_id,))[0]
to_activity = get_activities((to_id,))[0]
click.secho(
2020-12-18 15:12:05 +00:00
'Moving facts from "@{0[2]} » @{0[0]}: {0[1]}" to "@{1[2]} » @{1[0]}: {1[1]}"'.format(
from_activity, to_activity
), fg='green'
)
sql = '''
SELECT
start_time,
activities.name
FROM
facts
LEFT JOIN
activities
ON
facts.activity_id = activities.id
WHERE
activities.id = ?
'''
results = c.execute(sql, (from_id,))
for r in results:
2022-11-12 00:43:30 +00:00
click.secho('@{0[0]}, {0[1]}'.format(r), fg='blue')
click.confirm('Do you want to continue?', abort=True)
c.execute(
'UPDATE facts SET activity_id = ? WHERE activity_id = ?',
(to_id, from_id)
)
conn.commit()
click.secho('Moved {0} facts'.format(results.rowcount), fg='green')
2021-09-07 21:53:36 +00:00
click.confirm(
2022-11-12 00:43:30 +00:00
'Would you like to delete @{0[2]} » @{0[0]}: {0[1]}?'.format(
from_activity),
2021-09-07 21:53:36 +00:00
abort=True
)
delete_activities((from_id,))
@activities.command()
def find_duplicates():
""" Show activities which are not unique in their categories """
sql = '''
SELECT
categories.id,
categories.name,
activities.id,
activities.name,
COUNT(activities.id) c
FROM
activities
LEFT JOIN
categories
ON
activities.category_id = categories.id
GROUP BY
activities.name,
activities.category_id
HAVING c > 1
'''
results = c.execute(sql)
for r in results:
2022-11-12 00:43:30 +00:00
click.secho(
'@{0[0]}: {0[1]} » @{0[2]}: {0[3]} ({0[4]})'.format(r), fg='blue')
2020-09-30 18:36:27 +00:00
@cli.group()
def kimai():
2020-09-30 18:36:27 +00:00
pass
2023-07-08 12:13:07 +00:00
def _get_kimai_mapping_file(path, category_search=None):
2020-09-30 18:36:27 +00:00
try:
return open(path)
2020-09-30 18:36:27 +00:00
except FileNotFoundError:
click.confirm(
'Mapping file {} not found, create it?:'.format(path),
2020-09-30 18:36:27 +00:00
abort=True
)
mapping_file = open(path, 'w')
2020-09-30 18:36:27 +00:00
mapping_writer = csv.writer(mapping_file)
mapping_writer.writerow([
'FROM category',
'FROM activity',
'TO Customer',
'TO Project',
2023-01-13 00:34:51 +00:00
'TO Activity',
'TO Tag',
'TO Note'
2020-09-30 18:36:27 +00:00
])
results = get_activities(category_search=category_search)
for r in results:
mapping_writer.writerow([
r[2], r[1]
])
mapping_file.close()
return open(path)
@kimai.command()
2023-07-08 12:13:07 +00:00
@click.option('--mapping-path', help='Mapping file (default ~/.local/share/hamster/mapping.kimai.csv)', multiple=True)
@click.argument('username')
@click.argument('api_key')
@click.option('--just-errors', 'just_errors', is_flag=True, help='Only display errors')
@click.option('--ignore-activities', is_flag=True, help='Ignore missing activities')
def sync(username, api_key, just_errors, ignore_activities, mapping_path=None):
"""
Download customer / project / activity data from Kimai
"""
kimai_api_url = 'https://kimai.autonomic.zone/api'
if type(mapping_path) == tuple:
mapping_files = []
for mapping_path_item in mapping_path:
mapping_file = _get_kimai_mapping_file(mapping_path_item)
next(mapping_file)
mapping_files.append(mapping_file)
mapping_reader = csv.reader(chain(*mapping_files))
else:
if mapping_path is None:
mapping_path = HAMSTER_DIR / 'mapping.kimai.csv'
mapping_file = _get_kimai_mapping_file(mapping_path)
mapping_reader = csv.reader(mapping_file)
next(mapping_reader)
mapping_data = [
[row[2], row[3], row[4]]
for row in mapping_reader
]
mapping_file.close()
auth_headers = {
'X-AUTH-USER': username,
'X-AUTH-TOKEN': api_key
}
2022-11-12 00:43:30 +00:00
customers = requests.get(
f'{kimai_api_url}/customers?visible=3', headers=auth_headers).json()
projects = requests.get(
f'{kimai_api_url}/projects?visible=3', headers=auth_headers).json()
activities = requests.get(
f'{kimai_api_url}/activities?visible=3', headers=auth_headers).json()
found_customers = []
found_projects = []
found_activities = []
for row in mapping_data:
# Check if each mapping still exists in Kimai
2022-11-12 00:43:30 +00:00
matching_customers = list(
filter(lambda x: x['name'] == row[0], customers))
if row[0] in found_customers:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Skipping existing customer '{0}'".format(row[0]), fg='green')
else:
if len(matching_customers) > 1:
2022-11-12 00:43:30 +00:00
click.secho(
"More than one match for customer '{0}'".format(row[0]), fg='red')
continue
elif len(matching_customers) < 1:
2022-11-12 00:43:30 +00:00
click.secho("Missing customer '{0}'".format(
row[0]), fg='yellow')
continue
else:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Found customer '{0}'".format(row[0]), fg='green')
found_customers.append(row[0])
project_str = ':'.join(row[0:2])
matching_projects = list(filter(
lambda x: x['name'] == row[1] and
2022-11-12 00:43:30 +00:00
x['customer'] == matching_customers[0]['id'],
projects)
)
if project_str in found_projects:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Skipping existing project '{0}'".format(project_str), fg='green')
else:
if len(matching_projects) > 1:
2022-11-12 00:43:30 +00:00
click.secho("More than one match for project '{0}'".format(
project_str), fg='red')
continue
elif len(matching_projects) < 1:
2022-11-12 00:43:30 +00:00
click.secho("Missing project '{0}'".format(
project_str), fg='yellow')
continue
else:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Found project '{0}'".format(project_str), fg='green')
found_projects.append(project_str)
if ignore_activities:
continue
activity_str = ':'.join(row)
if activity_str in found_activities:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Skipping existing activity '{0}'".format(activity_str), fg='green')
else:
matching_activities = list(filter(
2022-11-12 00:43:30 +00:00
lambda x: x['name'] == row[2]
and x['project'] == matching_projects[0]['id'],
activities
))
if len(matching_activities) > 1:
2022-11-12 00:43:30 +00:00
click.secho("More than one match for activity '{0}'".format(
activity_str), fg='red')
elif len(matching_activities) < 1:
2022-11-12 00:43:30 +00:00
click.secho("Missing activity '{0}'".format(
activity_str), fg='yellow')
else:
2022-11-12 00:43:30 +00:00
just_errors or click.secho(
"Found activity '{0}'".format(activity_str), fg='green')
found_activities.append(activity_str)
2021-11-29 13:24:51 +00:00
@kimai.command('import')
2023-07-08 12:13:07 +00:00
@click.option('--mapping-path', help='Mapping file (default ~/.local/share/hamster/mapping.kimai.csv)', multiple=True)
@click.option('--output', help='Output file (default kimai.csv)')
@click.option('--category-search', help='Category search string')
@click.option('--after', help='Only show time entries after this date')
@click.option('--show-missing', help='Just report on the missing entries', is_flag=True)
@click.argument('username')
def _import(username, mapping_path=None, output=None, category_search=None, after=None, show_missing=False):
"""
Export time tracking data in Kimai format
"""
if mapping_path is None:
mapping_path = HAMSTER_DIR / 'mapping.kimai.csv'
if output is None:
timestamp = datetime.now().strftime('%F')
output = f'kimai_{timestamp}.csv'
2023-07-08 12:13:07 +00:00
if type(mapping_path) == tuple:
mapping_files = []
for mapping_path_item in mapping_path:
mapping_file = _get_kimai_mapping_file(mapping_path_item, category_search)
next(mapping_file)
mapping_files.append(mapping_file)
mapping_reader = csv.reader(chain(*mapping_files))
else:
mapping_file = _get_kimai_mapping_file(mapping_path, category_search)
next(mapping_file)
mapping_reader = csv.reader(mapping_file)
2020-09-30 18:36:27 +00:00
mapping = {
2021-11-29 13:24:51 +00:00
'{0}:{1}'.format(row[0], row[1]): [row[2], row[3], row[4], row[5]]
2020-09-30 18:36:27 +00:00
for row in mapping_reader
}
2023-07-08 12:13:07 +00:00
if type(mapping_path) == tuple:
for mapping_file in mapping_files:
mapping_file.close()
else:
mapping_file.close()
2020-09-30 18:36:27 +00:00
output_file = open(output, 'w')
output_writer = csv.writer(output_file)
args = []
sql = '''
SELECT
facts.id, facts.start_time, facts.end_time, facts.description,
activities.id, activities.name,
categories.name, categories.id
FROM
facts
LEFT JOIN
activities ON facts.activity_id = activities.id
LEFT JOIN
2020-09-30 18:36:27 +00:00
categories ON activities.category_id = categories.id '''
if category_search is not None:
sql = sql + " WHERE categories.name LIKE ?"
category_search = '%{0}%'.format(category_search)
args.append(category_search)
if after is not None:
if category_search is not None:
sql = sql + ' AND '
else:
sql = sql + ' WHERE '
sql = sql + f"DATE(facts.start_time) > DATE(?)"
args.append(after)
2020-09-30 18:36:27 +00:00
results = c.execute(sql, args)
results = c.fetchall()
if not show_missing:
output_writer.writerow([
"Date",
"From",
"To",
"Duration",
"Rate",
"User",
"Customer",
"Project",
"Activity",
"Description",
"Exported",
"Tags",
2023-07-08 12:13:07 +00:00
"HourlyRate",
"FixedRate",
"InternalRate"
])
2020-09-30 18:36:27 +00:00
for fact in results:
k = '{0}:{1}'.format(fact[6], fact[5])
try:
mapping_ = mapping[k]
except KeyError:
if show_missing:
output_writer.writerow([fact[6], fact[5]])
2022-11-12 00:43:30 +00:00
click.secho(
"Can't find mapping for '{0}', skipping".format(k), fg='yellow')
2020-09-30 18:36:27 +00:00
continue
if show_missing:
continue
2020-09-30 18:36:27 +00:00
if fact[1] is None or fact[2] is None:
click.secho("Missing duration data '{0}-{1}', skipping".format(
fact[1],
fact[2]
), fg='yellow')
continue
2023-01-13 00:34:51 +00:00
if len(mapping_) < 5:
mapping_.append(None)
2020-09-30 18:36:27 +00:00
date_start, date_end = (
datetime.strptime(fact[2].split('.')[0], '%Y-%m-%d %H:%M:%S'),
datetime.strptime(fact[1].split('.')[0], '%Y-%m-%d %H:%M:%S')
)
duration = (
date_start - date_end
).seconds / 3600
2020-09-30 18:36:27 +00:00
output_writer.writerow([
date_start.strftime('%Y-%m-%d'),
date_start.strftime('%H:%M'),
2022-11-12 00:43:30 +00:00
'', # To (time)
2020-09-30 18:36:27 +00:00
duration,
2022-11-12 00:43:30 +00:00
'', # Rate
username,
2020-09-30 18:36:27 +00:00
mapping_[0],
mapping_[1],
mapping_[2],
2023-01-13 00:34:51 +00:00
fact[3] or mapping_[4] or '',
2022-11-12 00:43:30 +00:00
'0', # Exported
2021-11-29 13:24:51 +00:00
mapping_[3],
2022-11-12 00:43:30 +00:00
'', # Hourly rate
'', # Fixed rate
2020-09-30 18:36:27 +00:00
])
output_file.close()
2023-10-27 00:13:08 +00:00
@cli.command()
def app():
from .app import HamsterToolsApp
2023-10-27 03:04:30 +00:00
#app = HamsterToolsApp(db_cursor=c, db_connection=conn)
app = HamsterToolsApp()
2023-10-27 00:13:08 +00:00
app.run()
2020-09-21 18:17:59 +00:00
@cli.command()
def hamster():
click.echo('🐹')
2019-12-27 10:28:25 +00:00
if __name__ == "__main__":
cli()