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

database: fix billing calculcation

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-11-02 11:56:42 +01:00
parent 922957d72d
commit 462513f2a4
10 changed files with 173 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
wgmaster/migrate.py
View File

@ -44,6 +44,7 @@ MEMBER_MAP: Optional[Dict[int, Dict[str, Any]]] = None
GROSSLAGE_KG_MAP: Optional[Dict[int, int]] = None
DELIVERY_MAP: Optional[Dict[int, Tuple[int, int, int]]] = None
MODIFIER_MAP: Optional[Dict[str, Dict]] = None
PARAMETERS: Optional[Dict[str, str]] = None
AUSTRIA = 40
WGMASTER_PRECISION = 4
@ -601,14 +602,17 @@ def migrate_reeds(in_dir: str, out_dir: str) -> None:
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', 'max_kg_per_ha', 'fill_lower_bins', 'active')
f.header('attrid', 'name', 'active', 'max_kg_per_ha', 'strict', 'fill_lower')
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, 0, True)
max_kg = int(a['KgProHa']) if a['KgProHa'] is not None else None
if max_kg == 10_000:
max_kg = None
f.row(a['SANR'], a['Attribut'], True, max_kg, False, 0)
if CLIENT == WG.MATZEN:
f.row('M', 'Matzen', None, 0, False)
f.row('HU', 'Huber', None, 0, False)
f.row('M', 'Matzen', False, None, False, 0)
f.row('HU', 'Huber', False, None, False, 0)
def migrate_cultivations(in_dir: str, out_dir: str) -> None:
@ -616,7 +620,7 @@ def migrate_cultivations(in_dir: str, out_dir: str) -> None:
CULTIVATION_MAP = {}
with utils.csv_open(f'{out_dir}/wine_cultivation.csv') as f:
f.header('cultid', 'name')
f.header('cultid', 'name', 'description')
for c in utils.csv_parse_dict(f'{in_dir}/TBewirtschaftungsarten.csv'):
name: str = c['Bezeichnung']
cultid = name[0].upper()
@ -625,28 +629,28 @@ def migrate_cultivations(in_dir: str, out_dir: str) -> None:
elif 'biolog' in name.lower():
cultid = 'BIO'
CULTIVATION_MAP[c['BANR']] = cultid
f.row(cultid, name)
f.row(cultid, name, None)
def migrate_area_commitment_types(in_dir: str, out_dir: str) -> None:
with utils.csv_open(f'{out_dir}/area_commitment_type.csv') as f:
f.header('vtrgid', 'sortid', 'attrid', 'disc',
'min_kg_per_ha', 'max_kg_per_ha', 'penalty_amount')
f.header('vtrgid', 'sortid', 'attrid', 'disc', 'min_kg_per_ha',
'penalty_per_kg', 'penalty_amount', 'penalty_none')
for t in utils.csv_parse_dict(f'{in_dir}/TLiefermengen.csv'):
sortid: str = t['SNR']
if not sortid or sortid == 'SV':
continue
menge = int(t['ErwarteteLiefermengeProHa'])
f.row(sortid + (t['SANR'] or ''), sortid[:2], t['SANR'] or sortid[2:] or None, None,
menge, menge, None)
f.row(sortid + (t['SANR'] or ''), sortid[:2], t['SANR'] or sortid[2:] or None, None, menge,
None, None, None)
bio = []
if CLIENT == WG.MATZEN:
bio = ['GV', 'ZW', 'MT']
f.row('BM', 'BM', None, None, None, None, None)
f.row('BM', 'BM', None, None, None, None, None, None)
elif CLIENT == WG.WINZERKELLER:
bio = ['GV', 'ZW', 'WR', 'MT', 'RR', 'WB', 'CH', 'MU']
for sortid in bio:
f.row(f'{sortid}B', sortid, 'B', None, None, None, None)
f.row(f'{sortid}B', sortid, 'B', None, None, None, None, None)
def normalize_name(family_name: str, given_name: str) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]]:
@ -767,7 +771,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
f_m.header(
'mgnr', 'predecessor_mgnr', 'prefix', 'given_name', 'middle_names', 'family_name', 'suffix',
'birthday', 'entry_date', 'exit_date', 'business_shares', 'accounting_nr', 'zwstid',
'lfbis_nr', 'ustid_nr', 'volllieferant', 'buchführend', 'funktionär', 'active', 'deceased',
'lfbis_nr', 'ustid_nr', 'volllieferant', 'buchführend', 'organic', 'funktionär', 'active', 'deceased',
'iban', 'bic', 'country', 'postal_dest', 'address', 'default_kgnr', 'comment')
f_mba.header('mgnr', 'name', 'country', 'postal_dest', 'address')
f_tel.header('mgnr', 'nr', 'type', 'number', 'comment')
@ -952,7 +956,7 @@ def migrate_members(in_dir: str, out_dir: str) -> None:
mgnr, pred, prefix, given_name, middle_names, family_name, suffix,
m['Geburtsjahr'], m['Eintrittsdatum'], m['Austrittsdatum'], m['Geschäftsanteile1'] or 0,
m['BHKontonummer'], zwstid, bnr, ustid_nr,
m['Volllieferant'] or False, m['Buchführend'] or False, funktionaer, active, deceased,
m['Volllieferant'] or False, m['Buchführend'] or False, False, funktionaer, active, deceased,
iban, bic, AUSTRIA, postal_dest, address or '-', kgnr, m['Anmerkung']
)
@ -1521,10 +1525,14 @@ def migrate_deliveries(in_dir: str, out_dir: str) -> None:
with utils.csv_open(f'{out_dir}/season.csv') as f_season, \
utils.csv_open(f'{out_dir}/modifier.csv') as f_mod:
f_season.header('year', 'currency', 'precision', 'start_date', 'end_date')
f_season.header('year', 'currency', 'precision', 'max_kg_per_ha', 'vat_normal', 'vat_flatrate',
'min_kg_per_bs', 'max_kg_per_bs', 'penalty_per_kg', 'penalty_amount', 'penalty_none',
'start_date', 'end_date')
f_mod.header('year', 'modid', 'ordering', 'name', 'abs', 'rel', 'standard', 'quick_select')
for y, s in seasons.items():
f_season.row(y, s['currency'], s['precision'], s['start'], s['end'])
f_season.row(y, s['currency'], s['precision'], 10_000, 0.10, 0.13,
PARAMETERS['LIEFERPFLICHT/GA1'], PARAMETERS['LIEFERRECHT/GA1'],
None, None, None, s['start'], s['end'])
for m in modifiers.values():
abs_v = round(m['AZAS'] * pow(10, s['precision'])) if m['AZAS'] is not None else None
rel_v = m['AZASProzent'] / 100.0 if m['AZASProzent'] is not None else None
@ -1677,9 +1685,9 @@ def migrate_payments(in_dir: str, out_dir: str) -> None:
# TODO database migration
with utils.csv_open(f'{out_dir}/payment_delivery_part.csv') as f_del_pay, \
utils.csv_open(f'{out_dir}/delivery_part_bin.csv') as f_bin:
utils.csv_open(f'{out_dir}/delivery_part_bucket.csv') as f_bucket:
f_del_pay.header('year', 'did', 'dpnr', 'avnr', 'amount')
f_bin.header('year', 'did', 'dpnr', 'binnr', 'discr', 'value')
f_bucket.header('year', 'did', 'dpnr', 'bktnr', 'discr', 'value')
deliveries = {d['LINR']: d for d in utils.csv_parse_dict(f'{in_dir}/TLieferungen.csv')}
for linr, (y, did, dpnr) in DELIVERY_MAP.items():
p = deliveries[linr]
@ -1688,8 +1696,8 @@ def migrate_payments(in_dir: str, out_dir: str) -> None:
gew, geb_gew = int(p['Gewicht']), int(p['BGewichtGebunden'])
b1 = gew - geb_gew
b2 = geb_gew
f_bin.row(y, did, dpnr, 1, '_', b1)
f_bin.row(y, did, dpnr, 2, p['SANR'] or '', b2)
f_bucket.row(y, did, dpnr, 0, '_', b1)
f_bucket.row(y, did, dpnr, 1, p['SANR'] or '', b2)
for aznr, avnr, tznr in variant_year_map[y]:
val = p[f'BTeilzahlung{tznr}' if tznr < 6 else 'BEndauszahlung']
val = round(val * pow(10, WGMASTER_PRECISION))
@ -1709,9 +1717,10 @@ def migrate_payments(in_dir: str, out_dir: str) -> None:
def migrate_parameters(in_dir: str, out_dir: str) -> None:
params: Dict[str, str] = {p['Bezeichnung']: p['Wert'] for p in utils.csv_parse_dict(f'{in_dir}/TParameter.csv')}
name = params['MANDANTENNAME1'].title().replace('F.', 'für').replace('U.', 'und').replace(' Im ', ' im ')
suffix = params['MANDANTENNAME2'].lower().replace(' ', '').replace('.', '')
global PARAMETERS
PARAMETERS = {p['Bezeichnung']: p['Wert'] for p in utils.csv_parse_dict(f'{in_dir}/TParameter.csv')}
name = PARAMETERS['MANDANTENNAME1'].title().replace('F.', 'für').replace('U.', 'und').replace(' Im ', ' im ')
suffix = PARAMETERS['MANDANTENNAME2'].lower().replace(' ', '').replace('.', '')
types = {
'reggenmbh': 'reg. Gen.m.b.H.'
}
@ -1720,32 +1729,27 @@ def migrate_parameters(in_dir: str, out_dir: str) -> None:
WG.WINZERKELLER: ('WKW', 'Winzerkeller')
}.get(CLIENT, (None, None))
ort = params['MANDANTENORT'].title()
ort = PARAMETERS['MANDANTENORT'].title()
new_params: Dict[str, Optional[str]] = {
'CLIENT_NAME_TOKEN': tokens[0],
'CLIENT_NAME_SHORT': tokens[1],
'CLIENT_NAME': name,
'CLIENT_NAME_SUFFIX': None,
'CLIENT_NAME_TYPE': types[suffix],
'CLIENT_PLZ': params['MANDANTENPLZ'],
'CLIENT_PLZ': PARAMETERS['MANDANTENPLZ'],
'CLIENT_ORT': ort,
'CLIENT_ADDRESS': params['MANDANTENSTRASSE'],
'CLIENT_ADDRESS': PARAMETERS['MANDANTENSTRASSE'],
'CLIENT_IBAN': None,
'CLIENT_BIC': None,
'CLIENT_USTIDNR': params['MANDANTENUID'].replace(' ', ''),
'CLIENT_LFBISNR': params['MANDANTENBETRIEBSNUMMER'],
'CLIENT_PHONE': normalize_phone_nr(params['MANDANTENTELEFON'], ort),
'CLIENT_FAX': normalize_phone_nr(params['MANDANTENTELEFAX'], ort),
'CLIENT_EMAIL': params['MANDANTENEMAIL'],
'CLIENT_WEBSITE': params.get('MANDANTENHOMEPAGE', None),
'DELIVERY_OBLIGATION': params.get('LIEFERPFLICHT/GA1', None),
'DELIVERY_RIGHT': params.get('LIEFERRECHT/GA1', None),
'VAT_NORMAL': '0.20',
'VAT_REDUCED': '0.10',
'VAT_FLATRATE': '0.13',
'DOCUMENT_SENDER': params.get('ABSENDERTEXT2', None),
'TEXT_DELIVERYNOTE': params.get('LIEFERSCHEINTEXT', None).replace(' daß ', ' dass ').replace('obige Angaben maßgeblicher Veränderungen', 'maßgeblichen Veränderungen obiger Angaben'),
'TEXT_DELIVERYCONFIRMATION': params.get('ANLIEFTEXT', None),
'CLIENT_USTIDNR': PARAMETERS['MANDANTENUID'].replace(' ', ''),
'CLIENT_LFBISNR': PARAMETERS['MANDANTENBETRIEBSNUMMER'],
'CLIENT_PHONE': normalize_phone_nr(PARAMETERS['MANDANTENTELEFON'], ort),
'CLIENT_FAX': normalize_phone_nr(PARAMETERS['MANDANTENTELEFAX'], ort),
'CLIENT_EMAIL': PARAMETERS['MANDANTENEMAIL'],
'CLIENT_WEBSITE': PARAMETERS.get('MANDANTENHOMEPAGE', None),
'DOCUMENT_SENDER': PARAMETERS.get('ABSENDERTEXT2', None),
'TEXT_DELIVERYNOTE': PARAMETERS.get('LIEFERSCHEINTEXT', None).replace(' daß ', ' dass ').replace('obige Angaben maßgeblicher Veränderungen', 'maßgeblichen Veränderungen obiger Angaben'),
'TEXT_DELIVERYCONFIRMATION': PARAMETERS.get('ANLIEFTEXT', None),
}
with utils.csv_open(f'{out_dir}/client_parameter.csv') as f:
@ -1777,6 +1781,7 @@ def main() -> None:
DB_CNX = sqlite3.connect(args.database)
migrate_parameters(args.in_dir, args.out_dir)
migrate_gradation(args.in_dir, args.out_dir)
migrate_branches(args.in_dir, args.out_dir)
migrate_grosslagen(args.in_dir, args.out_dir)
@ -1789,7 +1794,6 @@ def main() -> None:
migrate_area_commitments(args.in_dir, args.out_dir)
migrate_deliveries(args.in_dir, args.out_dir)
migrate_payments(args.in_dir, args.out_dir)
migrate_parameters(args.in_dir, args.out_dir)
DB_CNX.close()