jrnl.DayOneJournal.DayOne.write() - Code Metrics - jrnl-org/jrnl - Measure and Improve Code Quality continuously with Scrutinizer

jrnl.DayOneJournal.DayOne.write() F
last analyzed 2020-12-19 22:18 UTC

↳ Parent: jrnl.DayOneJournal

Complexity

Conditions

Size

Total Lines	63
Code Lines	48

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	14
eloc	48
nop	1
dl	0
loc	63
rs	3.6
c	0
b	0
f	0

How to fix Long Method Complexity

#!/usr/bin/env python

from datetime import datetime
import fnmatch
import os
from pathlib import Path
import platform
import plistlib
import re
import socket
import time
import uuid
from xml.parsers.expat import ExpatError

import pytz
import tzlocal

from . import Entry
from . import Journal
from . import __title__
from . import __version__


class DayOne(Journal.Journal):
    """A special Journal handling DayOne files"""

    # InvalidFileException was added to plistlib in Python3.4
    PLIST_EXCEPTIONS = (
        (ExpatError, plistlib.InvalidFileException)
        if hasattr(plistlib, "InvalidFileException")
        else ExpatError
    )

    def __init__(self, **kwargs):
        self.entries = []
        self._deleted_entries = []
        super().__init__(**kwargs)

    def open(self):
        filenames = [
            os.path.join(self.config["journal"], "entries", f)
            for f in os.listdir(os.path.join(self.config["journal"], "entries"))
        ]
        filenames = []
        for root, dirnames, f in os.walk(self.config["journal"]):
            for filename in fnmatch.filter(f, "*.doentry"):
                filenames.append(os.path.join(root, filename))
        self.entries = []
        for filename in filenames:
            with open(filename, "rb") as plist_entry:
                try:
                    dict_entry = plistlib.load(plist_entry, fmt=plistlib.FMT_XML)
                except self.PLIST_EXCEPTIONS:
                    pass
                else:
                    try:
                        timezone = pytz.timezone(dict_entry["Time Zone"])
                    except (KeyError, pytz.exceptions.UnknownTimeZoneError):
                        timezone = tzlocal.get_localzone()
                    date = dict_entry["Creation Date"]
                    # convert the date to UTC rather than keep messing with
                    # timezones
                    if timezone.zone != "UTC":
                        date = date + timezone.utcoffset(date, is_dst=False)

                    entry = Entry.Entry(
                        self,
                        date,
                        text=dict_entry["Entry Text"],
                        starred=dict_entry["Starred"],
                    )
                    entry.uuid = dict_entry["UUID"]
                    entry._tags = [
                        self.config["tagsymbols"][0] + tag.lower()
                        for tag in dict_entry.get("Tags", [])
                    ]

                    """Extended DayOne attributes"""
                    try:
                        entry.creator_device_agent = dict_entry["Creator"][
                            "Device Agent"
                        ]
                    except:
                        pass
                    try:
                        entry.creator_generation_date = dict_entry["Creator"][
                            "Generation Date"
                        ]
                    except:
                        entry.creator_generation_date = date
                    try:
                        entry.creator_host_name = dict_entry["Creator"]["Host Name"]
                    except:
                        pass
                    try:
                        entry.creator_os_agent = dict_entry["Creator"]["OS Agent"]
                    except:
                        pass
                    try:
                        entry.creator_software_agent = dict_entry["Creator"][
                            "Software Agent"
                        ]
                    except:
                        pass
                    try:
                        entry.location = dict_entry["Location"]
                    except:
                        pass
                    try:
                        entry.weather = dict_entry["Weather"]
                    except:
                        pass
                    self.entries.append(entry)
        self.sort()
        return self

    def write(self):
        """Writes only the entries that have been modified into plist files."""
        for entry in self.entries:
            if entry.modified:
                utc_time = datetime.utcfromtimestamp(
                    time.mktime(entry.date.timetuple())
                )

                if not hasattr(entry, "uuid"):
                    entry.uuid = uuid.uuid1().hex
                if not hasattr(entry, "creator_device_agent"):
                    entry.creator_device_agent = ""  # iPhone/iPhone5,3
                if not hasattr(entry, "creator_generation_date"):
                    entry.creator_generation_date = utc_time
                if not hasattr(entry, "creator_host_name"):
                    entry.creator_host_name = socket.gethostname()
                if not hasattr(entry, "creator_os_agent"):
                    entry.creator_os_agent = "{}/{}".format(
                        platform.system(), platform.release()
                    )
                if not hasattr(entry, "creator_software_agent"):
                    entry.creator_software_agent = "{}/{}".format(
                        __title__, __version__
                    )

                fn = (
                    Path(self.config["journal"])
                    / "entries"
                    / (entry.uuid.upper() + ".doentry")
                )

                entry_plist = {
                    "Creation Date": utc_time,
                    "Starred": entry.starred if hasattr(entry, "starred") else False,
                    "Entry Text": entry.title + "\n" + entry.body,
                    "Time Zone": str(tzlocal.get_localzone()),
                    "UUID": entry.uuid.upper(),
                    "Tags": [
                        tag.strip(self.config["tagsymbols"]).replace("_", " ")
                        for tag in entry.tags
                    ],
                    "Creator": {
                        "Device Agent": entry.creator_device_agent,
                        "Generation Date": entry.creator_generation_date,
                        "Host Name": entry.creator_host_name,
                        "OS Agent": entry.creator_os_agent,
                        "Software Agent": entry.creator_software_agent,
                    },
                }
                if hasattr(entry, "location"):
                    entry_plist["Location"] = entry.location
                if hasattr(entry, "weather"):
                    entry_plist["Weather"] = entry.weather

                # plistlib expects a binary object
                with fn.open(mode="wb") as f:
                    plistlib.dump(entry_plist, f, fmt=plistlib.FMT_XML, sort_keys=False)

        for entry in self._deleted_entries:
            filename = os.path.join(
                self.config["journal"], "entries", entry.uuid + ".doentry"
            )
            os.remove(filename)

    def editable_str(self):
        """Turns the journal into a string of entries that can be edited
        manually and later be parsed with eslf.parse_editable_str."""
        return "\n".join([f"{str(e)}\n# {e.uuid}\n" for e in self.entries])

    def _update_old_entry(self, entry, new_entry):
        for attr in ("title", "body", "date"):
            old_attr = getattr(entry, attr)
            new_attr = getattr(new_entry, attr)
            if old_attr != new_attr:
                entry.modified = True
                setattr(entry, attr, new_attr)

    def _get_and_remove_uuid_from_entry(self, entry):
        uuid_regex = "^ *?# ([a-zA-Z0-9]+) *?$"
        m = re.search(uuid_regex, entry.body, re.MULTILINE)
        entry.uuid = m.group(1) if m else None

        # remove the uuid from the body
        entry.body = re.sub(uuid_regex, "", entry.body, flags=re.MULTILINE, count=1)
        entry.body = entry.body.rstrip()

        return entry

    def parse_editable_str(self, edited):
        """Parses the output of self.editable_str and updates its entries."""
        # Method: create a new list of entries from the edited text, then match
        # UUIDs of the new entries against self.entries, updating the entries
        # if the edited entries differ, and deleting entries from self.entries
        # if they don't show up in the edited entries anymore.
        entries_from_editor = self._parse(edited)

        for entry in entries_from_editor:
            entry = self._get_and_remove_uuid_from_entry(entry)

        # Remove deleted entries
        edited_uuids = [e.uuid for e in entries_from_editor]
        self._deleted_entries = [e for e in self.entries if e.uuid not in edited_uuids]
        self.entries[:] = [e for e in self.entries if e.uuid in edited_uuids]

        for entry in entries_from_editor:
            for old_entry in self.entries:
                if entry.uuid == old_entry.uuid:
                    self._update_old_entry(old_entry, entry)
                    break


1			#!/usr/bin/env python
2
3			from datetime import datetime
4			import fnmatch
5			import os
6			from pathlib import Path
7			import platform
8			import plistlib
9			import re
10			import socket
11			import time
12			import uuid
13			from xml.parsers.expat import ExpatError
14
15			import pytz
16			import tzlocal
17
18			from . import Entry
19			from . import Journal
20			from . import __title__
21			from . import __version__
22
23
24			class DayOne(Journal.Journal):
25			"""A special Journal handling DayOne files"""
26
27			# InvalidFileException was added to plistlib in Python3.4
28			PLIST_EXCEPTIONS = (
29			(ExpatError, plistlib.InvalidFileException)
30			if hasattr(plistlib, "InvalidFileException")
31			else ExpatError
32			)
33
34			def __init__(self, **kwargs):
35			self.entries = []
36			self._deleted_entries = []
37			super().__init__(**kwargs)
38
39			def open(self):
40			filenames = [
41			os.path.join(self.config["journal"], "entries", f)
42			for f in os.listdir(os.path.join(self.config["journal"], "entries"))
43			]
44			filenames = []
45			for root, dirnames, f in os.walk(self.config["journal"]):
46			for filename in fnmatch.filter(f, "*.doentry"):
47			filenames.append(os.path.join(root, filename))
48			self.entries = []
49			for filename in filenames:
50			with open(filename, "rb") as plist_entry:
51			try:
52			dict_entry = plistlib.load(plist_entry, fmt=plistlib.FMT_XML)
53			except self.PLIST_EXCEPTIONS:
54			pass
55			else:
56			try:
57			timezone = pytz.timezone(dict_entry["Time Zone"])
58			except (KeyError, pytz.exceptions.UnknownTimeZoneError):
59			timezone = tzlocal.get_localzone()
60			date = dict_entry["Creation Date"]
61			# convert the date to UTC rather than keep messing with
62			# timezones
63			if timezone.zone != "UTC":
64			date = date + timezone.utcoffset(date, is_dst=False)
65
66			entry = Entry.Entry(
67			self,
68			date,
69			text=dict_entry["Entry Text"],
70			starred=dict_entry["Starred"],
71			)
72			entry.uuid = dict_entry["UUID"]
73			entry._tags = [
74			self.config["tagsymbols"][0] + tag.lower()
75			for tag in dict_entry.get("Tags", [])
76			]
77
78			"""Extended DayOne attributes"""
79			try:
80			entry.creator_device_agent = dict_entry["Creator"][
81			"Device Agent"
82			]
83			except:
84			pass
85			try:
86			entry.creator_generation_date = dict_entry["Creator"][
87			"Generation Date"
88			]
89			except:
90			entry.creator_generation_date = date
91			try:
92			entry.creator_host_name = dict_entry["Creator"]["Host Name"]
93			except:
94			pass
95			try:
96			entry.creator_os_agent = dict_entry["Creator"]["OS Agent"]
97			except:
98			pass
99			try:
100			entry.creator_software_agent = dict_entry["Creator"][
101			"Software Agent"
102			]
103			except:
104			pass
105			try:
106			entry.location = dict_entry["Location"]
107			except:
108			pass
109			try:
110			entry.weather = dict_entry["Weather"]
111			except:
112			pass
113			self.entries.append(entry)
114			self.sort()
115			return self
116
117			def write(self):
118			"""Writes only the entries that have been modified into plist files."""
119			for entry in self.entries:
120			if entry.modified:
121			utc_time = datetime.utcfromtimestamp(
122			time.mktime(entry.date.timetuple())
123			)
124
125			if not hasattr(entry, "uuid"):
126			entry.uuid = uuid.uuid1().hex
127			if not hasattr(entry, "creator_device_agent"):
128			entry.creator_device_agent = "" # iPhone/iPhone5,3
129			if not hasattr(entry, "creator_generation_date"):
130			entry.creator_generation_date = utc_time
131			if not hasattr(entry, "creator_host_name"):
132			entry.creator_host_name = socket.gethostname()
133			if not hasattr(entry, "creator_os_agent"):
134			entry.creator_os_agent = "{}/{}".format(
135			platform.system(), platform.release()
136			)
137			if not hasattr(entry, "creator_software_agent"):
138			entry.creator_software_agent = "{}/{}".format(
139			__title__, __version__
140			)
141
142			fn = (
143			Path(self.config["journal"])
144			/ "entries"
145			/ (entry.uuid.upper() + ".doentry")
146			)
147
148			entry_plist = {
149			"Creation Date": utc_time,
150			"Starred": entry.starred if hasattr(entry, "starred") else False,
151			"Entry Text": entry.title + "\n" + entry.body,
152			"Time Zone": str(tzlocal.get_localzone()),
153			"UUID": entry.uuid.upper(),
154			"Tags": [
155			tag.strip(self.config["tagsymbols"]).replace("_", " ")
156			for tag in entry.tags
157			],
158			"Creator": {
159			"Device Agent": entry.creator_device_agent,
160			"Generation Date": entry.creator_generation_date,
161			"Host Name": entry.creator_host_name,
162			"OS Agent": entry.creator_os_agent,
163			"Software Agent": entry.creator_software_agent,
164			},
165			}
166			if hasattr(entry, "location"):
167			entry_plist["Location"] = entry.location
168			if hasattr(entry, "weather"):
169			entry_plist["Weather"] = entry.weather
170
171			# plistlib expects a binary object
172			with fn.open(mode="wb") as f:
173			plistlib.dump(entry_plist, f, fmt=plistlib.FMT_XML, sort_keys=False)
174
175			for entry in self._deleted_entries:
176			filename = os.path.join(
177			self.config["journal"], "entries", entry.uuid + ".doentry"
178			)
179			os.remove(filename)
180
181			def editable_str(self):
182			"""Turns the journal into a string of entries that can be edited
183			manually and later be parsed with eslf.parse_editable_str."""
184			return "\n".join([f"{str(e)}\n# {e.uuid}\n" for e in self.entries])
185
186			def _update_old_entry(self, entry, new_entry):
187			for attr in ("title", "body", "date"):
188			old_attr = getattr(entry, attr)
189			new_attr = getattr(new_entry, attr)
190			if old_attr != new_attr:
191			entry.modified = True
192			setattr(entry, attr, new_attr)
193
194			def _get_and_remove_uuid_from_entry(self, entry):
195			uuid_regex = "^ ?# ([a-zA-Z0-9]+) ?$"
196			m = re.search(uuid_regex, entry.body, re.MULTILINE)
197			entry.uuid = m.group(1) if m else None
198
199			# remove the uuid from the body
200			entry.body = re.sub(uuid_regex, "", entry.body, flags=re.MULTILINE, count=1)
201			entry.body = entry.body.rstrip()
202
203			return entry
204
205			def parse_editable_str(self, edited):
206			"""Parses the output of self.editable_str and updates its entries."""
207			# Method: create a new list of entries from the edited text, then match
208			# UUIDs of the new entries against self.entries, updating the entries
209			# if the edited entries differ, and deleting entries from self.entries
210			# if they don't show up in the edited entries anymore.
211			entries_from_editor = self._parse(edited)
212
213			for entry in entries_from_editor:
214			entry = self._get_and_remove_uuid_from_entry(entry)
215
216			# Remove deleted entries
217			edited_uuids = [e.uuid for e in entries_from_editor]
218			self._deleted_entries = [e for e in self.entries if e.uuid not in edited_uuids]
219			self.entries[:] = [e for e in self.entries if e.uuid in edited_uuids]
220
221			for entry in entries_from_editor:
222			for old_entry in self.entries:
223			if entry.uuid == old_entry.uuid:
224			self._update_old_entry(old_entry, entry)
225			break
226

jrnl-org / jrnl

jrnl.DayOneJournal.DayOne.write() F last analyzed 2020-12-19 22:18 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

jrnl.DayOneJournal.DayOne.write() F
last analyzed 2020-12-19 22:18 UTC