# Part of the odoo-at-payroll project. License: LGPL-3.
"""TASY-LSWH-Import (SV-Tarifsystem, sozialversicherung.at/tarifsystem).
Zwei Stufen:
1. ``l10n.at.payroll.tasy.import.wizard`` — SV-Werte als
hr.rule.parameter (Mapping-gesteuert, Vorschau, historische
Werte bleiben erhalten; GP4).
2. ``l10n.at.payroll.tasy.katalog.wizard`` — Vollständiger
Tarifsystem-Katalog (AP0, Variante b): BSGR/ERGB/Dienstgeber-
gruppen/Tarifgruppen/Tarifsets/Tarife/Zu-/Abschläge, alle
Träger, versioniert über von/bis. Die mBGM-Tarifierung liest
daraus; der Hand-Parameter ``at_mbgm_tarif`` bleibt Fallback.
"""
from lxml import etree
import base64
from odoo import _, api, fields, models
from odoo.exceptions import UserError
def _l10n_at_tasy_root(file_field):
"""TASY-XML robust laden: rohe XML-Bytes ODER base64."""
parser = etree.XMLParser(resolve_entities=False, no_network=True)
data = file_field or b''
if isinstance(data, str):
data = data.encode()
# Binary kann XML-Bytes ODER base64 (str/Bytes) liefern:
# dekodieren, wenn kein '<' am Anfang steht.
if not data.lstrip().startswith(b'<'):
data = base64.b64decode(data)
return etree.fromstring(data, parser)
class L10nAtPayrollTasyMapping(models.Model):
"""Mapping TASY-Code/-Zweig -> SV-Regelparameter (hr.rule.parameter).
Hält die Zuordnung deklarativ (data XML), damit der TASY-Import ohne
feste Werte im Code funktioniert und für künftige Export-Formate
(z. B. BVAEB) erweiterbar bleibt. Für Tarifzweige wird an den
Parameter-Code ``_dn``/``_dg`` angehängt.
"""
_name = 'l10n.at.payroll.tasy.mapping'
_description = 'TASY-Mapping (SV-Regelparameter)'
_order = 'tasy_art, tasy_code'
tasy_art = fields.Selection(
[('bbb', 'Beitragsbemessungsbetrag (Grenzwert)'),
('zweig', 'Tarifzweig (Beitragssatz)')],
required=True)
tasy_code = fields.Char(required=True, index=True)
parameter_code = fields.Char(
required=True, index=True,
help="Code des hr.rule.parameter; bei Tarifzweigen wird _dn/_dg "
"angehängt.")
cent_werte = fields.Boolean(
default=True, string='Cent-Werte',
help="TASY speichert Beitragsbemessungsbeträge in Cent.")
faktor = fields.Float(
default=1.0,
help="Multiplikator, z. B. 30 für die monatliche HBG aus dem "
"täglichen Wert (§ 45 Abs 1 ASVG).")
active = fields.Boolean(default=True)
_unique_mapping = models.Constraint(
'unique (tasy_art, tasy_code)',
"Je TASY-Art und -Code gibt es nur ein Mapping.")
@api.depends('tasy_art', 'tasy_code', 'parameter_code')
def _compute_display_name(self):
art_label = dict(self._fields['tasy_art'].selection)
for mapping in self:
art = art_label.get(mapping.tasy_art, '')
mapping.display_name = (
f"{mapping.tasy_code} → {mapping.parameter_code}"
f" ({art})" if art else mapping.tasy_code)
class L10nAtPayrollTasyImportLine(models.TransientModel):
_name = 'l10n.at.payroll.tasy.import.line'
_description = 'TASY-Import: Vorschauzeile'
wizard_id = fields.Many2one(
'l10n.at.payroll.tasy.import.wizard', required=True, ondelete='cascade')
parameter_code = fields.Char(required=True)
name = fields.Char()
alter_wert = fields.Char(string='Alter Wert', readonly=True)
neuer_wert = fields.Char(string='Neuer Wert', required=True)
date_from = fields.Date(required=True)
uebernehmen = fields.Boolean(string='Übernehmen', default=True)
warnung = fields.Char(help="Leer = eindeutig aufgelöst; sonst "
"Plausibilitäts-/Auflösungshinweis. "
"Warnzeilen werden nie automatisch übernommen.")
class L10nAtPayrollTasyImportWizard(models.TransientModel):
"""Lädt einen jährlichen ÖGK TASY-LSWH-Export und schreibt nach
expliziter Bestätigung SV-Werte als hr.rule.parameter-Werte
(Versionierung über date_from, historische Werte bleiben erhalten).
Auflösung am Gültigkeitstichtag: Beitragsbemessungsbeträge direkt
über den Code; Beitragssätze über die Kette Beschäftigtengruppe
(Standard: B002 „Angestellte") → Tarifgruppe → tarifgruppeSetZuord →
Tarifset → Tarif. Bei mehreren Varianten gewinnt die Split-Variante
(DN>0 und DG>0), danach die einseitige Dienstgeber-Variante (UV/BV);
freie-Dienstnehmer-Varianten (nur DN) verlieren. Mehrdeutige
Auflösungen erzeugen Warnzeilen ohne automatische Übernahme.
"""
_name = 'l10n.at.payroll.tasy.import.wizard'
_description = 'SV-Werte aus TASY-LSWH-Export importieren'
file = fields.Binary(required=True, string='TASY-Export (XML)')
filename = fields.Char()
stichtag = fields.Date(readonly=True,
help="Gültigkeitstichtag des Exports.")
beschaeftigtengruppe_code = fields.Char(
default='B002', required=True,
help="TASY-Code der Beschäftigtengruppe (B002 = Angestellte).")
line_ids = fields.One2many(
'l10n.at.payroll.tasy.import.line', 'wizard_id',
string='Vorschau')
protocol = fields.Text(readonly=True)
# ------------------------------------------------------------------
# Parsing
# ------------------------------------------------------------------
def _tasy_root(self):
self.ensure_one()
return _l10n_at_tasy_root(self.file)
@staticmethod
def _tasy_text(node, tag):
child = node.find(tag)
return (child.text or '').strip() if child is not None else ''
def _parse_bbbwerte(self, root):
codes = {b.get('xmlId'): self._tasy_text(b, 'code')
for b in root.iter('beitragbemessungsbetrag')}
werte = []
for w in root.iter('bbbwert'):
werte.append({
'code': codes.get(self._tasy_text(w, 'beitragbemessungsbetrag_id')),
'betrag': float(self._tasy_text(w, 'betrag')),
'von': self._tasy_text(w, 'vonDatum'),
'bis': self._tasy_text(w, 'bisDatum'),
})
return werte
def _parse_tarife(self, root):
return {t.get('xmlId'): {
'zweig': self._tasy_text(t, 'tarifZweig'),
'dn': float(self._tasy_text(t, 'dienstnehmerAnteil')),
'dg': float(self._tasy_text(t, 'dienstgeberAnteil')),
'von': self._tasy_text(t, 'vonDatum'),
'bis': self._tasy_text(t, 'bisDatum'),
} for t in root.iter('tarif')}
def _resolve_zweig_pairs(self, root, stichtag):
"""Standard-Beitragssätze je Zweig über die Gruppenkette."""
self.ensure_one()
gruppen_ids = {g.get('xmlId') for g in root.iter('beschaeftigtengruppe')
if self._tasy_text(g, 'code') == self.beschaeftigtengruppe_code}
tarifgruppen = {tg.get('xmlId') for tg in root.iter('tarifgruppe')
if any(self._tasy_text(tg, 'beschaeftigtengruppen') == gid
for gid in gruppen_ids)}
sets = {s.get('xmlId'): [t.text.strip() for t in s.iter('tarife')]
for s in root.iter('tarifset')}
valid_sets = {self._tasy_text(z, 'tarifset')
for z in root.iter('tarifgruppeSetZuord')
if self._tasy_text(z, 'tarifgruppe') in tarifgruppen
and self._tasy_text(z, 'vonDatum') <= stichtag
<= self._tasy_text(z, 'bisDatum')}
tarife = self._parse_tarife(root)
pairs = {}
for sid in valid_sets:
for tid in sets.get(sid, []):
t = tarife.get(tid)
if not t or not (t['von'] <= stichtag <= t['bis']):
continue
pairs.setdefault(t['zweig'], {}).setdefault(
(t['dn'], t['dg']), t['von'])
return pairs
# ------------------------------------------------------------------
# Actions
# ------------------------------------------------------------------
def action_parse(self):
self.ensure_one()
root = self._tasy_root()
stichtag = (root.findtext('gueltigkeitstichtag') or '').strip()
if not stichtag:
raise UserError(_('Ungültiger TASY-Export: '
'Gültigkeitstichtag fehlt.'))
self.stichtag = stichtag
lines, protocol = self._build_lines(root, stichtag)
self.line_ids = [(5, 0, 0)] + lines
self.protocol = "\n".join(protocol)
return True
def _build_lines(self, root, stichtag):
self.ensure_one()
line_vals, protocol = [], []
parameter_obj = self.env['hr.rule.parameter']
bbbwerte = self._parse_bbbwerte(root)
for mapping in self.env['l10n.at.payroll.tasy.mapping'].search(
[('tasy_art', '=', 'bbb')]):
candidates = [w for w in bbbwerte if w['code'] == mapping.tasy_code
and w['von'] <= stichtag <= w['bis']]
if not candidates:
protocol.append(
f"KEIN WERT: {mapping.tasy_code} "
f"({mapping.parameter_code}) am Stichtag")
continue
w = max(candidates, key=lambda x: x['von'])
value = w['betrag'] * mapping.faktor
if mapping.cent_werte:
value /= 100.0
line_vals.append(self._line_vals(
mapping.parameter_code, round(value, 2), w['von'],
parameter_obj, protocol))
pairs = self._resolve_zweig_pairs(root, stichtag)
for mapping in self.env['l10n.at.payroll.tasy.mapping'].search(
[('tasy_art', '=', 'zweig')]):
variants = pairs.get(mapping.tasy_code, {})
chosen = None
split = [(dn, dg, von) for (dn, dg), von in variants.items()
if dn > 0 and dg > 0]
employer = [(dn, dg, von) for (dn, dg), von in variants.items()
if dg > 0 and dn == 0]
if len(variants) == 1:
(dn, dg), von = next(iter(variants.items()))
chosen = (dn, dg, von)
elif len(split) == 1:
chosen = split[0]
elif len(employer) == 1:
chosen = employer[0]
if not chosen:
protocol.append(
f"NICHT EINDEUTIG: Zweig {mapping.tasy_code} "
f"({mapping.parameter_code}): {dict(variants)}")
for (dn, dg), von in variants.items():
line_vals.append(self._line_vals(
f"{mapping.parameter_code}_dn", dn, von, parameter_obj,
protocol, warnung=f"Mehrdeutig: DN {dn} / DG {dg}"))
continue
dn, dg, von = chosen
line_vals.append(self._line_vals(
f"{mapping.parameter_code}_dn", dn, von, parameter_obj, protocol))
line_vals.append(self._line_vals(
f"{mapping.parameter_code}_dg", dg, von, parameter_obj, protocol))
return line_vals, protocol
def _line_vals(self, code, value, date_from, parameter_obj, protocol,
warnung=None):
self.ensure_one()
old = parameter_obj._get_parameter_from_code(code, date_from,
raise_if_not_found=False)
# _get_parameter_from_code liefert den geparsten Wert (float/dict)
# oder None — nicht den Parameter-Record.
old_value = repr(old) if old is not None else ''
old_float = None
try:
old_float = float(old_value)
except (TypeError, ValueError):
old_float = None
if old_float and abs(value - old_float) / old_float > 0.10:
warnung = (warnung or '') + f" Sprung > 10 % ({old_value} → {value})"
return (0, 0, {
'parameter_code': code,
'name': parameter_obj.search(
[('code', '=', code)]).name or code,
'alter_wert': old_value,
'neuer_wert': repr(value),
'date_from': date_from,
'uebernehmen': not warnung,
'warnung': warnung or '',
})
def action_apply(self):
self.ensure_one()
parameter_obj = self.env['hr.rule.parameter']
value_obj = self.env['hr.rule.parameter.value']
country = self.env.ref('base.at')
applied = 0
for line in self.line_ids.filtered('uebernehmen'):
parameter = parameter_obj.search(
[('code', '=', line.parameter_code)], limit=1)
if not parameter:
parameter = parameter_obj.create({
'name': line.name or line.parameter_code,
'code': line.parameter_code,
'country_id': country.id,
})
value = value_obj.search([
('rule_parameter_id', '=', parameter.id),
('date_from', '=', line.date_from)], limit=1)
if value:
value.parameter_value = line.neuer_wert
else:
value_obj.create({
'rule_parameter_id': parameter.id,
'date_from': line.date_from,
'parameter_value': line.neuer_wert,
})
applied += 1
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'type': 'success',
'message': _("%(count)s Parameterwert(e) geschrieben.",
count=applied),
'next': {'type': 'ir.actions.act_window_close'},
},
}
class L10nAtPayrollTasyKatalogWizard(models.TransientModel):
"""Lädt den vollständigen Tarifsystem-Katalog aus dem
TASY-LSWH-Export (AP0, Variante b) — alle Trägergruppen,
versioniert über von/bis-Datum. Idempotent: wiederholter Import
derselben Datei aktualisiert statt zu duplizieren; historische
Gültigkeitsfenster bleiben erhalten (Jahreswechsel = neuer
Import, keine Löschung).
Die Sätze selbst werden NICHT hieraus in hr.rule.parameter
geschrieben — die mBGM-Tarifierung liest den Katalog direkt
(models/tasy_katalog.py); nur die Zweig-Split-Werte für den
GP4-Kern laufen über den Wert-Import-Wizard.
"""
_name = 'l10n.at.payroll.tasy.katalog.wizard'
_description = 'SV-Tarifsystem-Katalog importieren (TASY)'
file = fields.Binary(required=True, string='TASY-Export (XML)')
filename = fields.Char()
stichtag = fields.Date(
readonly=True,
help="Gültigkeitstichtag des Exports (nur Dokumentation — "
"die Katalogzeilen tragen eigene von/bis-Fenster).")
protocol = fields.Text(readonly=True)
@staticmethod
def _tasy_text(node, tag):
child = node.find(tag)
return (child.text or '').strip() if child is not None else ''
@staticmethod
def _tasy_texts(node, tag):
"""Alle Kindelemente dieses Tags — TASY nutzt wiederholte
Referenz-Elemente (z. B. xmlId)."""
return [child.text.strip() for child in node.findall(tag)
if child.text and child.text.strip()]
@staticmethod
def _tasy_bool(text):
return (text or '').strip().lower() in ('true', '1', 'j', 'ja',
'yes', 'y')
def _tasy_root(self):
self.ensure_one()
return _l10n_at_tasy_root(self.file)
@staticmethod
def _section(root, tag):
"""Sektionsscharfe Iteration — root.iter() erfasste auch die
Referenz-Elemente (z. B. xmlId in
Zuordnungszeilen); die Sektionen enthalten nur echte Zeilen."""
section = root.find(tag)
return list(section) if section is not None else []
# ------------------------------------------------------------------
# Upsert-Helfer
# ------------------------------------------------------------------
def _upsert_code_katalog(self, model, nodes, protocol, label,
extra_vals=None):
"""Katalogzeilen mit (Trägergruppe, Code, von) als Schlüssel."""
created = updated = 0
for node in nodes:
code = self._tasy_text(node, 'code')
von = self._tasy_text(node, 'vonDatum')
if not code or not von:
continue
vals = {
'code': code,
'bezeichnung': self._tasy_text(node, 'bezeichnung'),
'kurzbezeichnung': self._tasy_text(
node, 'kurzbezeichnung'),
'beschreibung': self._tasy_text(node, 'beschreibung'),
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
}
vals.update(extra_vals(node) if extra_vals else {})
rec = self.env[model].search([
('traegergruppe', '=', vals['traegergruppe']),
('code', '=', code),
('von_datum', '=', vals['von_datum'])], limit=1)
if rec:
rec.write(vals)
updated += 1
else:
self.env[model].create(vals)
created += 1
protocol.append('%s: %d neu, %d aktualisiert'
% (label, created, updated))
def _upsert_tarif(self, nodes, protocol):
"""Tarife ohne Code — Identität ist der Inhalt (Zweig, Sätze,
Gültigkeit); xmlId zuerst, dann Inhaltsvergleich."""
created = updated = 0
for node in nodes:
zweig = self._tasy_text(node, 'tarifZweig')
von = self._tasy_text(node, 'vonDatum')
if not zweig or not von:
continue
vals = {
'zweig': zweig,
'dn': float(self._tasy_text(
node, 'dienstnehmerAnteil') or 0.0),
'dg': float(self._tasy_text(
node, 'dienstgeberAnteil') or 0.0),
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
}
rec = self.env['l10n.at.payroll.tasy.tarif'].search([
('zweig', '=', zweig),
('traegergruppe', '=', vals['traegergruppe']),
('von_datum', '=', vals['von_datum'])]).filtered(
lambda t, v=vals: abs(t.dn - v['dn']) < 1e-9
and abs(t.dg - v['dg']) < 1e-9)[:1]
if rec:
rec.write(vals)
updated += 1
else:
self.env['l10n.at.payroll.tasy.tarif'].create(vals)
created += 1
protocol.append('Tarife: %d neu, %d aktualisiert'
% (created, updated))
def _upsert_tarifset(self, nodes, tarif_map, protocol):
"""Tarifsets — Schlüssel (Träger, Kurzbezeichnung, von);
Tarif-Verknüpfungen werden je Import neu gesetzt."""
created = updated = 0
for node in nodes:
von = self._tasy_text(node, 'vonDatum')
kurz = self._tasy_text(node, 'kurzbezeichnung')
if not von:
continue
vals = {
'bezeichnung': self._tasy_text(node, 'bezeichnung'),
'kurzbezeichnung': kurz,
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
}
rec = self.env['l10n.at.payroll.tasy.tarifset'].search([
('traegergruppe', '=', vals['traegergruppe']),
('kurzbezeichnung', '=', kurz),
('von_datum', '=', vals['von_datum'])], limit=1)
tarife = [tarif_map[ref] for ref in self._tasy_texts(
node, 'tarife') if ref in tarif_map]
vals['tarif_ids'] = [(6, 0, [t.id for t in tarife])]
if rec:
rec.write(vals)
updated += 1
else:
rec = self.env['l10n.at.payroll.tasy.tarifset'].create(
vals)
created += 1
protocol.append('Tarifsets: %d neu, %d aktualisiert'
% (created, updated))
# ------------------------------------------------------------------
# Import
# ------------------------------------------------------------------
def action_import(self):
self.ensure_one()
root = self._tasy_root()
stichtag = (root.findtext('gueltigkeitstichtag') or '').strip()
if not stichtag:
raise UserError(_('Ungültiger TASY-Export: '
'Gültigkeitstichtag fehlt.'))
self.stichtag = fields.Date.to_date(stichtag)
protocol = ['Stichtag des Exports: %s' % stichtag]
# 1) Code-Kataloge: BSGR, Ergänzungen, Dienstgebergruppen,
# Zu-/Abschläge (inkl. Steuerung).
self._upsert_code_katalog(
'l10n.at.payroll.tasy.bsgr',
self._section(root, 'beschaeftigtengruppen'), protocol,
'BSGR')
self._upsert_code_katalog(
'l10n.at.payroll.tasy.ergaenzung',
self._section(root, 'ergaenzungen'), protocol,
'Ergänzungen')
self._upsert_code_katalog(
'l10n.at.payroll.tasy.dienstgebergruppe',
self._section(root, 'dienstgebergruppen'), protocol,
'Dienstgebergruppen')
bsgr_map = {b.tasy_xml_id: b for b in self.env[
'l10n.at.payroll.tasy.bsgr'].search([])}
erg_map = {e.tasy_xml_id: e for e in self.env[
'l10n.at.payroll.tasy.ergaenzung'].search([])}
dgg_map = {d.tasy_xml_id: d for d in self.env[
'l10n.at.payroll.tasy.dienstgebergruppe'].search([])}
zuabschlag_map = {}
zuabschlag_nodes = self._section(root, 'zuAbschlaege')
self._upsert_code_katalog(
'l10n.at.payroll.tasy.zuabschlag',
zuabschlag_nodes, protocol, 'Zu-/Abschläge',
extra_vals=lambda n: {
'zuabschlagkz': self._tasy_text(n, 'zuabschlagkz'),
'versicherungsrelevant': self._tasy_bool(
self._tasy_text(n, 'versicherungsRelevant')),
})
for za in self.env['l10n.at.payroll.tasy.zuabschlag'].search([]):
zuabschlag_map[za.tasy_xml_id] = za
steuerung_created = 0
steuerung_obj = self.env[
'l10n.at.payroll.tasy.zuabschlag.steuerung']
for node in self._section(root, 'zuAbschlaegeSteuerung'):
zuabschlag = zuabschlag_map.get(
self._tasy_text(node, 'zuAbschlag'))
von = self._tasy_text(node, 'vonDatum')
if not zuabschlag or not von:
continue
vals = {
'zuabschlag_id': zuabschlag.id,
'altersgrenze_min_jahr': float(
self._tasy_text(node, 'altersgrenzeMinJahr') or 0.0),
'altersgrenze_soll_jahr': float(
self._tasy_text(node, 'altersgrenzeSollJahr') or 0.0),
'geboren_bis': fields.Date.to_date(
self._tasy_text(node, 'geborenBis'))
if self._tasy_text(node, 'geborenBis') else False,
'geschlecht': self._tasy_text(node, 'geschlecht'),
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
}
if not steuerung_obj.search([
('tasy_xml_id', '=', vals['tasy_xml_id'])], limit=1):
steuerung_obj.create(vals)
steuerung_created += 1
protocol.append('Zu-/Abschlag-Steuerungen: %d neu'
% steuerung_created)
# 2) Tarife, dann Tarifgruppen (m2m auf die Kataloge),
# dann Tarifsets (m2m auf die Tarife).
self._upsert_tarif(self._section(root, 'tarife'), protocol)
tarif_map = {t.tasy_xml_id: t for t in self.env[
'l10n.at.payroll.tasy.tarif'].search([])}
tarifgruppe_obj = self.env['l10n.at.payroll.tasy.tarifgruppe']
created = updated = 0
for node in self._section(root, 'tarifgruppen'):
code = self._tasy_text(node, 'code')
von = self._tasy_text(node, 'vonDatum')
if not code or not von:
continue
vals = {
'code': code,
'bezeichnung': self._tasy_text(node, 'bezeichnung'),
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
'fallweise_zulaessig': self._tasy_bool(
self._tasy_text(node,
'fallweiseBeschaeftigungZulaessig')),
'bv_zulaessig': self._tasy_bool(
self._tasy_text(node,
'betrieblicheVorsorgeZulaessig')),
}
rec = tarifgruppe_obj.search([
('traegergruppe', '=', vals['traegergruppe']),
('code', '=', code),
('von_datum', '=', vals['von_datum'])], limit=1)
if not rec:
rec = tarifgruppe_obj.create(vals)
created += 1
else:
rec.write(vals)
updated += 1
rec.write({
'bsgr_ids': [(6, 0, [
bsgr_map[r].id for r in self._tasy_texts(
node, 'beschaeftigtengruppen')
if r in bsgr_map])],
'ergaenzung_ids': [(6, 0, [
erg_map[r].id for r in self._tasy_texts(
node, 'ergaenzungen') if r in erg_map])],
'dienstgebergruppe_ids': [(6, 0, [
dgg_map[r].id for r in self._tasy_texts(
node, 'dienstgeberGruppen')
if r in dgg_map])],
})
protocol.append('Tarifgruppen: %d neu, %d aktualisiert'
% (created, updated))
self._upsert_tarifset(self._section(root, 'tarifsets'),
tarif_map, protocol)
tarifset_map = {s.tasy_xml_id: s for s in self.env[
'l10n.at.payroll.tasy.tarifset'].search([])}
tarifgruppe_map = {t.tasy_xml_id: t for t in tarifgruppe_obj
.search([])}
# 3) Verrechnungsbasis-/Positionstypen + Kombinationen
# auflösen, dann tarifgruppeSetZuord importieren.
basis_typ = {b.get('xmlId'): self._tasy_text(b, 'code')
for b in self._section(root, 'verrechBasisTypen')}
pos_typ = {p.get('xmlId'): self._tasy_text(p, 'code')
for p in self._section(root, 'verrechPosTypen')}
kombinationen = {}
for k in self._section(root, 'verrechBasisPositionKombinationen'):
kombinationen[k.get('xmlId')] = (
basis_typ.get(self._tasy_text(k, 'verrechnungBasisTyp'),
''),
pos_typ.get(self._tasy_text(k, 'verrechnungPosTyp'), ''),
)
zuord_obj = self.env['l10n.at.payroll.tasy.tarifset.zuord']
created = updated = 0
for node in self._section(root, 'tarifgruppeSetZuordnungen'):
tarifgruppe = tarifgruppe_map.get(
self._tasy_text(node, 'tarifgruppe'))
tarifset = tarifset_map.get(self._tasy_text(node, 'tarifset'))
von = self._tasy_text(node, 'vonDatum')
if not tarifgruppe or not tarifset or not von:
continue
basis, pos = kombinationen.get(
self._tasy_text(node, 'verrechBasPosKombination'),
('', ''))
vals = {
'tarifgruppe_id': tarifgruppe.id,
'tarifset_id': tarifset.id,
'basis_typ': basis,
'pos_typ': pos,
'traegergruppe': self._tasy_text(node, 'traegergruppe'),
'von_datum': fields.Date.to_date(von),
'bis_datum': fields.Date.to_date(
self._tasy_text(node, 'bisDatum') or '2999-12-31'),
'tasy_xml_id': node.get('xmlId'),
}
rec = zuord_obj.search([
('tarifgruppe_id', '=', tarifgruppe.id),
('tarifset_id', '=', tarifset.id),
('basis_typ', '=', basis),
('pos_typ', '=', pos),
('von_datum', '=', vals['von_datum'])], limit=1)
if rec:
rec.write(vals)
updated += 1
else:
zuord_obj.create(vals)
created += 1
protocol.append('Tarifgruppe-Tarifset-Zuordnungen: '
'%d neu, %d aktualisiert' % (created, updated))
# 4) Zu-/Abschlag-Zuordnungen (m2m auf Tarifsets/Tarifgruppen).
verknuepft = 0
for node in self._section(root, 'zuAbschlagTarifsetZuordnungen'):
zuabschlag = zuabschlag_map.get(
self._tasy_text(node, 'zuAbschlag'))
if not zuabschlag:
continue
tarifset = tarifset_map.get(
self._tasy_text(node, 'tarifsetByTarifsetId'))
tarifgruppen = [
tarifgruppe_map[r] for r in self._tasy_texts(
node, 'tarifgruppe') if r in tarifgruppe_map]
if tarifset:
zuabschlag.tarifset_ids = [(4, tarifset.id)]
for tarifgruppe in tarifgruppen:
zuabschlag.tarifgruppe_ids = [(4, tarifgruppe.id)]
verknuepft += 1
protocol.append('Zu-/Abschlag-Zuordnungen: %d verknüpft'
% verknuepft)
self.protocol = '\n'.join(protocol)
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'type': 'success',
'message': _('Tarifsystem-Katalog importiert.'),
'next': {'type': 'ir.actions.act_window_close'},
},
}