302 lines
13 KiB
Python
302 lines
13 KiB
Python
# Part of the odoo-at-payroll project. License: LGPL-3.
|
|
from lxml import etree
|
|
|
|
import base64
|
|
|
|
from odoo import _, api, fields, models
|
|
from odoo.exceptions import UserError
|
|
|
|
|
|
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()
|
|
parser = etree.XMLParser(resolve_entities=False, no_network=True)
|
|
data = self.file 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)
|
|
|
|
@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'},
|
|
},
|
|
} |