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

Fix migrating GWK

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-05-29 20:08:18 +02:00
parent 9d9ca3c036
commit f50daf733e
4 changed files with 312 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
sql/91.plz-fix.sql
View File

@ -5,5 +5,7 @@ VALUES (2241, 3560, 'Schönkirchen-Reyersdorf'),
(2134, 5115, 'Staatz-Kautendorf');
UPDATE AT_ort SET name = 'Etzmannsdorf am Kamp' WHERE okz = 3938;
UPDATE AT_ort SET kgnr = 18002 WHERE okz = 3779;
UPDATE AT_ort SET kgnr = 5005 WHERE okz = 3818;
DELETE FROM AT_plz_dest WHERE (plz, okz) = (2231, 5011);

12
sql/sample.sql
View File

@ -5,14 +5,14 @@ INSERT INTO wb_gl VALUES
(3, 'Falkensteiner Hügelland');
INSERT INTO branch VALUES
('M', 'Matzen', 'AT', 224303541, 'Winzerstraße 1', '+4312345678');
('M', 'Matzen', 'AT', 224303541, 'Winzerstraße 1', '+43 2289 12345', '+43 2289 12345', NULL);
INSERT INTO wine_attribute VALUES
('B', 'BIO AT-BIO-302', 10000),
('HK', 'HK>17,5', 10000),
('K', 'Kabinett', 10000),
('S', 'Saft', 10000),
('Z', 'Sekt', 10000);
('B', 'BIO AT-BIO-302', 10000, TRUE),
('HK', 'HK>17,5', 10000, TRUE),
('K', 'Kabinett', 10000, TRUE),
('S', 'Saft', 10000, TRUE),
('Z', 'Sekt', 10000, TRUE);
INSERT INTO wine_cultivation VALUES
('N', 'Normal'),

11
sql/v01/10.create.sql
View File

@ -263,10 +263,11 @@ CREATE TABLE branch (
) STRICT;
CREATE TABLE wine_attribute (
attrid TEXT NOT NULL CHECK (attrid REGEXP '^[A-Z]+$'),
name TEXT NOT NULL,
attrid TEXT NOT NULL CHECK (attrid REGEXP '^[A-Z]+$'),
name TEXT NOT NULL,
kg_per_ha INTEGER NOT NULL DEFAULT 10000,
max_kg_per_ha INTEGER,
active INTEGER NOT NULL CHECK (active IN (TRUE, FALSE)) DEFAULT TRUE,
CONSTRAINT pk_wine_attribute PRIMARY KEY (attrid)
) STRICT;
@ -312,7 +313,7 @@ CREATE TABLE member (
email TEXT CHECK (email REGEXP '^[^@ ]+@([a-z0-9_\x2Däöüß]+\.)+[a-z]{2,}$') DEFAULT NULL,
default_kgnr INTEGER CHECK (NOT active OR default_kgnr IS NOT NULL),
default_kgnr INTEGER,
contact_postal INTEGER NOT NULL CHECK (contact_postal IN (TRUE, FALSE)) DEFAULT TRUE,
contact_email INTEGER NOT NULL CHECK (contact_email IN (TRUE, FALSE)) DEFAULT FALSE,
@ -456,7 +457,7 @@ CREATE TABLE delivery (
did INTEGER NOT NULL,
date TEXT NOT NULL CHECK (date LIKE year || '-%' AND date REGEXP '^[1-9][0-9]{3}-(0[1-9]|1[012])-(0[1-9]|[12][0-9]|3[01])$') DEFAULT CURRENT_DATE,
time TEXT NOT NULL CHECK (time REGEXP '^([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]$') DEFAULT CURRENT_TIME,
time TEXT CHECK (time REGEXP '^([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]$') DEFAULT CURRENT_TIME,
zwstid TEXT NOT NULL,
lnr INTEGER NOT NULL CHECK (lnr >= 1 AND lnr <= 999),
lsnr TEXT NOT NULL DEFAULT 'UNSET',

405
wgmaster/migrate.py
View File

@ -26,9 +26,10 @@ 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
REED_MAP: Optional[Dict[int, Tuple[int, int, str]]] = None
GROSSLAGE_MAP: Optional[Dict[int, int]] = None
MEMBER_MAP: Optional[Dict[int, Dict[str, Any]]] = None
GROSSLAGE_KG_MAP: Optional[Dict[int, int]] = None
QUAL_MAP: Dict[int, str] = {
0: 'WEI',
@ -39,44 +40,102 @@ QUAL_MAP: Dict[int, str] = {
5: 'SPL',
}
# TODO GWK streetnames
ORT_NAMES: Dict[str, Optional[str]] = {
'Pirawarth': None,
'Raggendorf': None,
'Matzen': 'Matzner',
'Matzn': None,
'Stillfried': None,
'Harras': None,
'Gänserndorf': None,
'Sulz': None,
'Brünn': None,
'Wien': None,
'Angern': None,
'Schweinbarth': None,
'Hohenruppersdorf': None,
'Grub': None,
'Auersthal': None,
'Ollersdorf': None,
'Spannberg': None,
'Ebenthal': None,
'Bockfließ': None,
'Dörfless': 'Dörfleser',
'Dörfles': None,
'Ableiding': None,
'Absberg': None,
'Eibesbrunn': None,
'Engersdorf': None,
'Enzersfeld': None,
'Großebersdorf': None,
'Hollabrunn': None,
'Korneuburg': None,
'Königsbrunn': None,
'Laa': None,
'Leopoldau': None,
'Manhartsbrunn': None,
'Mannhartsbrunn': 'Manhartsbrunner',
'Münichsthal': None,
'Pernau': None,
'Pillichsdorf': None,
'Retz': None,
'Russbach': None,
'Schleinbach': None,
'Seefeld': None,
'Seyring': None,
'Stammersdorf': None,
'Stelzendorf': None,
'Traunfeld': None,
'Tresdorf': None,
'Trumau': None,
'Wolkersdorf': None,
'Znaim': None,
'Obersdorf': None,
}
STREET_NAMES: Dict[str, str] = {
'Hans-Wagnerstraße': 'Hans-Wagner-Straße',
'J.Seitzstraße': 'Josef-Seitz-Straße',
'Kurhaus-Str.': 'Kurhausstraße',
'Kurhaus-Straße': 'Kurhausstraße',
'Pirawartherstraße': 'Pirawarther Straße',
'Raggendorferstraße': 'Raggendorfer Straße',
'Matznerstraße': 'Matzner Straße',
'Stillfriederstraße': 'Stillfrieder Straße',
'Harraserstraße': 'Harraser Straße',
'Gänserndorferstraße': 'Gänserdorfer Straße',
'Hofrat Döltlstraße': 'Hofrat-Döltl-Straße',
'Sulzerstraße': 'Sulzer Straße',
'Brünnerstraße': 'Brünner Straße',
'Flustraße': 'Flurstraße',
'Wienerstraße': 'Wiener Straße',
'St.Laurentstraße': 'St.-Laurentstraße',
'Angernerstraße': 'Angerner Straße',
'Schweinbartherstraße': 'Schweinbarther Straße',
'Hohenruppersdorferstraße': 'Hohenruppersdorfer Straße',
'Gruberhauptstraße': 'Gruber Hauptstraße',
'Josef Seitzstraße': 'Josef-Seitz-Straße',
'Auersthalerstraße': 'Auerstahler Straße',
'Ollersdorferstraße': 'Ollersdorfer Straße',
'Ritter Zoppelstraße': 'Ritter-Zoppel-Straße',
'Spannbergerstraße': 'Spannberger Straße',
'Ritter Zoppel Straße': 'Ritter-Zoppel-Straße',
'R. Virchow-Straße': 'Rudolf-Virchow-Straße',
'Ebenthalerstraße': 'Ebenthaler Straße',
'Bockfließerstraße': 'Bockfließer Straße',
'Dörfleserstraße': 'Dörfleser Straße',
'Dörflesserstraße': 'Dörfleser Straße',
'Grubere Hauptstraße': 'Gruber Hauptstraße',
'Groß Inzersdorf': 'Großinzersdorf',
'Erdpress': 'Erdpreß',
'Hochleitengasse': 'Hochleithengasse',
'Bei Der Gösselmühle': 'Bei der Gösslmühle',
'Dr. Peschlstraße': 'Dr.-Peschl-Straße',
'Dr.Peschlstraße': 'Dr.-Peschl-Straße',
'Dr. Salzbornstraße': 'Dr.-Salzborn-Straße',
'Elsa Brandström-Straße': 'Elsa-Brandström-Straße',
'Franz Ecker Siedlung': 'Franz-Ecker-Siedlung',
'Franz-Ecker Siedlung': 'Franz-Ecker-Siedlung',
'Franz Gillygasse': 'Franz-Gilly-Gasse',
'Franz V. Zülowstraße': 'Franz-von-Zülow-Straße',
'Gr. Nondorf': 'Großnondorf',
'In Der Trift': 'In der Trift',
'Johann Degengasse': 'Johann-Degen-Gasse',
'Josef Fürnkranz Siedlung': 'Josef-Fürnkranz-Siedlung',
'Kaiser Franz Josef Platz': 'Kaiser-Franz-Josef-Platz',
'Klein Haugsdorf': 'Kleinhaugsdorf',
'Leopold Leuthnerstraße': 'Leopold-Leuthner-Straße',
'Lh.-Mayer-Platz': 'Landeshauptmann-Mayer-Platz',
'Manhartsbr.Straße': 'Manhartsbrunner Straße',
'Maria Lourd Weg': 'Maria-Lourd-Weg',
'U. Weißgasse Straße': 'Untere Weißgerberstraße',
}
def new(t: str, ids: Any, name: str, comment: str = None) -> None:
print(f'\x1B[1;32mNew {t:>6}: {str(ids):>10} ({name}{", " + comment if comment else ""})\x1B[0m', file=sys.stderr)
def success(mgnr: int, key: str, value) -> None:
if not QUIET:
print(f'\x1B[1;32m{mgnr:>6}: {key:<12} {value}\x1B[0m', file=sys.stderr)
@ -92,15 +151,15 @@ def invalid(mgnr: int, key: str, value) -> None:
def renumber_delivery(lsnr_1: str, lsnr_2: str) -> None:
if not QUIET:
print(f'\x1B[1m{lsnr_1:<14} -> {lsnr_2:<14}\x1B[0m')
print(f'\x1B[1m{lsnr_1:<15} -> {lsnr_2:<15}\x1B[0m', file=sys.stderr)
def warning_delivery(lsnr: str, mgnr: int, key: str, value) -> None:
print(f'\x1B[1;33m{lsnr:<13} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
print(f'\x1B[1;33m{lsnr:<15} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def invalid_delivery(lsnr: str, mgnr: int, key: str, value) -> None:
print(f'\x1B[1;31m{lsnr:<13} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
print(f'\x1B[1;31m{lsnr:<15} ({mgnr:>6}): {key:<12} {value}\x1B[0m', file=sys.stderr)
def convert(mgnr: int, key: str, old_value: str, new_value) -> None:
@ -169,6 +228,15 @@ def normalize_phone_nr(nr: Optional[str]) -> Optional[str]:
return nr
def fix_street_name(name: str) -> str:
if name in STREET_NAMES:
return STREET_NAMES[name]
orte = [(k, v) for k, v in ORT_NAMES.items() if name.startswith(k + 'er')]
if (name.endswith('straße') or name.endswith('platz')) and len(orte) == 1:
return f'{orte[0][1] or orte[0][0] + "er"} {name[len(orte[0][0]) + 2:].title()}'.replace(' ', ' ')
return name
def get_bev_gst_size(kgnr: int, gstnr: str) -> Optional[int]:
r = requests.get(f'https://kataster.bev.gv.at/api/gst/{kgnr:05}/{gstnr}/')
if r.status_code != 200:
@ -262,11 +330,13 @@ def lookup_gem_name(name: str) -> List[Tuple[int, int]]:
elif name.lower() == 'grub':
name = 'Grub an der March'
elif WG == 'GWK':
hkid = "'WLWV', 'WIEN', 'WLWG', 'WLWA'"
hkid = "'WLWV', 'WIEN', 'WLWG'"
if name.endswith('*'):
# TODO do something with *
# TODO GWK do something with * in gemeinde
name = name[:-1].strip()
if name.lower() == 'kreuttal':
if name.lower() == 'joching':
return [(12185, 31351)]
elif name.lower() == 'kreuttal':
return [(15206, 31627), (15221, 31627), (15226, 31627)]
elif name.lower() == 'hochleithen':
return [(15219, 31622), (15223, 31622), (15202, 31622)]
@ -348,7 +418,11 @@ def lookup_kg_name(kgnr: int) -> str:
cur.execute("SELECT name FROM AT_kg WHERE kgnr = ?", (kgnr,))
rows = cur.fetchall()
cur.close()
return rows[0][0]
return rows[0][0] if len(rows) > 0 else None
def lookup_rnr_name(rnr: int) -> str:
return REED_MAP[rnr][2]
def lookup_hkid(kgnr: Optional[int], qualid: str) -> str:
@ -373,6 +447,26 @@ def lookup_hkid(kgnr: Optional[int], qualid: str) -> str:
return hkid
def guess_glnr(kgnr: int) -> Optional[int]:
cur = DB_CNX.cursor()
cur.execute("SELECT kgnr FROM AT_kg "
"WHERE gkz / 100 != 900 AND gkz / 100 = (SELECT gkz / 100 FROM AT_kg WHERE kgnr = ?)", (kgnr,))
rows0 = cur.fetchall()
cur.execute("SELECT kgnr FROM AT_kg "
"WHERE gkz / 100 != 900 AND gkz = (SELECT gkz FROM AT_kg WHERE kgnr = ?)", (kgnr,))
rows1 = cur.fetchall()
cur.close()
glnrs = list(set([GROSSLAGE_KG_MAP[k] for k, in rows0 if k in GROSSLAGE_KG_MAP]))
if len(glnrs) == 0:
return None
elif len(glnrs) == 1:
return glnrs[0]
glnrs = list(set([GROSSLAGE_KG_MAP[k] for k, in rows1 if k in GROSSLAGE_KG_MAP]))
return glnrs[0] if len(glnrs) > 0 else None
def migrate_gradation(in_dir: str, out_dir: str) -> None:
global GRADATION_MAP
GRADATION_MAP = {}
@ -406,13 +500,16 @@ def migrate_grosslagen(in_dir: str, out_dir: str) -> None:
f.header('glnr', 'name')
for gl in utils.csv_parse_dict(f'{in_dir}/TGrosslagen.csv'):
glnr += 1
if WG == 'GWK' and gl['GLNR'] == 8:
GROSSLAGE_MAP[8] = 6
continue
GROSSLAGE_MAP[gl['GLNR']] = glnr
f.row(glnr, gl['Bezeichnung'])
def migrate_gemeinden(in_dir: str, out_dir: str) -> None:
global GEM_MAP
GEM_MAP = {}
global GEM_MAP, GROSSLAGE_KG_MAP
GEM_MAP, GROSSLAGE_KG_MAP = {}, {}
inserted = set()
with utils.csv_open(f'{out_dir}/wb_kg.csv') as f:
@ -424,7 +521,9 @@ def migrate_gemeinden(in_dir: str, out_dir: str) -> None:
if kgnr in inserted:
continue
inserted.add(kgnr)
f.row(kgnr, GROSSLAGE_MAP[g['GLNR']])
glnr = GROSSLAGE_MAP[g['GLNR']]
GROSSLAGE_KG_MAP[kgnr] = glnr
f.row(kgnr, glnr)
def migrate_reeds(in_dir: str, out_dir: str) -> None:
@ -435,7 +534,7 @@ def migrate_reeds(in_dir: str, out_dir: str) -> None:
f.header('kgnr', 'rdnr', 'name')
for r in utils.csv_parse_dict(f'{in_dir}/TRiede.csv'):
name: str = r['Bezeichnung'].strip()
if name.isupper():
if name.isupper() or name.islower():
name = name.title()
try:
@ -447,21 +546,24 @@ def migrate_reeds(in_dir: str, out_dir: str) -> None:
print(f'Invalid GNR {r["GNR"]} for reed {name}')
continue
rdnr = max([n for k, n in REED_MAP.values() if k == kgnr] or [0]) + 1
REED_MAP[r['RNR']] = (kgnr, rdnr)
rdnr = max([n for k, n, _ in REED_MAP.values() if k == kgnr] or [0]) + 1
REED_MAP[r['RNR']] = (kgnr, rdnr, name)
f.row(kgnr, rdnr, name)
def migrate_attributes(in_dir: str, out_dir: str) -> None:
with utils.csv_open(f'{out_dir}/wine_attribute.csv') as f:
f.header('attrid', 'name', 'kg_per_ha')
f.header('attrid', 'name', 'max_kg_per_ha', 'active')
for a in utils.csv_parse_dict(f'{in_dir}/TSortenAttribute.csv'):
if a['SANR'] is None:
continue
f.row(a['SANR'], a['Attribut'], int(a['KgProHa']) if a['KgProHa'] is not None else None)
f.row(a['SANR'], a['Attribut'], int(a['KgProHa']) if a['KgProHa'] is not None else None, True)
if WG == 'MATZEN':
f.row('M', 'Matzen', 10000)
f.row('HU', 'Huber', 10000)
f.row('M', 'Matzen', None, False)
f.row('HU', 'Huber', None, False)
elif WG == 'GWK':
# TODO GWK attribute F?
f.row('F', '?', None, False)
def migrate_cultivations(in_dir: str, out_dir: str) -> None:
@ -503,7 +605,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
f_tel.header('mgnr', 'nr', 'type', 'number', 'comment')
for m in members:
# TODO handle * in GWK
# TODO GWK handle * in member name
mgnr: int = m['MGNR']
family_name: str = m['Nachname']
given_name: str = m['Vorname']
@ -658,20 +760,23 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
address_old = address
address = re.sub(r'([0-9]) ?([A-Z])\b', lambda a: a.group(1) + a.group(2).lower(),
re.sub(r'\s+', ' ', address).strip().title())
if address.startswith('Haus Nr.') or \
address.startswith('Nr. ') or \
address.startswith('Nr ') or \
address.isdigit():
address = ort.title() + ' ' + address.split(' ')[-1]
address = address.replace('strasse', 'straße').replace('strassse', 'straße')\
.replace('Strasse', 'Straße').replace('Str.', 'Straße')\
.replace('Strasse', 'Straße').replace('Str.', 'Straße').replace('stasse', 'straße')\
.replace('str.', 'straße').replace('ster.', 'straße').replace('g. ', 'gasse ')\
.replace('Gross', 'Groß').replace('Bockfliess', 'Bockfließ').replace('Weiss', 'Weiß')\
.replace('Preussen', 'Preußen').replace('Schloss', 'Schloß').replace('luss', 'luß')\
.replace('Haupstraße', 'Hauptstraße')
.replace('Haupstraße', 'Hauptstraße').replace('Russ', 'Ruß').replace('Ross', 'Roß')
address = re.sub('([a-z])([0-9])', lambda a: a.group(1) + ' ' + a.group(2), address)
if address.startswith('Nr. ') or address.startswith('Nr ') or address.isdigit():
address = ort.title() + ' ' + address.split(' ')[-1]
elif address.startswith('Ob. '):
if address.startswith('Ob. '):
address = address.replace('Ob. ', 'Obere ', 1)
address = address.replace(' Nr. ', ' ')
address = re.sub(r'([^0-9]+?)( [0-9])',
lambda a: STREET_NAMES.get(a.group(1), a.group(1)) + a.group(2), address)
address = re.sub(r'([^0-9]+?)( [0-9])', lambda a: fix_street_name(a.group(1)) + a.group(2), address)
address = re.sub(r'\s+', ' ', address).strip()
if address_old != address:
convert(mgnr, 'Adresse', address_old, address)
@ -702,21 +807,20 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
if kgnr is None:
invalid(mgnr, 'KGNr.', ort)
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.row(kgnr, glnr)
if 9999 not in GEM_MAP:
GEM_MAP[9999] = []
GEM_MAP[9999].append((kgnr, 0))
glnr = guess_glnr(kgnr)
if glnr:
new('KG', kgnr, lookup_kg_name(kgnr), f'GL {glnr}')
f_kg.row(kgnr, glnr)
if 9999 not in GEM_MAP:
GEM_MAP[9999] = []
GEM_MAP[9999].append((kgnr, 0))
else:
kgnr = None
if postal_dest is None:
invalid(mgnr, 'PLZ', None)
continue
if active and kgnr is None:
print(m)
raise RuntimeError('No default KgNr. set')
pred = m['MGNR-Vorgänger'] if m['MGNR-Vorgänger'] in mgnrs else None
f_m.row(
mgnr, pred, prefix, given_name, middle_names, family_name, suffix,
@ -731,36 +835,67 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
phone_3: Optional[str] = m['Mobiltelefon']
numbers = []
if phone_1:
phone_1 = normalize_phone_nr(phone_1)
if len(phone_1) <= 10 or phone_1[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Telefon'])
else:
numbers.append(phone_1)
if phone_1[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_1, None)
if WG == 'GWK':
# Telefax (phone_2) not used
numbers = {}
if phone_1:
pass # TODO GWK phone_1
if phone_3:
for nr in phone_3.split(','):
nr = nr.strip()
parts = nr.split(' ')
comment = None
if parts[-1].startswith('(') and parts[-1].endswith(')'):
nr = nr[:nr.rindex(' ')].strip()
comment = parts[-1][1:-1].strip()
elif parts[-1].isalpha():
nr = nr[:nr.rindex(' ')].strip()
comment = parts[-1].strip()
nr = normalize_phone_nr(nr)
mob = nr[4] == '6'
numbers[nr] = {'mobile': mob, 'landline': not mob, 'fax': False, 'comment': comment}
count = 0
for nr, data in numbers.items():
if data['mobile']:
count += 1
f_tel.row(mgnr, count, 'mobile', nr, data['comment'])
if data['landline']:
count += 1
f_tel.row(mgnr, count, 'landline', nr, data['comment'])
if data['fax']:
count += 1
f_tel.row(mgnr, count, 'fax', nr, data['comment'])
else:
if phone_1:
phone_1 = normalize_phone_nr(phone_1)
if len(phone_1) <= 10 or phone_1[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Telefon'])
else:
f_tel.row(mgnr, len(numbers), 'landline', phone_1, None)
if phone_2:
phone_2 = normalize_phone_nr(phone_2)
if len(phone_2) <= 8 or phone_2[0] != '+':
invalid(mgnr, 'Fax.Nr.', m['Telefax'])
else:
numbers.append(phone_2)
if phone_2[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_2, None)
numbers.append(phone_1)
if phone_1[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_1, None)
else:
f_tel.row(mgnr, len(numbers), 'landline', phone_1, None)
if phone_2:
phone_2 = normalize_phone_nr(phone_2)
if len(phone_2) <= 8 or phone_2[0] != '+':
invalid(mgnr, 'Fax.Nr.', m['Telefax'])
else:
f_tel.row(mgnr, len(numbers), 'fax', phone_2, None)
if phone_3:
phone_3 = normalize_phone_nr(phone_3)
if len(phone_3) <= 10 or phone_3[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Mobiltelefon'])
elif phone_3 not in numbers:
numbers.append(phone_3)
if phone_3[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_3, None)
else:
f_tel.row(mgnr, len(numbers), 'landline', phone_3, None)
numbers.append(phone_2)
if phone_2[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_2, None)
else:
f_tel.row(mgnr, len(numbers), 'fax', phone_2, None)
if phone_3:
phone_3 = normalize_phone_nr(phone_3)
if len(phone_3) <= 10 or phone_3[0] != '+':
invalid(mgnr, 'Tel.Nr.', m['Mobiltelefon'])
elif phone_3 not in numbers:
numbers.append(phone_3)
if phone_3[4] == '6':
f_tel.row(mgnr, len(numbers), 'mobile', phone_3, None)
else:
f_tel.row(mgnr, len(numbers), 'landline', phone_3, None)
MEMBER_MAP[mgnr] = {
'default_kgnr': kgnr
@ -862,8 +997,14 @@ def migrate_area_commitments(in_dir: str, out_dir: str) -> None:
text = re.sub(r'([0-9]+(, |$)){3,}', lambda m: replace_nrs(m, ', '), text)
return text
reeds: Dict[int, Dict[int, str]] = {k: {r: n
for rk, r, n in REED_MAP.values() if rk == k}
for k in set([k for k, _, _ in REED_MAP.values()])}
new_reeds: Dict[Tuple[int, int], int] = {}
with utils.csv_open(f'{out_dir}/area_commitment.csv') as f_fb, \
utils.csv_open(f'{out_dir}/area_commitment_attribute.csv',) as f_attr:
utils.csv_open(f'{out_dir}/area_commitment_attribute.csv') as f_attr, \
utils.csv_open(f'{out_dir}/wb_rd.csv', 'a+') as f_rd:
f_fb.header('fbnr', 'mgnr', 'sortid', 'cultid', 'area', 'kgnr', 'gstnr', 'rdnr',
'year_from', 'year_to', 'comment')
f_attr.header('fbnr', 'attrid')
@ -875,28 +1016,59 @@ def migrate_area_commitments(in_dir: str, out_dir: str) -> None:
fbnr: int = fb['FBNR']
mgnr: int = fb['MGNR']
gem = GEM_MAP[fb['GNR']]
kgnr = gem[0][0]
kgnrs = [kgnr for kgnr, gkz in gem]
rnr = fb['RNR']
rd_kgnr, rdnr, _ = REED_MAP.get(rnr, (None, None, None)) if rnr else (None, None, None)
if mgnr not in MEMBER_MAP:
continue
kgnr = None
if rd_kgnr is None:
kgnr = kgnrs[0]
elif rd_kgnr in kgnrs:
kgnr = rd_kgnr
elif (kgnrs[0], rnr) in new_reeds:
kgnr = kgnrs[0]
rdnr = new_reeds[(kgnr, rnr)]
else:
rname = lookup_rnr_name(rnr)
for k in kgnrs:
if k not in reeds:
continue
try:
pos = list(reeds[k].values()).index(rname)
r = list(reeds[k].keys())[pos]
kgnr = k
rdnr = r
new_reeds[(kgnr, rnr)] = rdnr
break
except ValueError:
continue
if kgnr is None:
kgnr = kgnrs[0]
rdnr = max([r for _, r, _ in REED_MAP.values() if k == kgnr] +
[r for (k, _), r in new_reeds.items() if k == kgnr]) + 1
f_rd.row(kgnr, rdnr, rname)
new_reeds[(kgnr, rnr)] = rdnr
new('Reed', (kgnr, rdnr), rname)
area = int(fb['Flaeche'])
if WG == 'MATZEN':
gstnrs = parse_gstnrs(parz, kgnr, fb['MGNR'])
else:
# TODO GstNrs GWK
# TODO GWK GstNrs
gstnrs = []
comment, gstnr = None, None
if parz is None or parz == '0000':
invalid(mgnr, 'GstNr.', f'{kgnr or 0:05}-{parz}')
invalid(mgnr, 'GstNr.', f'{lookup_kg_name(kgnr)} {kgnr or 0:05}-{parz}')
gstnrs = []
gstnr = '-'
if len(gstnrs) == 0:
comment = f'KG {kgnr:05}: {parz}'
comment = f'KG {kgnr or 0:05}: {parz}'
gstnr = format_gstnr(gstnrs) or gstnr or parz
if parz != gstnr.replace('+', '/'):
convert(mgnr, f'GstNr. ({fbnr})', parz, gstnr)
rdnr = REED_MAP.get(fb['RNR'], (None, None))[1] if fb['RNR'] else None
to = fb['Bis'] if fb['Bis'] and fb['Bis'] < 3000 else None
f_fb.row(fbnr, mgnr, fb['SNR'], CULTIVATION_MAP[fb['BANR'] or 1], area,
kgnr, gstnr, rdnr, fb['Von'], to, comment)
@ -931,12 +1103,18 @@ def fix_deliveries(deliveries: Iterable[Dict[str, Any]]) -> Iterable[Tuple[str,
lsnrs = {d[1] for d in deliveries}
for lnr, lsnr, date, zwstnr, mgnr in deliveries:
for lnr, lsnr, date, zwstid, mgnr in deliveries:
if len(lsnr) < 8:
continue
if lsnr.startswith('22'):
lsnr = '20' + lsnr[2:]
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]))
lsnr_zwstid = lsnr[8]
if lsnr_zwstid != zwstid and lsnr_zwstid in BRANCH_MAP.values():
zwstid = lsnr_zwstid
if len(lsnr) == 12:
if date != lsdate:
if date.year == lsdate.year:
@ -948,8 +1126,8 @@ def fix_deliveries(deliveries: Iterable[Dict[str, Any]]) -> Iterable[Tuple[str,
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
if zwstid not in last_dates or not date < last_dates[zwstid]:
last_dates[zwstid] = date
add(lsnr, lnr, date, unique=True)
else:
add(lsnr[:12], lnr, date)
@ -977,6 +1155,7 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
deliveries = list(utils.csv_parse_dict(f'{in_dir}/TLieferungen.csv'))
delivery_dict = {d['LINR']: d for d in deliveries}
fixed = fix_deliveries(deliveries)
updated_varieties = {}
with utils.csv_open(f'{out_dir}/delivery.csv') as f_delivery, \
utils.csv_open(f'{out_dir}/delivery_part.csv') as f_part, \
@ -1002,13 +1181,20 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
s['nr'] += 1
snr = s['nr']
mgnr = delivery_dict[linrs[0]]['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]
zwstid = lsnr[8]
if zwstid not in branches:
branches[zwstid] = {}
if date not in branches[zwstid]:
branches[zwstid][date] = 0
branches[zwstid][date] += 1
lnr = branches[zwstid][date]
if BRANCH_MAP[znr] != zwstid:
if zwstid not in BRANCH_MAP.values():
zwstid = BRANCH_MAP[znr]
comments = []
attributes = set()
@ -1044,7 +1230,10 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
attributes.remove('W')
if d['SNR'] != sortid:
print(f'{d["SNR"]}/{d["SANR"]} -> {sortid}/{attributes}')
line = f'{d["SNR"]}/{d["SANR"]} -> {sortid}/{",".join(list(attributes)) or None}'
if line not in updated_varieties:
updated_varieties[line] = 0
updated_varieties[line] += 1
qualid = QUAL_MAP[d['QSNR']]
kgnr, rdnr = None, None
@ -1053,14 +1242,14 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
if len(gem) == 1:
kgnr = gem[0][0]
if d['RNR']:
kgnr, rdnr = REED_MAP[d['RNR']]
kgnr, rdnr, _ = REED_MAP[d['RNR']]
if kgnr is None:
m = MEMBER_MAP[d['MGNR']]
m = MEMBER_MAP[mgnr]
kgnr = m['default_kgnr']
if kgnr is None:
warning_delivery(lsnr, d['MGNR'], 'KGNr.', None)
pass
elif kgnr not in [kg[0] for gem in GEM_MAP.values() for kg in gem]:
warning_delivery(lsnr, d['MGNR'], 'KGNr.', kgnr)
warning_delivery(lsnr, mgnr, 'KGNr.', kgnr)
kgnr = None
hkid = lookup_hkid(kgnr, qualid)
@ -1096,8 +1285,10 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
)
for attrid in attributes:
f_attr.row(date.year, snr, dpnr, attrid)
f_delivery.row(date.year, snr, date, d['Uhrzeit'], BRANCH_MAP[d['ZNR']], lnr, lsnr, d['MGNR'],
f_delivery.row(date.year, snr, date, d['Uhrzeit'], zwstid, lnr, lsnr, mgnr,
'; '.join(comments) or None)
for k, v in updated_varieties.items():
print(k + (f' ({v} times)' if v > 1 else ''))
with utils.csv_open(f'{out_dir}/delivery_part_modifier.csv') as f_part_mod:
f_part_mod.header('year', 'did', 'dpnr', 'modid')
@ -1116,7 +1307,7 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
for m in modifiers.values():
abs_v = int(m['AZAS'] * pow(10, s['precision'])) if m['AZAS'] is not None else None
f_mod.row(y, m['id'], m['Bezeichnung'], abs_v, m['AZASProzent'],
m.get('Standard', None), m['Schnellauswahl'])
m.get('Standard', False), m['Schnellauswahl'])
def migrate_payments(in_dir: str, out_dir: str) -> None: