use toggl api v9; set last edit time to last entry end time; better traceback
This commit is contained in:
parent
5d3e3dd998
commit
e0eea41214
1 changed files with 51 additions and 45 deletions
|
@ -18,7 +18,9 @@ ignore it. (IGNORE_PROJECTS_WITHOUT_SHEET option)
|
|||
"""
|
||||
from collections import defaultdict
|
||||
import os
|
||||
from typing import Dict, List
|
||||
from typing import Dict, List, Tuple, Any
|
||||
import traceback
|
||||
import sys
|
||||
|
||||
import datetime
|
||||
import pygsheets
|
||||
|
@ -83,75 +85,73 @@ def append_hours(worksheet: Worksheet, data: List[List[str]]):
|
|||
|
||||
def get_toggl_entries(
|
||||
token: str, work_id: int, last_edit: datetime.datetime
|
||||
) -> Dict[str, List]:
|
||||
entries = []
|
||||
page = 1
|
||||
count = 0
|
||||
total_count = 1 # 0 < 1, so we will certainly make at least one request
|
||||
while count < total_count:
|
||||
) -> Tuple[Dict[str, List], Any]:
|
||||
last_edit = last_edit.astimezone(datetime.timezone.utc)
|
||||
|
||||
resp = requests.get(
|
||||
"https://api.track.toggl.com/reports/api/v2/details",
|
||||
params={
|
||||
"workspace_id": f"{work_id}",
|
||||
"user_agent": "autohours",
|
||||
"since": last_edit.strftime("%Y-%m-%d"),
|
||||
"page": page,
|
||||
"order_desc": "off",
|
||||
},
|
||||
f"https://api.track.toggl.com/api/v9/workspaces/{work_id}/projects",
|
||||
auth=(token, "api_token"),
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
raise ValueError(
|
||||
f"Toggl sent a non 200 response: {resp.status_code}, {resp.text}"
|
||||
f'Toggl responded with non-200 status code: {resp}, body: "{resp.text}"'
|
||||
)
|
||||
json = resp.json()
|
||||
entries.extend(json["data"])
|
||||
projects = {p["id"]: p["name"] for p in resp.json()}
|
||||
|
||||
total_count = json["total_count"]
|
||||
count += json["per_page"]
|
||||
page += 1
|
||||
entries = []
|
||||
resp = requests.get(
|
||||
f"https://api.track.toggl.com/api/v9/me/time_entries",
|
||||
params={"user_agent": "toggl2sheets", "since": int(last_edit.timestamp())},
|
||||
auth=(token, "api_token"),
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
raise ValueError(
|
||||
f'Toggl responded with non-200 status code: {resp}, body: "{resp.text}"'
|
||||
)
|
||||
entries = resp.json()
|
||||
last_entry = next(filter(lambda x: x["duration"] > 0, entries), None)
|
||||
|
||||
# construct a dict of d["project name"] = [[date, description, duration (hours), wage], ...]
|
||||
out = defaultdict(list)
|
||||
for entry in entries:
|
||||
start = datetime.datetime.fromisoformat(entry["start"])
|
||||
end = datetime.datetime.fromisoformat(entry["end"])
|
||||
if end < last_edit:
|
||||
for entry in reversed(entries):
|
||||
if entry["duration"] <= 0 or entry["server_deleted_at"] is not None:
|
||||
continue # running entries and deleted entries
|
||||
start = datetime.datetime.fromisoformat(entry["start"].replace("Z", "+00:00"))
|
||||
end = datetime.datetime.fromisoformat(entry["stop"].replace("Z", "+00:00"))
|
||||
if end <= last_edit:
|
||||
continue # get better accuracy than toggl lets us in their requests
|
||||
out[entry["project"]].append(
|
||||
out[projects[entry["project_id"]]].append(
|
||||
[
|
||||
start.strftime("%-d. %-m. %Y"),
|
||||
start.astimezone().strftime("%-d. %-m. %Y"),
|
||||
entry["description"],
|
||||
round(entry["dur"] / 1000 / 60 / 60, 2),
|
||||
round(entry["duration"] / 60 / 60, 2),
|
||||
HOURLY_WAGE,
|
||||
]
|
||||
)
|
||||
|
||||
return out
|
||||
return out, last_entry
|
||||
|
||||
|
||||
def get_or_create_last_edit(sh: Spreadsheet) -> datetime.datetime:
|
||||
try:
|
||||
worksheet: Worksheet = sh.worksheet("title", "_toggl2sheets") # type: ignore
|
||||
except pygsheets.WorksheetNotFound:
|
||||
time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7)
|
||||
worksheet = sh.add_worksheet("_toggl2sheets")
|
||||
worksheet.hidden = True
|
||||
worksheet.update_value("A1", time.isoformat())
|
||||
worksheet.cell("A2").set_text_format("bold", True).value = ( # type: ignore
|
||||
"Internal record of when toggl2sheets last "
|
||||
"updated the spreadsheet, please do not modify"
|
||||
)
|
||||
return time
|
||||
else:
|
||||
val = worksheet.get_value("A1")
|
||||
if val:
|
||||
return datetime.datetime.fromisoformat(val)
|
||||
return datetime.datetime.now() - datetime.timedelta(days=7)
|
||||
|
||||
|
||||
def update_last_edit(sh: Spreadsheet):
|
||||
def update_last_edit(sh: Spreadsheet, last_entry: Dict[str, Any]):
|
||||
worksheet: Worksheet = sh.worksheet("title", "_toggl2sheets") # type: ignore
|
||||
time = datetime.datetime.now(datetime.timezone.utc)
|
||||
worksheet.update_value("A1", time.isoformat())
|
||||
worksheet.update_value("A1", last_entry["stop"].replace("Z", "+00:00"))
|
||||
|
||||
|
||||
def main():
|
||||
|
@ -166,7 +166,7 @@ def main():
|
|||
# get last edit time
|
||||
last_edit = get_or_create_last_edit(sh)
|
||||
|
||||
entries = get_toggl_entries(TOGGL_TOKEN, TOGGL_WORKSPACE, last_edit)
|
||||
entries, last_entry = get_toggl_entries(TOGGL_TOKEN, TOGGL_WORKSPACE, last_edit)
|
||||
|
||||
for key, val in entries.items():
|
||||
try:
|
||||
|
@ -183,12 +183,18 @@ def main():
|
|||
|
||||
append_hours(worksheet, val)
|
||||
|
||||
update_last_edit(sh)
|
||||
if last_entry is not None:
|
||||
update_last_edit(sh, last_entry)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as e:
|
||||
print("ERROR:", str(e))
|
||||
stk = [
|
||||
f"{s[0].split('/')[-1]}:{s[1]}:{s[2]}"
|
||||
for s in traceback.extract_tb(sys.exc_info()[2])
|
||||
][:5]
|
||||
print(f"ERROR: func stack: {stk}")
|
||||
print(f"ERROR: {e.__class__.__name__}: {e}")
|
||||
exit(1)
|
||||
|
|
Loading…
Reference in a new issue