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

Migrate deliveries

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-04-02 15:55:10 +02:00
parent 7f908a2966
commit fef88f2f20
3 changed files with 199 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
wgmaster/migrate.py
View File

@ -7,21 +7,36 @@ import re
import sys
import sqlite3
import requests
import datetime
import csv
DB_CNX: Optional[sqlite3.Connection] = None
HKID: Optional[str] = None
USTID_RE = re.compile(r'[A-Z]{2}[A-Z0-9]{2,12}')
BIC_RE = re.compile(r'[A-Z0-9]{4}[A-Z]{2}[A-Z0-9]{2}([A-Z0-9]{3})?')
IBAN_RE = re.compile(r'[A-Z]{2}[0-9]{2}[A-Z0-9]{8,30}')
EMAIL_RE = re.compile(r'[^@\s]+@([a-z0-9_äöüß-]+\.)+[a-z]{2,}')
GRADATION_MAP: Optional[Dict[float, float]] = None
CULTIVATION_MAP: Optional[Dict[int, str]] = None
BRANCH_MAP: Optional[Dict[int, str]] = None
GEM_MAP: Optional[Dict[int, List[Tuple[int, int]]]] = None
REED_MAP: Optional[Dict[int, Tuple[int, int]]] = None
GROSSLAGE_MAP: Optional[Dict[int, int]] = None
MEMBER_MAP: Optional[Dict[int, Dict[str, Any]]] = None
QUAL_MAP: Dict[int, str] = {
0: 'WEI',
1: 'RSW',
2: 'LDW',
3: 'QUW',
4: 'KAB',
5: 'SPL',
}
STREET_NAMES: Dict[str, str] = {
'Hans-Wagnerstraße': 'Hans-Wagner-Straße',
@ -73,6 +88,14 @@ def invalid(mgnr: int, key: str, value: str) -> None:
print(f'\x1B[1;31m{mgnr:>6}: {key:<12} {value}\x1B[0m', file=sys.stderr)
def warning_delivery(lsnr: str, mgnr: int, key:str, value: str) -> None:
print(f'\x1B[1;33m{lsnr:<13} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def invalid_delivery(lsnr: str, mgnr: int, key: str, value: str) -> None:
print(f'\x1B[1;31m{lsnr:<13} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def convert(mgnr: int, key: str, old_value: str, new_value: str) -> None:
if not args.quiet:
print(f'\x1B[1m{mgnr:>6}: {key:<12} "{old_value}" -> "{new_value}"\x1B[0m', file=sys.stderr)
@ -218,6 +241,13 @@ def lookup_gem_name(name: str) -> List[Tuple[int, int]]:
raise RuntimeError()
def migrate_gradation(in_dir: str, out_dir: str) -> None:
global GRADATION_MAP
GRADATION_MAP = {}
for g in csv.parse(f'{in_dir}/TUmrechnung.csv'):
GRADATION_MAP[g['Oechsle']] = g['KW']
def migrate_branches(in_dir: str, out_dir: str) -> None:
global BRANCH_MAP
BRANCH_MAP = {}
@ -303,6 +333,9 @@ def migrate_cultivations(in_dir: str, out_dir: str) -> None:
def migrate_members(in_dir: str, out_dir: str) -> None:
global MEMBER_MAP
MEMBER_MAP = {}
members = csv.parse(f'{in_dir}/TMitglieder.csv')
fbs = parse_flaechenbindungen(in_dir)
@ -525,6 +558,9 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
phone_mobile[0] if len(phone_mobile) > 0 else None, phone_mobile[1] if len(phone_mobile) > 1 else None,
kgnr, m['Anmerkung']
))
MEMBER_MAP[mgnr] = {
'default_kgnr': kgnr
}
if billing_name:
f_mba.write(csv.format_row(mgnr, billing_name, 'AT', postal_dest, address or '-'))
@ -614,6 +650,129 @@ def migrate_contracts(in_dir: str, out_dir: str) -> None:
))
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 = {}
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_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 sorted(csv.parse(f'{in_dir}/TLieferungen.csv'), key=lambda l: f'{l["Datum"]}T{l["Uhrzeit"]}'):
lsnr: str = d['Lieferscheinnummer']
if d['Storniert'] or lsnr is None:
comments[lsnr] = d['Anmerkung']
continue
date: datetime.date = d['Datum']
if date.year not in seasons:
seasons[date.year] = {
'currency': 'EUR' if date.year >= 2001 else 'ATS',
'precision': 4,
'start': date,
'end': date,
'nr': 0,
}
s = seasons[date.year]
if date > s['end']:
s['end'] = date
snr, dpnr = 1, 1
comment: Optional[str] = d['Anmerkung']
if lsnr.endswith('A'):
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.endswith('A'):
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']
))
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 = int(waage[1])
weighing_id = int(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
))
with open(f'{out_dir}/delivery_part_modifier.csv', 'w+') as f_part_mod:
f_part_mod.write('year;did;dpnr;mnr\n')
for m in csv.parse(f'{in_dir}/TLieferungAbschlag.csv'):
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']))
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')
f_mod.write('year;mnr;name;abs;rel;standard;quick_select\n')
for y, s in seasons.items():
f_season.write(csv.format_row(y, s['currency'], s['precision'], s['start'], s['end']))
for m in modifiers.values():
f_mod.write(csv.format_row(
y, m['ASNR'], m['Bezeichnung'], m['AZAS'], m['AZASProzent'], m['Standard'], m['Schnellauswahl']
))
def migrate_payments(in_dir: str, out_dir: str) -> None:
pass # TODO migrate payments
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('in_dir', type=str,
@ -624,12 +783,17 @@ if __name__ == '__main__':
help='Be less verbose')
parser.add_argument('-d', '--database', metavar='DB', required=True,
help='The sqlite database file to look up information')
parser.add_argument('-o', '--origin', metavar='HKID', required=True,
help='The default wine origin identifier')
args = parser.parse_args()
os.makedirs(args.out_dir, exist_ok=True)
HKID = args.origin
DB_CNX = sqlite3.connect(args.database)
migrate_gradation(args.in_dir, args.out_dir)
migrate_branches(args.in_dir, args.out_dir)
migrate_grosslagen(args.in_dir, args.out_dir)
migrate_gemeinden(args.in_dir, args.out_dir)
@ -638,5 +802,7 @@ if __name__ == '__main__':
migrate_cultivations(args.in_dir, args.out_dir)
migrate_members(args.in_dir, args.out_dir)
migrate_contracts(args.in_dir, args.out_dir)
migrate_deliveries(args.in_dir, args.out_dir)
migrate_payments(args.in_dir, args.out_dir)
DB_CNX.close()