# Part of the odoo-at-payroll project. License: LGPL-3. from lxml import etree 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) return etree.fromstring(self.file, 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) old_value = old.parameter_value if old 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'}, }, }