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

Import reeds

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-02-24 18:15:05 +01:00
parent 25f5c3efcc
commit c3c8d51e68
2 changed files with 82 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
sql/v01/01.create.sql
View File

@ -217,10 +217,19 @@ CREATE TABLE wb_rd (
CREATE TABLE branch (
zwstid TEXT NOT NULL CHECK (zwstid REGEXP '^[A-Z]$'),
name TEXT NOT NULL,
zwstid TEXT NOT NULL CHECK (zwstid REGEXP '^[A-Z]$'),
name TEXT NOT NULL,
CONSTRAINT pk_branch PRIMARY KEY (zwstid)
country TEXT,
postal_dest TEXT,
address TEXT,
phone_nr TEXT,
CONSTRAINT pk_branch PRIMARY KEY (zwstid),
CONSTRAINT fk_branch_postal_dest FOREIGN KEY (country, postal_dest) REFERENCES postal_dest (country, id)
ON UPDATE CASCADE
ON DELETE RESTRICT
);
CREATE TABLE wine_attribute (

102
wgmaster/migrate.py
View File

@ -17,9 +17,13 @@ BIC_RE = re.compile('[A-Z0-9]{4}[A-Z]{2}[A-Z0-9]{2}([A-Z0-9]{3})?')
IBAN_RE = re.compile('[A-Z]{2}[0-9]{2}[A-Z0-9]{8,30}')
EMAIL_RE = re.compile('[^@\s]+@([a-z0-9_äöüß-]+\.)+[a-z]{2,}')
CULTIDS = {1: 'N', 2: 'KIP', 3: 'BIO'}
CULTIVATION_MAP: Optional[Dict[int, str]] = {1: 'N', 2: 'KIP', 3: 'BIO'}
STREET_NAMES = {
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
STREET_NAMES: Dict[str, str] = {
'Hans-Wagnerstraße': 'Hans-Wagner-Straße',
'J.Seitzstraße': 'Josef-Seitz-Straße',
'Kurhaus-Str.': 'Kurhausstraße',
@ -208,14 +212,6 @@ def get_bev_gst_size(kgnr: int, gstnr: str) -> Optional[int]:
return sum([n['fl'] for n in data['properties']['nutzungen']])
def parse_branches(in_dir: str) -> Dict[str, Dict[str, Any]]:
return {b['ZNR']: b for b in parse_csv(f'{in_dir}/TZweigstellen.csv')}
def parse_gemeinden(in_dir: str) -> Dict[int, Dict[str, Any]]:
return {g['GNR']: g for g in parse_csv(f'{in_dir}/TGemeinden.csv')}
def parse_flaechenbindungen(in_dir: str) -> Dict[int, Dict[int, Dict[str, Any]]]:
fbs = parse_csv(f'{in_dir}/TFlaechenbindungen.csv')
members = {}
@ -272,28 +268,72 @@ def lookup_kgnr(okz: Optional[int]) -> Optional[int]:
return None
def lookup_kgnr_name(name: str) -> Optional[int]:
def lookup_gem_name(name: str) -> List[Tuple[int, int]]:
if name.lower() == 'dörfles':
return [(6004, 30860)]
elif name.lower() == 'velm-götzendorf':
return [(6027, 30859), (6007, 30859)]
elif name.lower() == 'grub':
name = 'Grub an der March'
cur = DB_CNX.cursor()
cur.execute("SELECT k.kgnr FROM AT_kg k JOIN AT_gem g ON g.gkz = k.gkz JOIN wb_gem wg ON wg.gkz = g.gkz "
"WHERE LOWER(k.name) = LOWER(?) AND wg.hkid = 'WLWV'",
(name.replace('Gr.', 'Groß ').replace('-', '').replace(' ', ''),))
rows: List[Tuple[int]] = cur.fetchall()
cur.execute("SELECT k.kgnr, k.name, g.gkz, g.name FROM AT_kg k JOIN AT_gem g ON g.gkz = k.gkz JOIN wb_gem wg ON wg.gkz = g.gkz "
"WHERE LOWER(k.name) LIKE (LOWER(?) || '%') AND wg.hkid = 'WLWV'",
(name.replace('Gr.', 'Groß ').replace('Groß ', 'Groß').replace('-', ''),))
rows: List[Tuple[int, str, int, str]] = cur.fetchall()
cur.close()
if len(rows) == 1:
return rows[0][0]
if name == 'Velm-Götzendorf':
return None
return [(k, g) for k, _, g, _ in rows]
print(name, rows)
raise RuntimeError()
def migrate_branches(in_dir: str, out_dir: str) -> None:
global BRANCH_MAP
BRANCH_MAP = {}
with open(f'{out_dir}/branch.csv', 'w+') as f:
f.write('zwstid;name;country;postal_dest;address;phone_nr\n')
for b in parse_csv(f'{in_dir}/TZweigstellen.csv'):
BRANCH_MAP[b['ZNR']] = b['Kennbst']
address = b['Straße']
postal_dest = lookup_plz(int(b['PLZ']) if b['PLZ'] else None, b['Ort'], address)
f.write(format_row(b['Kennbst'], b['Name'], 'AT', postal_dest, address, b['Telefon']))
def migrate_gemeinden(in_dir: str, out_dir: str) -> None:
global GEM_MAP
GEM_MAP = {}
for g in parse_csv(f'{in_dir}/TGemeinden.csv'):
GEM_MAP[g['GNR']] = lookup_gem_name(g['Bezeichnung'])
def migrate_reeds(in_dir: str, out_dir: str) -> None:
global REED_MAP
REED_MAP = {}
with open(f'{out_dir}/wb_rd.csv', 'w+') as f:
f.write('kgnr;rdnr;name\n')
for r in parse_csv(f'{in_dir}/TRiede.csv'):
name: str = r['Bezeichnung'].strip()
if name.isupper():
name = name.title()
gem = GEM_MAP[r['GNR']]
kgnr = gem[0][0]
if len(gem) != 1:
print(gem, name, '->', gem[0])
rdnr = max([n for k, n in REED_MAP.values() if k == kgnr] or [0]) + 1
REED_MAP[r['RNR']] = (kgnr, rdnr)
f.write(format_row(kgnr, rdnr, name))
def migrate_members(in_dir: str, out_dir: str) -> None:
members = parse_csv(f'{in_dir}/TMitglieder.csv')
branches = parse_branches(in_dir)
gemeinden = parse_gemeinden(in_dir)
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:
@ -485,13 +525,13 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
elif phone_landline != phone_2:
invalid(mgnr, 'Tel.Nr.', phone_2)
zwstid = m['ZNR'] and branches[m['ZNR']]['Kennbst'] or len(branches) == 1 and list(branches.values())[0]['Kennbst']
zwstid = m['ZNR'] and BRANCH_MAP[m['ZNR']] or len(BRANCH_MAP) == 1 and list(BRANCH_MAP.values())[0]
postal_dest = lookup_plz(int(m['PLZ']) if m['PLZ'] else None, m['Ort'], address)
#if mgnr in fbs:
# gems = {v['GNR'] for k, v in fbs[mgnr].items() if v['Bis'] and int(v['Bis']) >= 2020}
# if len(gems) == 1:
# print(gemeinden[list(gems)[0]])
# print(GEM_MAP[list(gems)[0]])
okz = postal_dest % 100000 if postal_dest else None
kgnr = lookup_kgnr(okz)
@ -512,8 +552,6 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
def migrate_contracts(in_dir: str, out_dir: str) -> None:
gemeinden = parse_gemeinden(in_dir)
def parse_gstnrs(nr_str: str, kgnr: int, mgnr: int) -> List[str]:
if nr_str is None:
return []
@ -573,11 +611,8 @@ def migrate_contracts(in_dir: str, out_dir: str) -> None:
continue
parz: str = fb['Parzellennummer']
vnr: int = fb['FBNR']
gem = gemeinden[fb['GNR']]
kgnr = lookup_kgnr_name(gem['Bezeichnung'])
if kgnr is None:
# Götzendorf
kgnr = 6007
gem = GEM_MAP[fb['GNR']]
kgnr = gem[0][0]
f_c.write(format_row(vnr, fb['MGNR'], fb['Von'], fb['Bis'] if fb['Bis'] and fb['Bis'] < 3000 else None))
gstnrs = parse_gstnrs(parz, kgnr, fb['MGNR'])
@ -592,8 +627,8 @@ def migrate_contracts(in_dir: str, out_dir: str) -> None:
for i, gstnr in enumerate(gstnrs or ['0000']):
a = area - gst_area * (len(gstnrs) - 1) if i == 0 else gst_area
# TODO reed nr
f_fb.write(format_row(vnr, kgnr, gstnr, None, a, fb['SNR'], fb['SANR'], CULTIDS[fb['BANR']]))
rdnr = REED_MAP[fb['RNR']][1] if fb['RNR'] else None
f_fb.write(format_row(vnr, kgnr, gstnr, rdnr, a, fb['SNR'], fb['SANR'], CULTIVATION_MAP[fb['BANR']]))
if __name__ == '__main__':
@ -608,6 +643,9 @@ if __name__ == '__main__':
DB_CNX = sqlite3.connect(args.database)
migrate_branches(args.in_dir, args.out_dir)
migrate_gemeinden(args.in_dir, args.out_dir)
migrate_reeds(args.in_dir, args.out_dir)
migrate_members(args.in_dir, args.out_dir)
migrate_contracts(args.in_dir, args.out_dir)