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

Importing deliveries working

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-04-03 15:20:57 +02:00
parent 1de0498c35
commit 6f1cc0e922
3 changed files with 167 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
wgmaster/migrate.py
View File

@ -1,6 +1,6 @@
#!/bin/env python3
from typing import Dict, Any, Tuple, Optional, List
from typing import Dict, Any, Tuple, Optional, List, Iterable
import argparse
import os
import re
@ -241,6 +241,14 @@ def lookup_gem_name(name: str) -> List[Tuple[int, int]]:
raise RuntimeError()
def lookup_kg_name(kgnr: int) -> str:
cur = DB_CNX.cursor()
cur.execute("SELECT name FROM AT_kg WHERE kgnr = ?", (kgnr,))
rows = cur.fetchall()
cur.close()
return rows[0][0]
def migrate_gradation(in_dir: str, out_dir: str) -> None:
global GRADATION_MAP
GRADATION_MAP = {}
@ -339,7 +347,9 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
members = csv.parse(f'{in_dir}/TMitglieder.csv')
fbs = parse_flaechenbindungen(in_dir)
with open(f'{out_dir}/member.csv', 'w+') as f_m, open(f'{out_dir}/member_billing_address.csv', 'w+') as f_mba:
with open(f'{out_dir}/member.csv', 'w+') as f_m,\
open(f'{out_dir}/member_billing_address.csv', 'w+') as f_mba,\
open(f'{out_dir}/wb_kg.csv', 'a') as f_kg:
f_m.write('mgnr;predecessor_mgnr;prefix;given_name;middle_names;family_name;suffix;'
'birthday;entry_date;exit_date;business_shares;accounting_nr;zwstid;'
'lfbis_nr;ustid;volllieferant;buchführend;funktionär;active;iban;bic;'
@ -544,6 +554,13 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if kgnr is None:
invalid(mgnr, 'KGNr.', ort)
active = False
elif kgnr not in [kg[0] for gem in GEM_MAP.values() for kg in gem]:
glnr = list(GROSSLAGE_MAP.values())[0]
print(f'New KG: {lookup_kg_name(kgnr)} ({kgnr}, GL {glnr})')
f_kg.write(csv.format_row(kgnr, glnr))
if 9999 not in GEM_MAP:
GEM_MAP[9999] = []
GEM_MAP[9999].append((kgnr, 0))
if postal_dest is None:
invalid(mgnr, 'PLZ', None)
@ -650,27 +667,76 @@ def migrate_contracts(in_dir: str, out_dir: str) -> None:
))
def fix_deliveries(deliveries: Iterable[Dict[str, Any]]) -> Iterable[Tuple[str, List[int], datetime.date]]:
dates = {}
fixed = {}
last_dates = {}
def add(lsnr: str, linr: int, date: datetime.date, unique: bool = False) -> None:
if lsnr not in fixed:
fixed[lsnr] = []
dates[lsnr] = date
elif unique:
return add(lsnr + '/2', linr, date, unique)
fixed[lsnr].append(linr)
def get_lsnr(date: datetime.date, lsnr: str) -> str:
if date.year < 2000:
return date.strftime('%y%m%d00') + lsnr[8:]
else:
return date.strftime('%Y%m%d') + lsnr[8:]
deliveries: List[Tuple[int, str, datetime.date, int, int]] = [
(d['LINR'], d['Lieferscheinnummer'], d['Datum'], d['ZNR'], d['MGNR'])
for d in deliveries
if d['Lieferscheinnummer'] and not d['Storniert']
]
lsnrs = {d[1] for d in deliveries}
for lnr, lsnr, date, zwstnr, mgnr in deliveries:
lsdate = datetime.date(int(lsnr[:4]), int(lsnr[4:6]), int(lsnr[6:8])) if not lsnr.startswith('9') \
else datetime.date(1900 + int(lsnr[:2]), int(lsnr[2:4]), int(lsnr[4:6]))
if len(lsnr) == 12:
if date != lsdate:
if date.year == lsdate.year:
lsnr_n = get_lsnr(date, lsnr)
if lsnr_n not in lsnrs:
print(f'{lsnr} -> {lsnr_n}')
lsnr = lsnr_n
else:
warning_delivery(lsnr, mgnr, 'date', date)
else:
date = datetime.date(lsdate.year, date.month, date.day)
if zwstnr not in last_dates or not date < last_dates[zwstnr]:
last_dates[zwstnr] = date
add(lsnr, lnr, date, unique=True)
else:
add(lsnr[:12], lnr, date)
return sorted([(f[0], f[1], dates[f[0]]) for f in fixed.items()],
key=lambda f: f[0] if not f[0].startswith('9') else '19' + f[0])
def migrate_deliveries(in_dir: str, out_dir: str) -> None:
modifiers = {m['ASNR']: m for m in csv.parse(f'{in_dir}/TAbschlaege.csv') if m['Bezeichnung']}
delivery_map = {}
seasons = {}
comments: Dict[str, str] = {}
branches = {}
deliveries = list(csv.parse(f'{in_dir}/TLieferungen.csv'))
delivery_dict = {d['LINR']: d for d in deliveries}
fixed = fix_deliveries(deliveries)
with open(f'{out_dir}/delivery.csv', 'w+') as f_delivery, \
open(f'{out_dir}/delivery_part.csv', 'w+') as f_part:
f_delivery.write('year;did;date;time;zwstid;lnr;lsnr;mgnr\n')
f_delivery.write('year;did;date;time;zwstid;lnr;lsnr;mgnr;comment\n')
f_part.write('year;did;dpnr;sortid;attrid;weight;kmw;qualid;hkid;kgnr;rdnr;gerebelt;manual_weighing;spl_check;'
'hand_picked;lesemaschine;temperature;acid;scale_id;weighing_id;comment\n')
for d in deliveries:
lsnr: str = d['Lieferscheinnummer']
if d['Storniert'] or lsnr is None:
if lsnr is not None:
comments[lsnr] = d['Anmerkung']
continue
date: datetime.date = d['Datum']
for lsnr, linrs, date in fixed:
if date.year not in seasons:
seasons[date.year] = {
'currency': 'EUR' if date.year >= 2001 else 'ATS',
@ -682,74 +748,78 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
s = seasons[date.year]
if date > s['end']:
s['end'] = date
snr, dpnr = 1, 1
comment: Optional[str] = d['Anmerkung']
s['nr'] += 1
snr = s['nr']
if lsnr[-1].isupper():
if lsnr[:-1] in comments:
comment = comments[lsnr[:-1]]
if lsnr[:-1] in delivery_map:
d2 = delivery_map[lsnr[:-1]]
snr = d2[1]
dpnr = 2
else:
lsnr = lsnr[:-1]
if not lsnr[-1].isupper():
s['nr'] += 1
snr = s['nr']
delivery_map[d['LINR']] = (date.year, snr)
delivery_map[lsnr] = delivery_map[d['LINR']]
lnr = int(lsnr[9:12])
f_delivery.write(csv.format_row(
date.year, snr, date, d['Uhrzeit'], BRANCH_MAP[d['ZNR']], lnr, lsnr, d['MGNR']
znr = delivery_dict[linrs[0]]['ZNR']
if znr not in branches:
branches[znr] = {}
if date not in branches[znr]:
branches[znr][date] = 0
branches[znr][date] += 1
lnr = branches[znr][date]
comments = []
for dpnr, linr in enumerate(linrs, start=1):
d = delivery_dict[linr]
delivery_map[linr] = (date.year, snr, dpnr)
oe = d['OechsleOriginal'] or d['Oechsle']
kmw = GRADATION_MAP[oe]
sortid, attrid = d['SNR'], d['SANR']
if len(sortid) != 2:
attrid = sortid[-1]
sortid = sortid[:2]
print(f'{d["SNR"]} -> {sortid}/{attrid}')
kgnr, rdnr = None, None
if d['GNR']:
gem = GEM_MAP[d['GNR']]
if len(gem) == 1:
kgnr = gem[0][0]
if d['RNR']:
rd = REED_MAP[d['RNR']]
# TODO reed nr
if kgnr is None:
m = MEMBER_MAP[d['MGNR']]
kgnr = m['default_kgnr']
if kgnr is None:
warning_delivery(lsnr, d['MGNR'], 'KGNr.', None)
elif kgnr not in [kg[0] for gem in GEM_MAP.values() for kg in gem]:
warning_delivery(lsnr, d['MGNR'], 'KGNr.', kgnr)
kgnr = None
waage = d['Waagentext']
scale_id, weighing_id = None, None
if waage:
waage = re.split(r' +', waage)
scale_id = waage[1]
weighing_id = waage[3]
comment: Optional[str] = d['Anmerkung']
acid = d['Säure']
hand, lesemaschine = None, None
if comment:
comment = comment.replace('Söure', 'Säure')
if comment.startswith('Säure'):
acid = float(comment.split(' ')[-1].replace(',', '.'))
comment = None
elif comment == 'Maschine':
hand = False
comment = None
elif comment == 'Hand':
hand = True
comment = None
if comment:
comments.append(comment)
f_part.write(csv.format_row(
date.year, snr, dpnr, sortid, attrid, int(d['Gewicht']), kmw, QUAL_MAP[d['QSNR']], HKID, kgnr, rdnr,
d['Gerebelt'] or False, d['Handwiegung'] or False, d['Spaetlese-Ueberpruefung'] or False,
hand, lesemaschine, d['Temperatur'], acid, scale_id, weighing_id, comment
))
oe = d['OechsleOriginal'] or d['Oechsle']
kmw = GRADATION_MAP[oe]
sortid, attrid = d['SNR'], d['SANR']
if len(sortid) != 2:
attrid = sortid[-1]
sortid = sortid[:2]
print(f'{d["SNR"]} -> {sortid}/{attrid}')
kgnr, rdnr = None, None
if d['GNR']:
gem = GEM_MAP[d['GNR']]
if len(gem) == 1:
kgnr = gem[0][0]
if d['RNR']:
rd = REED_MAP[d['RNR']]
# TODO reed nr
if kgnr is None:
m = MEMBER_MAP[d['MGNR']]
kgnr = m['default_kgnr']
if kgnr is None:
warning_delivery(lsnr, d['MGNR'], 'KGNr.', None)
waage = d['Waagentext']
scale_id, weighing_id = None, None
if waage:
waage = re.split(r' +', waage)
scale_id = waage[1]
weighing_id = waage[3]
acid = d['Säure']
hand, lesemaschine = None, None
if comment:
comment = comment.replace('Söure', 'Säure')
if comment.startswith('Säure'):
acid = float(comment.split(' ')[-1].replace(',', '.'))
comment = None
elif comment == 'Maschine':
hand = False
comment = None
elif comment == 'Hand':
hand = True
comment = None
f_part.write(csv.format_row(
date.year, snr, dpnr, sortid, attrid, int(d['Gewicht']), kmw, QUAL_MAP[d['QSNR']], HKID, kgnr, rdnr,
d['Gerebelt'] or False, d['Handwiegung'] or False, d['Spaetlese-Ueberpruefung'] or False,
hand, lesemaschine, d['Temperatur'], acid, scale_id, weighing_id, comment
f_delivery.write(csv.format_row(
date.year, snr, date, d['Uhrzeit'], BRANCH_MAP[d['ZNR']], lnr, lsnr, d['MGNR'],
'; '.join(comments) or None
))
with open(f'{out_dir}/delivery_part_modifier.csv', 'w+') as f_part_mod:
@ -758,7 +828,7 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
if m['LINR'] not in delivery_map:
continue
nid = delivery_map[m['LINR']]
f_part_mod.write(csv.format_row(nid[0], nid[1], 1, m['ASNR']))
f_part_mod.write(csv.format_row(nid[0], nid[1], nid[2], m['ASNR']))
with open(f'{out_dir}/season.csv', 'w+') as f_season, open(f'{out_dir}/modifier.csv', 'w+') as f_mod:
f_season.write('year;currency;precision;start_date;end_date\n')