Adler Neves
73ef86e621
All checks were successful
continuous-integration/drone/push Build is passing
148 lines
6.0 KiB
Python
148 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- encoding: utf-8 -*-
|
|
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
from typing import List, Union
|
|
|
|
from ics import Calendar, Event
|
|
|
|
from .cli import (AxisMeaning, ContentMeaning, Format, Source, into_timezone,
|
|
parseArguments)
|
|
|
|
|
|
def main_decomposed_args(origin: Source,
|
|
path: str,
|
|
format_: Format,
|
|
line: AxisMeaning,
|
|
column: AxisMeaning,
|
|
content: ContentMeaning,
|
|
year: int,
|
|
month: int,
|
|
day: int,
|
|
lang: str,
|
|
timezone_: Union[datetime.tzinfo, str],
|
|
ignore_names: List[str] = [],
|
|
) -> Calendar:
|
|
timezone = (into_timezone(timezone_)
|
|
if isinstance(timezone_, str) else timezone_)
|
|
if line == column:
|
|
raise ValueError('Line and column meanings must be different')
|
|
should_transpose = column.value == 'time'
|
|
table = format_[origin[path]].parse(should_transpose)
|
|
table.interpret_lang(
|
|
'EN,ES,PT'.split(','),
|
|
lang,
|
|
content.as_index()
|
|
)
|
|
arg_ts = (year, month, day)
|
|
table.interpret_line_labels_as_time_series(arg_ts)
|
|
events = table.get_column_sequences()
|
|
calendar = Calendar()
|
|
for ((start, end), (place, _place_id), (name, description, language)) in events:
|
|
if name.lower() not in ignore_names:
|
|
ev = json.dumps((int(start.timestamp()), int(end.timestamp()),
|
|
place, name, description, language))
|
|
evh = hashlib.md5(ev.encode('ascii')).hexdigest()
|
|
start = timezone.localize(start)
|
|
end = timezone.localize(end)
|
|
calendar_event = Event(name=name, begin=start,
|
|
end=end, location=place, uid=f'{evh}@table2ics')
|
|
if description != '':
|
|
calendar_event.description = description
|
|
if language != '':
|
|
calendar_event.classification = language
|
|
calendar.events.add(calendar_event)
|
|
return calendar
|
|
|
|
def multi_decomposed_args(
|
|
ml_merging_strategy: bool,
|
|
ml_merging_sets: List[List[str]],
|
|
ml_ignore: List[str],
|
|
ml_year: int,
|
|
ml_month: int,
|
|
ml_day: int,
|
|
ml_tzcode: str,
|
|
en_origin: Source,
|
|
en_url: str,
|
|
en_format: Format,
|
|
en_axis: AxisMeaning,
|
|
en_lang: str,
|
|
en_content: ContentMeaning,
|
|
es_origin: Source,
|
|
es_url: str,
|
|
es_format: Format,
|
|
es_axis: AxisMeaning,
|
|
es_lang: str,
|
|
es_content: ContentMeaning,
|
|
pt_origin: Source,
|
|
pt_url: str,
|
|
pt_format: Format,
|
|
pt_axis: AxisMeaning,
|
|
pt_lang: str,
|
|
pt_content: ContentMeaning,
|
|
) -> List[dict]:
|
|
timezone = (into_timezone(ml_tzcode)
|
|
if isinstance(ml_tzcode, str) else ml_tzcode)
|
|
en_table = en_format[en_origin[en_url]].parse(en_axis == AxisMeaning.place)
|
|
en_table.interpret_lang('EN,ES,PT'.split(','), en_lang, en_content.as_index())
|
|
en_table.interpret_line_labels_as_time_series((ml_year, ml_month, ml_day))
|
|
es_table = es_format[es_origin[es_url]].parse(es_axis == AxisMeaning.place)
|
|
es_table.interpret_lang('EN,ES,PT'.split(','), es_lang, es_content.as_index())
|
|
es_table.interpret_line_labels_as_time_series((ml_year, ml_month, ml_day))
|
|
pt_table = pt_format[pt_origin[pt_url]].parse(pt_axis == AxisMeaning.place)
|
|
pt_table.interpret_lang('EN,ES,PT'.split(','), pt_lang, pt_content.as_index())
|
|
pt_table.interpret_line_labels_as_time_series((ml_year, ml_month, ml_day))
|
|
en_events = en_table.get_column_sequences()
|
|
es_events = es_table.get_column_sequences()
|
|
pt_events = pt_table.get_column_sequences()
|
|
ml_events: List[dict] = list()
|
|
for i, (en_event, es_event, pt_event) in enumerate(zip(en_events, es_events, pt_events)):
|
|
print(en_event)
|
|
print(es_event)
|
|
print(pt_event)
|
|
((en_start, en_end), (en_place, en_place_id), (en_name, en_description, en_language)) = en_event
|
|
((es_start, es_end), (es_place, es_place_id), (es_name, es_description, es_language)) = es_event
|
|
((pt_start, pt_end), (pt_place, pt_place_id), (pt_name, pt_description, pt_language)) = pt_event
|
|
if en_name in ml_ignore or es_name in ml_ignore or pt_name in ml_ignore:
|
|
continue
|
|
if en_start != es_start or en_start != pt_start:
|
|
raise ValueError(f'Tables differ on event #{i+1}\'s start: {en_start=} {es_start=} {pt_start=}')
|
|
if en_end != es_end or en_end != pt_end:
|
|
raise ValueError(f'Tables differ on event #{i+1}\'s end: {en_end=} {es_end=} {pt_end=}')
|
|
if en_language != es_language or en_language != pt_language:
|
|
raise ValueError(f'Tables differ on event #{i+1}\'s language: {en_language=} {es_language=} {pt_language=}')
|
|
if en_place_id != es_place_id or en_place_id != pt_place_id:
|
|
raise ValueError(f'Tables differ on event #{i+1}\'s place id: {en_place_id=} {es_place_id=} {pt_place_id=}')
|
|
ml_events.append(dict(
|
|
start=timezone.localize(en_start).isoformat(),
|
|
end=timezone.localize(en_end).isoformat(),
|
|
place=dict(en=en_place, es=es_place, pt=pt_place, id=en_place_id),
|
|
name=dict(en=en_name, es=es_name, pt=pt_name),
|
|
description=dict(en=en_description, es=es_description, pt=pt_description),
|
|
language=en_language,
|
|
uid=f'{hashlib.md5(f"{timezone.localize(en_start).isoformat()}_{timezone.localize(en_end).isoformat()}_{en_place_id}".encode("ascii")).hexdigest()}@table2ics'
|
|
))
|
|
return ml_events
|
|
|
|
|
|
def main_namespace_args(args) -> Calendar:
|
|
return main_decomposed_args(
|
|
args.origin,
|
|
args.path,
|
|
args.format,
|
|
args.line,
|
|
args.column,
|
|
args.content,
|
|
args.year,
|
|
args.month,
|
|
args.day,
|
|
args.lang,
|
|
args.timezone,
|
|
)
|
|
|
|
|
|
def main():
|
|
print(main_namespace_args(parseArguments()))
|