class Booking:
"""Represents lines of individual booking."""
- # pylint: disable=too-few-public-methods
+ next: Optional[Self]
+ prev: Optional[Self]
def __init__(self,
id_: int,
- dat_lines: list[DatLine],
- prev_booking: Optional[Self]
+ booked_lines: list[DatLine],
+ gap_lines: Optional[list[DatLine]] = None
) -> None:
- self.intro = IntroLine(self, dat_lines[0].code)
- dat_lines[0].booking_line = self.intro
- self._transfer_lines = []
- for i, dat_line in enumerate(dat_lines[1:]):
- dat_line.booking_line = TransferLine(self, dat_line.code, i + 1)
- self._transfer_lines += [dat_line.booking_line]
+ self.next, self.prev = None, None
+ self.id_, self.booked_lines = id_, booked_lines[:]
+ self._gap_lines = gap_lines[:] if gap_lines else []
+ # parse booked_lines into Intro- and TransferLines
+ self.intro_line = IntroLine(self, self.booked_lines[0].code)
+ self._transfer_lines = [
+ TransferLine(self, b_line.code, i+1) for i, b_line
+ in enumerate(self.booked_lines[1:])]
+ self.booked_lines[0].booking_line = self.intro_line
+ for i, b_line in enumerate(self._transfer_lines):
+ self.booked_lines[i + 1].booking_line = b_line
+ # calculate .account_changes
changes = Wealth()
sink_account = None
self.account_changes: dict[str, Wealth] = {}
self.account_changes[sink_account] += changes.as_sink
elif not changes.sink_empty:
self._transfer_lines[-1].errors += ['needed sink missing']
- self.id_ = id_
- self.dat_lines = dat_lines
- self.prev = prev_booking
+
+ def recalc_prev_next(self, bookings: list[Self]) -> None:
+ """Assuming .id_ to be index in bookings, link prev + next Bookings."""
+ self.prev = bookings[self.id_ - 1] if self.id_ > 0 else None
if self.prev:
self.prev.next = self
- self.next: Optional[Self] = None
+ self.next = (bookings[self.id_ + 1] if self.id_ + 1 < len(bookings)
+ else None)
+ if self.next:
+ self.next.prev = self
+
+ @property
+ def gap_lines(self) -> list[DatLine]:
+ """Return ._gap_lines or, if empty, list of one empty DatLine."""
+ return self._gap_lines if self._gap_lines else [DatLine('')]
+
+ @gap_lines.setter
+ def gap_lines(self, gap_lines=list[DatLine]) -> None:
+ self._gap_lines = gap_lines[:]
+
+ @property
+ def gap_lines_copied(self) -> list[DatLine]:
+ """Return new DatLines generated from .raw's of .gap_lines."""
+ return [DatLine(dat_line.raw) for dat_line in self.gap_lines]
+
+ @property
+ def booked_lines_copied(self) -> list[DatLine]:
+ """Return new DatLines generated from .raw's of .booked_lines."""
+ return [DatLine(dat_line.raw) for dat_line in self.booked_lines]
+
+ @property
+ def target(self) -> str:
+ """Return main other party for transaction."""
+ return self.intro_line.target
+
+ @property
+ def date(self) -> str:
+ """Return Booking's day's date."""
+ return self.intro_line.date
def can_move(self, up: bool) -> bool:
"""Whether movement rules would allow self to move up or down."""
- if (up and ((not self.prev)
- or self.prev.intro.date != self.intro.date)):
+ if up and ((not self.prev) or self.prev.date != self.date):
return False
- if ((not up) and ((not self.next)
- or self.next.intro.date != self.intro.date)):
+ if (not up) and ((not self.next) or self.next.date != self.date):
return False
return True
@property
def is_questionable(self) -> bool:
"""Whether lines count any errors."""
- for _ in [bl for bl in [self.intro] + self._transfer_lines
+ for _ in [bl for bl in [self.intro_line] + self._transfer_lines
if bl.errors]:
return True
return False
def do_POST(self) -> None:
# pylint: disable=invalid-name,missing-function-docstring
- if self.pagename.startswith('edit_'):
- id_ = int(self.path_toks[2])
- if 'reload_file' in self.postvars.as_dict:
- self.server.load()
- elif 'save_file' in self.postvars.as_dict:
- self.server.save()
- elif self.pagename.startswith('ledger_'):
- for key in self.postvars.keys_prefixed('move_'):
- toks = key.split('_')
- id_ = int(toks[1])
- self.server.move_booking(id_, up=toks[2] == 'up')
- self.redirect(Path(self.pagename).joinpath(f'#{id_}'))
- return
- for key in self.postvars.keys_prefixed('copy_'):
- toks = key.split('_', maxsplit=2)
- id_ = int(toks[1])
- new_id = self.server.copy_booking(int(toks[1]),
- to_end=toks[2] == 'to_end')
- self.redirect(Path('/bookings').joinpath(f'{new_id}'))
- return
- elif self.pagename == 'edit_structured':
- if self.postvars.first('apply'):
+ prefix_file, prefix_ledger = 'file_', 'ledger_'
+ if (file_prefixed := self.postvars.keys_prefixed(prefix_file)):
+ getattr(self.server, file_prefixed[0][len(prefix_file):])()
+ elif (self.pagename.startswith('edit_')
+ and self.postvars.first('apply')):
+ booking = self.server.bookings[int(self.path_toks[2])]
+ new_lines = []
+ if self.pagename == 'edit_structured':
line_keys = self.postvars.keys_prefixed('line_')
lineno_to_inputs: dict[int, list[str]] = {}
for key in line_keys:
if lineno not in lineno_to_inputs:
lineno_to_inputs[lineno] = []
lineno_to_inputs[lineno] += [toks[2]]
- new_dat_lines = []
indent = ' '
for lineno, input_names in lineno_to_inputs.items():
data = ''
comment = self.postvars.first(f'line_{lineno}_comment')
for name in input_names:
- input_ = self.postvars.first(f'line_{lineno}_{name}')
+ input_ = self.postvars.first(f'line_{lineno}_{name}'
+ ).strip()
if name == 'date':
data = input_
elif name == 'target':
- data += f' {input_}'
+ data += f' {input_}'
elif name == 'error':
data = f'{indent}{input_}'
elif name == 'account':
data = f'{indent}{input_}'
elif name in {'amount', 'currency'}:
data += f' {input_}'
- new_dat_lines += [
+ new_lines += [
DatLine(f'{data} ; {comment}' if comment else data)]
- # pylint: disable=possibly-used-before-assignment
- self.server.rewrite_booking(id_, new_dat_lines)
- elif self.pagename == 'edit_raw':
- if self.postvars.first('apply'):
- new_dat_lines = [
- DatLine(line) for line
- in self.postvars.first('booking').splitlines()]
- self.server.rewrite_booking(id_, new_dat_lines)
+ else: # edit_raw
+ new_lines += [DatLine(line) for line
+ in self.postvars.first('booking').splitlines()]
+ new_id = self.server.rewrite_booking(booking.id_, new_lines)
+ self.redirect(Path('/bookings').joinpath(f'{new_id}'))
+ return
+ elif self.pagename.startswith(prefix_ledger):
+ action, id_str, dir_ =\
+ self.postvars.keys_prefixed(prefix_ledger)[0].split('_')[1:]
+ new_id = getattr(self.server, f'{action}_booking')(
+ int(id_str), dir_ == 'up' if action == 'move' else 'to_end')
+ self.redirect(Path(self.path).joinpath(f'#{new_id}'))
+ return
self.redirect(Path(self.path))
def do_GET(self) -> None:
self.redirect(
Path('/').joinpath('edit_structured').joinpath(str(id_)))
elif self.pagename == 'edit_structured':
- ctx['dat_lines'] = [dl.as_dict for dl
- in self.server.bookings[ctx['id']].dat_lines]
+ ctx['dat_lines'] = [
+ dl.as_dict for dl
+ in self.server.bookings[ctx['id']].booked_lines]
self.send_rendered(Path('edit_structured.tmpl'), ctx)
elif self.pagename == 'edit_raw':
- ctx['dat_lines'] = self.server.bookings[ctx['id']].dat_lines
+ ctx['dat_lines'] = self.server.bookings[ctx['id']].booked_lines
self.send_rendered(Path('edit_raw.tmpl'), ctx)
elif self.pagename == 'ledger_raw':
self.send_rendered(Path('ledger_raw.tmpl'),
"""Extends parent by loading .dat file into database for Handler."""
bookings: list[Booking]
dat_lines: list[DatLine]
+ initial_gap_lines: list[DatLine]
def __init__(self, path_dat: Path, *args, **kwargs) -> None:
super().__init__(PATH_TEMPLATES, (SERVER_HOST, SERVER_PORT), Handler)
self.last_save_hash = self._hash_dat_lines()
self._load_bookings()
+ def save(self) -> None:
+ """Save current state to ._path_dat."""
+ self._path_dat.write_text(
+ '\n'.join([line.raw for line in self.dat_lines]), encoding='utf8')
+ self.load()
+
def _hash_dat_lines(self) -> int:
return hash(tuple(dl.raw for dl in self.dat_lines))
return self._hash_dat_lines() != self.last_save_hash
def _load_bookings(self) -> None:
- self.bookings = []
- booking_lines: list[DatLine] = []
- last_date = ''
+ booked_lines: list[DatLine] = []
+ gap_lines: list[DatLine] = []
+ booking: Optional[Booking] = None
+ self.bookings, self.initial_gap_lines, last_date = [], [], ''
for dat_line in self.dat_lines + [DatLine('')]:
if dat_line.code:
- booking_lines += [dat_line]
- elif booking_lines:
- booking = Booking(
- len(self.bookings), booking_lines,
- self.bookings[-1] if self.bookings else None)
- if last_date > booking.intro.date:
- booking.intro.errors += ['date < previous valid date']
- else:
- last_date = booking.intro.date
- self.bookings += [booking]
- booking_lines = []
-
- def save(self) -> None:
- """Save current state to ._path_dat."""
- self._path_dat.write_text(
- '\n'.join([line.raw for line in self.dat_lines]), encoding='utf8')
- self.load()
-
- def _margin_indices(self, booking: Booking) -> tuple[int, int]:
- start_idx = self.dat_lines.index(booking.dat_lines[0])
- end_idx = self.dat_lines.index(booking.dat_lines[-1])
- return start_idx, end_idx
-
- def _replace_from_to(self,
- start_idx: int,
- end_idx: int,
- new_lines: list[DatLine]
- ) -> None:
- self.dat_lines = (self.dat_lines[:start_idx]
- + new_lines
- + self.dat_lines[end_idx+1:])
- self._load_bookings()
-
- def move_booking(self, id_: int, up: bool) -> None:
- """Move Booking of id_ one step up or downwards"""
- id_other = id_ + (-1 if up else 1)
- agent = self.bookings[id_]
- other = self.bookings[id_other]
- start_agent, end_agent = self._margin_indices(agent)
- start_other, end_other = self._margin_indices(other)
- gap_lines = self.dat_lines[(end_other if up else end_agent) + 1:
- start_agent if up else start_other]
- self._replace_from_to(start_other if up else start_agent,
- end_agent if up else end_other,
- (agent.dat_lines if up else other.dat_lines)
- + gap_lines
- + (other.dat_lines if up else agent.dat_lines))
-
- def rewrite_booking(self, id_: int, new_dat_lines: list[DatLine]) -> None:
- """Rewrite .dat_lines for Booking of .id_ with new_dat_lines."""
- self._replace_from_to(*self._margin_indices(self.bookings[id_]),
- new_dat_lines)
+ if gap_lines:
+ if booking:
+ booking.gap_lines = gap_lines[:]
+ else:
+ self.initial_gap_lines = gap_lines[:]
+ gap_lines.clear()
+ booked_lines += [dat_line]
+ else:
+ if booked_lines:
+ booking = Booking(len(self.bookings), booked_lines[:])
+ if last_date > booking.date:
+ booking.intro_line.errors += [
+ 'date < previous valid date']
+ else:
+ last_date = booking.date
+ self.bookings += [booking]
+ booked_lines.clear()
+ gap_lines += [dat_line]
+ for booking in self.bookings:
+ booking.recalc_prev_next(self.bookings)
+ if booking:
+ booking.gap_lines = gap_lines[:-1]
+
+ def _recalc_dat_lines(self) -> None:
+ self.dat_lines = self.initial_gap_lines[:]
+ for booking in self.bookings:
+ self.dat_lines += booking.booked_lines
+ self.dat_lines += booking.gap_lines
+
+ def _move_booking(self, idx_from, idx_to) -> None:
+ moving = self.bookings[idx_from]
+ if idx_from >= idx_to: # moving upward, deletion must
+ del self.bookings[idx_from] # precede insertion to keep
+ self.bookings[idx_to:idx_to] = [moving] # deletion index, downwards
+ if idx_from < idx_to: # the other way around keeps
+ del self.bookings[idx_from] # insertion index
+ min_idx, max_idx = min(idx_from, idx_to), max(idx_from, idx_to)
+ for idx, booking in enumerate(self.bookings[min_idx:max_idx + 1]):
+ booking.id_ = min_idx + idx
+ booking.recalc_prev_next(self.bookings)
+
+ def move_booking(self, old_id: int, up: bool) -> int:
+ """Move Booking of old_id one step up or downwards"""
+ new_id = old_id + (-1 if up else 1)
+ self._move_booking(old_id, # moving down implies
+ new_id + (0 if up else 1)) # jumping over next item
+ self._recalc_dat_lines()
+ return new_id
+
+ def rewrite_booking(self, old_id: int, new_booked_lines: list[DatLine]
+ ) -> int:
+ """Rewrite Booking with new_booked_lines, move if changed date."""
+ old_booking = self.bookings[old_id]
+ new_date = new_booked_lines[0].code.lstrip().split(maxsplit=1)[0]
+ if new_date == old_booking.date:
+ new_booking = Booking(
+ old_id, new_booked_lines, old_booking.gap_lines_copied)
+ self.bookings[old_id] = new_booking
+ new_booking.recalc_prev_next(self.bookings)
+ else:
+ i_booking = self.bookings[0]
+ new_idx = None
+ while i_booking.next:
+ if not i_booking.prev and i_booking.date > new_date:
+ new_idx = i_booking.id_
+ break
+ if i_booking.next.date > new_date:
+ break
+ i_booking = i_booking.next
+ if new_idx is None:
+ new_idx = i_booking.id_ + 1
+ # ensure that, if we land in group of like-dated Bookings, we land
+ # on the edge closest to our last position
+ if i_booking.date == new_date and old_id < i_booking.id_:
+ new_idx = [b for b in self.bookings
+ if b.date == new_date][0].id_
+ new_booking = Booking(
+ new_idx, new_booked_lines, old_booking.gap_lines_copied)
+ self.bookings[old_id] = new_booking
+ self._move_booking(old_id, new_idx)
+ self._recalc_dat_lines()
+ return new_booking.id_
def copy_booking(self, id_: int, to_end: bool) -> int:
"""Add copy of Booking of id_ to_end of ledger, or after copied."""
copied = self.bookings[id_]
- empty_line = DatLine('')
- if to_end or copied is self.bookings[-1]:
+ new_id = len(self.bookings) if to_end else copied.id_ + 1
+ if to_end:
+ intro_comment = copied.booked_lines[0].comment
intro = DatLine(
- f'{dt_date.today().isoformat()} '
- f'{copied.intro.target} ; {copied.dat_lines[0].comment}')
- self.dat_lines += [empty_line, intro] + copied.dat_lines[1:]
- prev_id = self.bookings[-1].id_
+ f'{dt_date.today().isoformat()} {copied.target}'
+ + ' ; {intro_comment}' if intro_comment else '')
+ new_booking = Booking(new_id,
+ [intro] + copied.booked_lines_copied[:1],
+ copied.gap_lines_copied)
+ self.bookings += [new_booking]
else:
- start = self.dat_lines.index(copied.dat_lines[-1])
- self._replace_from_to(start + 1, start,
- [empty_line] + copied.dat_lines)
- prev_id = copied.id_
- self._load_bookings()
- return prev_id + 1
+ new_booking = Booking(new_id,
+ copied.booked_lines_copied,
+ copied.gap_lines_copied)
+ self.bookings[new_id:new_id] = [new_booking]
+ for b in self.bookings[new_id + 1:]:
+ b.id_ += 1
+ new_booking.recalc_prev_next(self.bookings)
+ self._recalc_dat_lines()
+ return new_id
@property
def dat_lines_sans_empty(self) -> list[DatLine]:
projects / plomledger / commitdiff