4367730824

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/bergercookie/syncall/commit/<a class=hub.com/bergercookie/syncall/commit/6507a9fa374713693cc089c75bbfa673706a051f">6507a9fa3<a href="https://github.com/bergercookie/syncall/commit/6507a9fa374713693cc089c75bbfa673706a051f">">Merge </a><a class="double-link" href="https://github.com/bergercookie/syncall/commit/<a class="double-link" href="https://github.com/bergercookie/syncall/commit/073d9666e2da7681073027ac76650cbf54151eaa">073d9666e</a>">073d9666e</a><a href="https://github.com/bergercookie/syncall/commit/6507a9fa374713693cc089c75bbfa673706a051f"> into 520e24635">520e24635</a>

Pull Request Pull Request #78: [caldav] Add sync project in tags

Run Details

13 of 13 new or added lines in 1 file covered. (100.0%)

1546 of 2730 relevant lines covered (56.63%)

0.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.75

/syncall/tw_caldav_utils.py

import traceback
from datetime import timedelta
from typing import Dict, List, Optional, Sequence, Tuple
from uuid import UUID

import icalendar
from bubop import logger
from item_synchronizer.types import Item

aliases_tw_caldav_status = {
    "completed": "completed",
    "pending": "needs-action",
    "waiting": "needs-action",
    "deleted": "cancelled",
}

aliases_caldav_tw_status = {
    "completed": "completed",
    "needs-action": "pending",
    "in-process": "pending",
    "cancelled": "deleted",
}

aliases_tw_caldav_priority = {
    "l": 9,
    "m": 5,
    "h": 1,
}

aliases_caldav_tw_priority = {v: k for k, v in aliases_tw_caldav_priority.items()}


def convert_tw_to_caldav(tw_item: Item) -> Item:
    assert all(
        i in tw_item.keys() for i in ("description", "status", "uuid")
    ), "Missing keys in tw_item"

    caldav_item: Item = {}

    caldav_item["summary"] = tw_item["description"]
    # description
    caldav_item["description"] = "IMPORTED FROM TASKWARRIOR\n"
    if "annotations" in tw_item.keys():
        for i, annotation in enumerate(tw_item["annotations"]):
            caldav_item["description"] += f"\n* Annotation {i + 1}: {annotation}"

    caldav_item["description"] += "\n"
    caldav_item["description"] += f'\n* uuid: {tw_item["uuid"]}'

    # Status
    caldav_item["status"] = aliases_tw_caldav_status[tw_item["status"]]

    # Priority
    if "priority" in tw_item.keys():
        caldav_item["priority"] = aliases_tw_caldav_priority[tw_item["priority"].lower()]

    # Timestamps
    if "modified" in tw_item.keys():
        caldav_item["last-modified"] = tw_item["modified"]

    # Start/due dates
    # - If given due date -> (start=due-1, end=due)
    if "due" in tw_item.keys():
        caldav_item["start"] = tw_item["due"] - timedelta(hours=1)
        caldav_item["due"] = tw_item["due"]

    # Tags
    if "tags" in tw_item.keys():
        caldav_item["categories"] = tw_item["tags"]

    # Project
    if "project" in tw_item.keys():
        # If there are tags use them to prevent overwritting
        if "tags" in tw_item.keys():
            project = tw_item["tags"]
        else:
            project = []
        project.append("proj:" + tw_item["project"])
        caldav_item["categories"] = project

    # if start-ed, override the status appropriately
    if "start" in tw_item.keys():
        caldav_item["status"] = "in-process"

    return caldav_item


def convert_caldav_to_tw(caldav_item: Item) -> Item:
    # Parse the description
    annotations, uuid = _parse_caldav_item_desc(caldav_item)
    assert isinstance(annotations, list)
    assert isinstance(uuid, UUID) or uuid is None

    # tasks created directly from thunderbird do not contains a status (Marked as "Not
    # Specified" in Thunderbird). Assume they're pending
    if not caldav_item["status"]:
        caldav_item["status"] = "needs-action"

    tw_item: Item = {}

    tw_item["description"] = caldav_item.get("summary")
    tw_item["annotations"] = annotations
    if uuid is not None:
        tw_item["uuid"] = uuid

    # Status
    tw_item["status"] = aliases_caldav_tw_status[caldav_item["status"]]

    # Priority
    prio = aliases_caldav_tw_priority.get(caldav_item["priority"])
    if prio:
        tw_item["priority"] = prio

    # Timestamps
    if "last-modified" in caldav_item.keys():
        tw_item["modified"] = caldav_item["last-modified"]

    # Start/due dates
    if "due" in caldav_item.keys():
        tw_item["due"] = caldav_item["due"]

    if "categories" in caldav_item.keys():
        categories = caldav_item["categories"]
        # Search the first 'proj:tag' and use it for the project
        # Then remove it from the list
        for i in categories:
            if i.startswith("proj:"):
                tw_item["project"] = i.split(":")[1]
                categories.remove(i)
                break
        tw_item["tags"] = categories

    if caldav_item["status"] == "in-process" and "last-modified" in caldav_item:
        tw_item["start"] = caldav_item["last-modified"]

    return tw_item


def parse_vcategory(vcategory: icalendar.prop.vCategory) -> Sequence[str]:
    return [str(category) for category in vcategory.cats]


def map_ics_to_item(vtodo) -> Dict:
    """
    Utility function that extracts the relevant info from an icalendar_component into a python
    dict
    """
    todo_item = {}
    todo_item["id"] = str(vtodo.get("uid"))

    def _convert_one(name: str) -> str:
        item = vtodo.get(name)
        if item:
            return str(item)

        return ""

    for name in ["status", "priority"]:
        todo_item[name] = _convert_one(name).lower()
    for name in ["description", "summary"]:
        todo_item[name] = _convert_one(name)

    for name in ["last-modified"]:
        item = vtodo.get(name)
        if item:
            todo_item[name] = item.dt

    if vtodo.get("due"):
        todo_item["due"] = vtodo["due"].dt

    vcategories = vtodo.get("categories")
    if vcategories is not None:
        # categories might be a vCategory containing the category names (strings) or might
        # return a List[vCategory], each vCategory with a single name
        # Option 1:
        #
        # CATEGORIES:bugwarrior
        # CATEGORIES:github_working_on_it
        # CATEGORIES:programming
        # CATEGORIES:remindme
        #
        # Option 2:
        # CATEGORIES:bugwarrior,github_bug,github_help_wanted,github_tw_gcal_sync,pro
        all_categories = []
        if isinstance(vcategories, Sequence):
            for vcategory in vcategories:
                all_categories.extend(parse_vcategory(vcategory))
        else:
            all_categories = [str(category) for category in vtodo["categories"].cats]
        todo_item["categories"] = all_categories

    return todo_item


def _parse_caldav_item_desc(
    caldav_item: Item,
) -> Tuple[List[str], Optional[UUID]]:
    """
    Parse and return the necessary TW fields off a caldav Item.

    Pretty much directly copied from tw_gcal_utils, however we handle status differently, so only return annotations/uuid
    """
    annotations: List[str] = []
    uuid = None

    if "description" not in caldav_item.keys():
        return annotations, uuid

    caldav_desc = caldav_item["description"]
    # strip whitespaces, empty lines
    lines = [line.strip() for line in caldav_desc.split("\n") if line][1:]

    # annotations
    i = 0
    for i, line in enumerate(lines):
        parts = line.split(":", maxsplit=1)
        if len(parts) == 2 and parts[0].lower().startswith("* annotation"):
            annotations.append(parts[1].strip())
        else:
            break

    if i == len(lines):
        return annotations, uuid

    # Iterate through rest of lines, find only the uuid
    for line in lines[i:]:
        parts = line.split(":", maxsplit=1)
        if len(parts) == 2 and parts[0].lower().startswith("* uuid"):
            try:
                uuid = UUID(parts[1].strip())
            except ValueError as err:
                logger.error(
                    f'Invalid UUID "{err}" provided during caldav -> TW conversion,'
                    f" Using None...\n\n{traceback.format_exc()}"
                )

    return annotations, uuid

1	import traceback	1✔
2	from datetime import timedelta	1✔
3	from typing import Dict, List, Optional, Sequence, Tuple	1✔
4	from uuid import UUID	1✔
5
6	import icalendar	1✔
7	from bubop import logger	1✔
8	from item_synchronizer.types import Item	1✔
9
10	aliases_tw_caldav_status = {	1✔
11	"completed": "completed",
12	"pending": "needs-action",
13	"waiting": "needs-action",
14	"deleted": "cancelled",
15	}
16
17	aliases_caldav_tw_status = {	1✔
18	"completed": "completed",
19	"needs-action": "pending",
20	"in-process": "pending",
21	"cancelled": "deleted",
22	}
23
24	aliases_tw_caldav_priority = {	1✔
25	"l": 9,
26	"m": 5,
27	"h": 1,
28	}
29
30	aliases_caldav_tw_priority = {v: k for k, v in aliases_tw_caldav_priority.items()}	1✔
31
32
33	def convert_tw_to_caldav(tw_item: Item) -> Item:	1✔
34	assert all(	1✔
35	i in tw_item.keys() for i in ("description", "status", "uuid")
36	), "Missing keys in tw_item"
37
38	caldav_item: Item = {}	1✔
39
40	caldav_item["summary"] = tw_item["description"]	1✔
41	# description
42	caldav_item["description"] = "IMPORTED FROM TASKWARRIOR\n"	1✔
43	if "annotations" in tw_item.keys():	1✔
44	for i, annotation in enumerate(tw_item["annotations"]):	1✔
45	caldav_item["description"] += f"\n* Annotation {i + 1}: {annotation}"	1✔
46
47	caldav_item["description"] += "\n"	1✔
48	caldav_item["description"] += f'\n* uuid: {tw_item["uuid"]}'	1✔
49
50	# Status
51	caldav_item["status"] = aliases_tw_caldav_status[tw_item["status"]]	1✔
52
53	# Priority
54	if "priority" in tw_item.keys():	1✔
55	caldav_item["priority"] = aliases_tw_caldav_priority[tw_item["priority"].lower()]	×
56
57	# Timestamps
58	if "modified" in tw_item.keys():	1✔
59	caldav_item["last-modified"] = tw_item["modified"]	1✔
60
61	# Start/due dates
62	# - If given due date -> (start=due-1, end=due)
63	if "due" in tw_item.keys():	1✔
64	caldav_item["start"] = tw_item["due"] - timedelta(hours=1)	1✔
65	caldav_item["due"] = tw_item["due"]	1✔
66
67	# Tags
68	if "tags" in tw_item.keys():	1✔
69	caldav_item["categories"] = tw_item["tags"]	1✔
70
71	# Project
72	if "project" in tw_item.keys():	1✔
73	# If there are tags use them to prevent overwritting
74	if "tags" in tw_item.keys():	×
75	project = tw_item["tags"]	×
76	else:
77	project = []	×
78	project.append("proj:" + tw_item["project"])	×
79	caldav_item["categories"] = project	×
80
81	# if start-ed, override the status appropriately
82	if "start" in tw_item.keys():	1✔
83	caldav_item["status"] = "in-process"	×
84
85	return caldav_item	1✔
86
87
88	def convert_caldav_to_tw(caldav_item: Item) -> Item:	1✔
89	# Parse the description
90	annotations, uuid = _parse_caldav_item_desc(caldav_item)	1✔
91	assert isinstance(annotations, list)	1✔
92	assert isinstance(uuid, UUID) or uuid is None	1✔
93
94	# tasks created directly from thunderbird do not contains a status (Marked as "Not
95	# Specified" in Thunderbird). Assume they're pending
96	if not caldav_item["status"]:	1✔
97	caldav_item["status"] = "needs-action"	×
98
99	tw_item: Item = {}	1✔
100
101	tw_item["description"] = caldav_item.get("summary")	1✔
102	tw_item["annotations"] = annotations	1✔
103	if uuid is not None:	1✔
104	tw_item["uuid"] = uuid	1✔
105
106	# Status
107	tw_item["status"] = aliases_caldav_tw_status[caldav_item["status"]]	1✔
108
109	# Priority
110	prio = aliases_caldav_tw_priority.get(caldav_item["priority"])	1✔
111	if prio:	1✔
112	tw_item["priority"] = prio	×
113
114	# Timestamps
115	if "last-modified" in caldav_item.keys():	1✔
116	tw_item["modified"] = caldav_item["last-modified"]	1✔
117
118	# Start/due dates
119	if "due" in caldav_item.keys():	1✔
120	tw_item["due"] = caldav_item["due"]	1✔
121
122	if "categories" in caldav_item.keys():	1✔
123	categories = caldav_item["categories"]	1✔
124	# Search the first 'proj:tag' and use it for the project
125	# Then remove it from the list
126	for i in categories:	1✔
127	if i.startswith("proj:"):	1✔
128	tw_item["project"] = i.split(":")[1]	×
129	categories.remove(i)	×
130	break	×
131	tw_item["tags"] = categories	1✔
132
133	if caldav_item["status"] == "in-process" and "last-modified" in caldav_item:	1✔
134	tw_item["start"] = caldav_item["last-modified"]	×
135
136	return tw_item	1✔
137
138
139	def parse_vcategory(vcategory: icalendar.prop.vCategory) -> Sequence[str]:	1✔
140	return [str(category) for category in vcategory.cats]	×
141
142
143	def map_ics_to_item(vtodo) -> Dict:	1✔
144	"""
145	Utility function that extracts the relevant info from an icalendar_component into a python
146	dict
147	"""
148	todo_item = {}	×
149	todo_item["id"] = str(vtodo.get("uid"))	×
150
151	def _convert_one(name: str) -> str:	×
152	item = vtodo.get(name)	×
153	if item:	×
154	return str(item)	×
155
156	return ""	×
157
158	for name in ["status", "priority"]:	×
159	todo_item[name] = _convert_one(name).lower()	×
160	for name in ["description", "summary"]:	×
161	todo_item[name] = _convert_one(name)	×
162
163	for name in ["last-modified"]:	×
164	item = vtodo.get(name)	×
165	if item:	×
166	todo_item[name] = item.dt	×
167
168	if vtodo.get("due"):	×
169	todo_item["due"] = vtodo["due"].dt	×
170
171	vcategories = vtodo.get("categories")	×
172	if vcategories is not None:	×
173	# categories might be a vCategory containing the category names (strings) or might
174	# return a List[vCategory], each vCategory with a single name
175	# Option 1:
176	#
177	# CATEGORIES:bugwarrior
178	# CATEGORIES:github_working_on_it
179	# CATEGORIES:programming
180	# CATEGORIES:remindme
181	#
182	# Option 2:
183	# CATEGORIES:bugwarrior,github_bug,github_help_wanted,github_tw_gcal_sync,pro
184	all_categories = []	×
185	if isinstance(vcategories, Sequence):	×
186	for vcategory in vcategories:	×
187	all_categories.extend(parse_vcategory(vcategory))	×
188	else:
189	all_categories = [str(category) for category in vtodo["categories"].cats]	×
190	todo_item["categories"] = all_categories	×
191
192	return todo_item	×
193
194
195	def _parse_caldav_item_desc(	1✔
196	caldav_item: Item,
197	) -> Tuple[List[str], Optional[UUID]]:
198	"""
199	Parse and return the necessary TW fields off a caldav Item.
200
201	Pretty much directly copied from tw_gcal_utils, however we handle status differently, so only return annotations/uuid
202	"""
203	annotations: List[str] = []	1✔
204	uuid = None	1✔
205
206	if "description" not in caldav_item.keys():	1✔
207	return annotations, uuid	1✔
208
209	caldav_desc = caldav_item["description"]	1✔
210	# strip whitespaces, empty lines
211	lines = [line.strip() for line in caldav_desc.split("\n") if line][1:]	1✔
212
213	# annotations
214	i = 0	1✔
215	for i, line in enumerate(lines):	1✔
216	parts = line.split(":", maxsplit=1)	1✔
217	if len(parts) == 2 and parts[0].lower().startswith("* annotation"):	1✔
218	annotations.append(parts[1].strip())	1✔
219	else:
220	break	1✔
221
222	if i == len(lines):	1✔
223	return annotations, uuid	×
224
225	# Iterate through rest of lines, find only the uuid
226	for line in lines[i:]:	1✔
227	parts = line.split(":", maxsplit=1)	1✔
228	if len(parts) == 2 and parts[0].lower().startswith("* uuid"):	1✔
229	try:	1✔
230	uuid = UUID(parts[1].strip())	1✔
231	except ValueError as err:	×
232	logger.error(	×
233	f'Invalid UUID "{err}" provided during caldav -> TW conversion,'
234	f" Using None...\n\n{traceback.format_exc()}"
235	)
236
237	return annotations, uuid	1✔

bergercookie / syncall / 4367730824

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous