winzer/elwig-misc
3
1
Fork 0
Code Pull Requests Activity

migrate.py: Indicate member active status in invalid/warning calls

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-09-04 19:12:46 +02:00
parent 2f63c3202a
commit e5433c53de
1 changed files with 23 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
wgmaster/migrate.py
View File

@ -146,15 +146,15 @@ def new(t: str, ids: Any, name: str, comment: str = None) -> None:
def success(mgnr: int, key: str, value) -> None: def success(mgnr: int, key: str, value) -> None:
if not QUIET: if not QUIET:
print(f'\x1B[1;32m{mgnr:>6}: {key:<12} {value}\x1B[0m', file=sys.stderr) print(f'\x1B[1;32m{mgnr:>6} : {key:<12} {value}\x1B[0m', file=sys.stderr)
def warning(mgnr: int, key: str, value) -> None: def warning(mgnr: int, key: str, value, active: bool) -> None:
print(f'\x1B[1;33m{mgnr:>6}: {key:<12} {value}\x1B[0m', file=sys.stderr) print(f'\x1B[1;33m{mgnr:>6} ({"A" if active else " "}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def invalid(mgnr: int, key: str, value) -> None: def invalid(mgnr: int, key: str, value, active: Optional[bool]) -> None:
print(f'\x1B[1;31m{mgnr:>6}: {key:<12} {value}\x1B[0m', file=sys.stderr) print(f'\x1B[1;31m{mgnr:>6} ({"A" if active else "?" if active is None else " "}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def renumber_delivery(lsnr_1: str, lsnr_2: str) -> None: def renumber_delivery(lsnr_1: str, lsnr_2: str) -> None:
@ -172,13 +172,13 @@ def invalid_delivery(lsnr: str, mgnr: int, key: str, value) -> None:
def convert(mgnr: int, key: str, old_value: str, new_value) -> None: def convert(mgnr: int, key: str, old_value: str, new_value) -> None:
if not QUIET: if not QUIET:
print(f'\x1B[1m{mgnr:>6}: {key:<12} "{old_value}" -> "{new_value}"\x1B[0m', file=sys.stderr) print(f'\x1B[1m{mgnr:>6} : {key:<12} "{old_value}" -> "{new_value}"\x1B[0m', file=sys.stderr)
def convert_name(mgnr: int, old_name: Tuple[str, str], new_name: Tuple[str, str, str, str, str], def convert_name(mgnr: int, old_name: Tuple[str, str], new_name: Tuple[str, str, str, str, str],
billing: Optional[str] = None) -> None: billing: Optional[str] = None) -> None:
if not QUIET: if not QUIET:
print(f'\x1B[1m{mgnr:>6}: ' print(f'\x1B[1m{mgnr:>6} : '
f'{" / ".join([e or "" for e in old_name])} -> ' f'{" / ".join([e or "" for e in old_name])} -> '
f'{" / ".join([e or "" for e in new_name])}' f'{" / ".join([e or "" for e in new_name])}'
f'{"(" + billing + ")" if billing else ""}\x1B[0m', file=sys.stderr) f'{"(" + billing + ")" if billing else ""}\x1B[0m', file=sys.stderr)
@ -782,7 +782,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if not given_name or not family_name: if not given_name or not family_name:
given_name = given_name or '' given_name = given_name or ''
family_name = family_name or '' family_name = family_name or ''
invalid(mgnr, 'Name', n1) invalid(mgnr, 'Name', n1, active)
bnr: Optional[str] = m['Betriebsnummer'] if m['Betriebsnummer'] != '-' else None bnr: Optional[str] = m['Betriebsnummer'] if m['Betriebsnummer'] != '-' else None
if bnr is not None: if bnr is not None:
@ -795,9 +795,9 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
bnr = bnr.split(' ')[0] bnr = bnr.split(' ')[0]
if not check_lfbis_nr(bnr): if not check_lfbis_nr(bnr):
if bnr in ('0', '1234567'): if bnr in ('0', '1234567'):
warning(mgnr, 'BetriebsNr.', bnr) warning(mgnr, 'BetriebsNr.', bnr, active)
else: else:
invalid(mgnr, 'BetriebsNr.', bnr) invalid(mgnr, 'BetriebsNr.', bnr, active)
bnr = None bnr = None
ustid_nr: Optional[str] = m['UID'] ustid_nr: Optional[str] = m['UID']
@ -806,13 +806,13 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if len(ustid_nr) == 8 and ustid_nr.isdigit(): if len(ustid_nr) == 8 and ustid_nr.isdigit():
ustid_nr = 'ATU' + ustid_nr ustid_nr = 'ATU' + ustid_nr
elif not USTID_NR_RE.fullmatch(ustid_nr): elif not USTID_NR_RE.fullmatch(ustid_nr):
invalid(mgnr, 'UID', ustid_nr) invalid(mgnr, 'UID', ustid_nr, active)
ustid_nr = None ustid_nr = None
if ustid_nr and not check_ustid_nr_at(ustid_nr): if ustid_nr and not check_ustid_nr_at(ustid_nr):
if ustid_nr == 'ATU11111111': if ustid_nr == 'ATU11111111':
warning(mgnr, 'UID', ustid_nr) warning(mgnr, 'UID', ustid_nr, active)
else: else:
invalid(mgnr, 'UID', ustid_nr) invalid(mgnr, 'UID', ustid_nr, active)
ustid_nr = None ustid_nr = None
iban: Optional[str] = m['IBAN'] iban: Optional[str] = m['IBAN']
@ -823,7 +823,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if iban is not None: if iban is not None:
iban = iban.replace(' ', '') iban = iban.replace(' ', '')
if not check_iban(iban): if not check_iban(iban):
invalid(mgnr, 'IBAN', iban) invalid(mgnr, 'IBAN', iban, active)
iban = None iban = None
if bic is not None: if bic is not None:
@ -833,7 +833,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
elif bic == 'RLNWATWMIB': elif bic == 'RLNWATWMIB':
bic = 'RLNWATWWMIB' bic = 'RLNWATWWMIB'
if not BIC_RE.fullmatch(bic): if not BIC_RE.fullmatch(bic):
invalid(mgnr, 'BIC', bic) invalid(mgnr, 'BIC', bic, active)
bic = None bic = None
if bic is not None: if bic is not None:
if len(bic) == 11 and bic.endswith('XXX'): if len(bic) == 11 and bic.endswith('XXX'):
@ -889,7 +889,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if email.isupper(): if email.isupper():
email = email.lower() email = email.lower()
if not EMAIL_RE.fullmatch(email): if not EMAIL_RE.fullmatch(email):
invalid(mgnr, 'E-Mail', m['EMail']) invalid(mgnr, 'E-Mail', m['EMail'], active)
email = None email = None
else: else:
parts = email.split('@') parts = email.split('@')
@ -909,7 +909,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
kgnr = lookup_kgnr(okz) kgnr = lookup_kgnr(okz)
active = m['Aktives Mitglied'] or False active = m['Aktives Mitglied'] or False
if kgnr is None: if kgnr is None:
invalid(mgnr, 'KGNr.', ort) invalid(mgnr, 'KGNr.', ort, active)
elif kgnr not in [kg[0] for gem in GEM_MAP.values() for kg in gem]: elif kgnr not in [kg[0] for gem in GEM_MAP.values() for kg in gem]:
glnr = guess_glnr(kgnr) glnr = guess_glnr(kgnr)
if glnr: if glnr:
@ -922,7 +922,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
kgnr = None kgnr = None
if postal_dest is None: if postal_dest is None:
invalid(mgnr, 'PLZ', None) invalid(mgnr, 'PLZ', None, active)
continue continue
pred = m['MGNR-Vorgänger'] if m['MGNR-Vorgänger'] in mgnrs else None pred = m['MGNR-Vorgänger'] if m['MGNR-Vorgänger'] in mgnrs else None
@ -1035,7 +1035,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if phone_1: if phone_1:
phone_1 = normalize_phone_nr(phone_1) phone_1 = normalize_phone_nr(phone_1)
if len(phone_1) <= 10 or phone_1[0] != '+': if len(phone_1) <= 10 or phone_1[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Telefon']) invalid(mgnr, 'Tel.Nr.', m['Telefon'], active)
else: else:
numbers.append(phone_1) numbers.append(phone_1)
if phone_1[4] == '6': if phone_1[4] == '6':
@ -1045,7 +1045,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if phone_2: if phone_2:
phone_2 = normalize_phone_nr(phone_2) phone_2 = normalize_phone_nr(phone_2)
if len(phone_2) <= 8 or phone_2[0] != '+': if len(phone_2) <= 8 or phone_2[0] != '+':
invalid(mgnr, 'Fax.Nr.', m['Telefax']) invalid(mgnr, 'Fax.Nr.', m['Telefax'], active)
else: else:
numbers.append(phone_2) numbers.append(phone_2)
if phone_2[4] == '6': if phone_2[4] == '6':
@ -1055,7 +1055,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if phone_3: if phone_3:
phone_3 = normalize_phone_nr(phone_3) phone_3 = normalize_phone_nr(phone_3)
if len(phone_3) <= 10 or phone_3[0] != '+': if len(phone_3) <= 10 or phone_3[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Mobiltelefon']) invalid(mgnr, 'Tel.Nr.', m['Mobiltelefon'], active)
elif phone_3 not in numbers: elif phone_3 not in numbers:
numbers.append(phone_3) numbers.append(phone_3)
if phone_3[4] == '6': if phone_3[4] == '6':
@ -1118,7 +1118,7 @@ def migrate_area_commitments(in_dir: str, out_dir: str) -> None:
for gst in parse_gstnrs(p, kgnr, mgnr) for gst in parse_gstnrs(p, kgnr, mgnr)
] ]
invalid(mgnr, 'GstNr.', f'{kgnr:05}-{nr_str}') invalid(mgnr, 'GstNr.', f'{kgnr:05}-{nr_str}', None)
return [] return []
def replace_nrs(m: re.Match, sep: str) -> str: def replace_nrs(m: re.Match, sep: str) -> str:
@ -1224,7 +1224,7 @@ def migrate_area_commitments(in_dir: str, out_dir: str) -> None:
comment, gstnr = None, None comment, gstnr = None, None
if parz is None or parz == '0000': if parz is None or parz == '0000':
if parz is not None: if parz is not None:
invalid(mgnr, 'GstNr.', f'{lookup_kg_name(kgnr)} {kgnr or 0:05}-{parz}') invalid(mgnr, 'GstNr.', f'{lookup_kg_name(kgnr)} {kgnr or 0:05}-{parz}', None)
gstnrs = [] gstnrs = []
gstnr = '-' gstnr = '-'
if WG == 'MATZEN' and len(gstnrs) == 0: if WG == 'MATZEN' and len(gstnrs) == 0: