open_workshop/models/ows_models.py

581 lines
22 KiB
Python

# -*- coding: utf-8 -*-
# ows_models.py
# Part of Odoo Open Workshop
from odoo import models, fields, api
from markupsafe import escape as html_escape
import logging
_logger = logging.getLogger(__name__)
_logger.info("✅ ows_models.py geladen")
class HREmployee(models.Model):
_inherit = 'hr.employee'
@api.model
def anonymize_for_testsystem(self):
"""Benennt Admin-Angestellten um und archiviert alle anderen für das Testsystem."""
admin_user = self.env['res.users'].search([('name', '=', 'Administrator')], limit=1)
_logger.info(f"[OWS] Administrator-Benutzer gefunden: {admin_user.name} (ID: {admin_user.id})")
admin_employee = self.search([('user_id', '=', admin_user.id)], limit=1)
if admin_employee:
admin_employee.write({
'name': 'TESTSYSTEM',
'job_title': 'Testumgebung',
'work_email': False,
'work_phone': False,
})
_logger.info("[OWS] Admin-Angestellter wurde umbenannt.")
else:
_logger.warning("[OWS] Kein Angestellter für user_admin gefunden.")
# Alle anderen Angestellten archivieren
other_employees = self.search([('id', '!=', admin_employee.id)])
other_employees.write({'active': False})
_logger.info("[OWS] %d Angestellte archiviert.", len(other_employees))
class ResPartner(models.Model):
_inherit = 'res.partner'
_logger.info("✅ ows ResPartner geladen")
ows_user_id = fields.One2many('ows.user', 'partner_id', string="OWS Benutzerdaten")
# Alte Felder (weiterhin sichtbar für externe Programme)
birthday = fields.Date(
string="Geburtstag",
compute='_compute_ows_user_fields',
inverse='_inverse_birthday',
store=False
)
rfid_card = fields.Text(
string="RFID Card ID",
compute='_compute_ows_user_fields',
inverse='_inverse_rfid_card',
store=False
)
security_briefing = fields.Boolean(
string="Haftungsausschluss",
compute='_compute_ows_user_fields',
inverse='_inverse_security_briefing',
store=False
)
security_id = fields.Text(
string="Haftungsausschluss ID",
compute='_compute_ows_user_fields',
inverse='_inverse_security_id',
store=False
)
# Neue direkte vvow_* Felder
vvow_birthday = fields.Date(string="Geburtstag (vvow)")
vvow_rfid_card = fields.Text(string="RFID Card ID (vvow)")
vvow_security_briefing = fields.Boolean(string="Haftungsausschluss (vvow)")
vvow_security_id = fields.Text(string="Haftungsausschluss ID (vvow)")
@api.depends('ows_user_id')
def _compute_ows_user_fields(self):
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
partner.birthday = user.birthday if user else False
partner.rfid_card = user.rfid_card if user else False
partner.security_briefing = user.security_briefing if user else False
partner.security_id = user.security_id if user else False
def _inverse_birthday(self):
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
if user:
user.birthday = partner.birthday
def _inverse_rfid_card(self):
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
if user:
user.rfid_card = partner.rfid_card
def _inverse_security_briefing(self):
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
if user:
user.security_briefing = partner.security_briefing
def _inverse_security_id(self):
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
if user:
user.security_id = partner.security_id
@api.model_create_multi
def create(self, vals_list):
partners = super().create(vals_list)
for vals, partner in zip(vals_list, partners):
self.env['ows.user'].create({
'partner_id': partner.id,
'birthday': vals.get('birthday') or vals.get('vvow_birthday'),
'rfid_card': vals.get('rfid_card') or vals.get('vvow_rfid_card'),
'security_briefing': vals.get('security_briefing') or vals.get('vvow_security_briefing'),
'security_id': vals.get('security_id') or vals.get('vvow_security_id'),
})
return partners
def write(self, vals):
res = super().write(vals)
for partner in self:
user = partner.ows_user_id[0] if partner.ows_user_id else False
if not user:
continue
# Synchronisation alt -> user
if 'birthday' in vals:
user.birthday = vals['birthday']
if 'rfid_card' in vals:
user.rfid_card = vals['rfid_card']
if 'security_briefing' in vals:
user.security_briefing = vals['security_briefing']
if 'security_id' in vals:
user.security_id = vals['security_id']
# Synchronisation vvow_* -> user + alt
if 'vvow_birthday' in vals:
user.birthday = vals['vvow_birthday']
partner.birthday = vals['vvow_birthday']
if 'vvow_rfid_card' in vals:
user.rfid_card = vals['vvow_rfid_card']
partner.rfid_card = vals['vvow_rfid_card']
if 'vvow_security_briefing' in vals:
user.security_briefing = vals['vvow_security_briefing']
partner.security_briefing = vals['vvow_security_briefing']
if 'vvow_security_id' in vals:
user.security_id = vals['vvow_security_id']
partner.security_id = vals['vvow_security_id']
return res
machine_access_ids = fields.One2many(
'ows.machine.access',
'partner_id',
string='Maschinenfreigaben'
)
machine_access_html = fields.Html(
string="Maschinenfreigabe",
compute="_compute_machine_access_html",
sanitize=False
)
@api.depends('machine_access_ids')
def _compute_machine_access_html(self):
areas = self.env['ows.machine.area'].search([], order="name")
for partner in self:
html = ""
for area in areas:
html += f"""
<div class="o_form_sheet">
<h3 class="o_form_label">{area.name}</h3>
<table class="table table-sm table-bordered o_form_table">
<thead>
<tr>
<th>Maschine</th>
<th>Status</th>
<th>Datum</th>
<th>Gültig bis</th>
</tr>
</thead>
<tbody>
"""
machines = self.env['ows.machine'].search([('area_id', '=', area.id)], order="name")
for machine in machines:
access = self.env['ows.machine.access'].search([
('partner_id', '=', partner.id),
('machine_id', '=', machine.id),
], limit=1)
icon = '<span class="text-success fa fa-check"/>' if access else '<span class="text-danger fa fa-times"/>'
date_granted = access.date_granted.strftime('%Y-%m-%d') if access and access.date_granted else "-"
date_expiry = access.date_expiry.strftime('%Y-%m-%d') if access and access.date_expiry else "-"
html += f"""
<tr>
<td>{machine.name}</td>
<td>{icon}</td>
<td>{date_granted}</td>
<td>{date_expiry}</td>
</tr>
"""
html += "</tbody></table></div>"
partner.machine_access_html = html
@api.model
def migrate_existing_partners(self):
"""
Erstellt für alle vorhandenen res.partner einen ows.user,
wenn noch keiner existiert, und übernimmt alte vvow_* Felder.
Führt am Ende automatisch einen commit durch.
Verwendung in der odoo shell:
env['res.partner'].migrate_existing_partners()
env['ows.user'].search_count([])
env['ows.user'].search([]).mapped('partner_id.name')
"""
migrated = 0
skipped = 0
partners = self.env['res.partner'].search([])
for partner in partners:
if partner.ows_user_id:
skipped += 1
continue
# Werte lesen (werden evtl. durch _inherit hinzugefügt)
vals = {
'partner_id': partner.id,
'birthday': getattr(partner, 'vvow_birthday', False),
'rfid_card': getattr(partner, 'vvow_rfid_card', False),
'security_briefing': getattr(partner, 'vvow_security_briefing', False),
'security_id': getattr(partner, 'vvow_security_id', False),
}
self.env['ows.user'].create(vals)
migrated += 1
_logger.info(f"Erzeuge ows.user für Partner {partner.id} ({partner.name}) mit Werten: {vals}")
_logger.info(f"[OWS Migration] ✅ Migriert: {migrated}, ❌ Übersprungen: {skipped}")
# 🔐 Commit am Ende
self.env.cr.commit()
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': "Migration abgeschlossen",
'message': f"{migrated} Partner migriert, {skipped} übersprungen.",
'sticky': False,
}
}
@api.model
def migrate_machine_access_from_old_fields(self):
"""
Migriert alte vvow_* Boolean-Felder in strukturierte ows.machine.access Einträge.
Das Freigabe-Datum wird aus dem Änderungsverlauf (mail.message.tracking_value_ids) extrahiert.
"""
mapping = [
('vvow_holz_sander', 'bandschleifer'),
('vvow_holz_felder_bandsaw', 'bandsaege_holz'),
('vvow_holz_felder_jointer', 'dickenhobel'),
('vvow_holz_felder_mill', 'felder_fraese'),
('vvow_holz_felder_tablesaw', 'formatkreissaege'),
('vvow_holz_mill', 'tischfraese'),
('vvow_holz_lathe', 'drechselbank'),
('vvow_holz_domino', 'festool_domino'),
('vvow_holz_duoduebler', 'maffel_duo'),
('vvow_holz_lamello', 'lamello_zeta_p2'),
('vvow_fablab_laser_sabko', 'sabako_laser'),
('vvow_fablab_3dprint_delta', '3d_delta'),
('vvow_fablab_3dprint_prusa', 'prusa'),
('vvow_fablab_3dprint_prusa_mmu', 'prusa_mmu'),
('vvow_fablab_cnc_beamicon', 'cnc_beamicon'),
('vvow_metall_welding_mig', 'mig_mag'),
('vvow_metall_welding_wig', 'wig'),
('vvow_metall_welding_misc', 'schweissen_allgemein'),
('vvow_metall_lathe', 'drehbank'),
('vvow_metall_bandsaw', 'bandsaege_metall'),
('vvow_metall_mill_fp2', 'fraese'),
('vvow_metall_bende', 'abkantbank'),
('vvow_metall_chop_saw', 'kreissaege_metall'),
]
MachineAccess = self.env['ows.machine.access']
MailMessage = self.env['mail.message']
TrackingValue = self.env['mail.tracking.value']
count_created = 0
for partner in self.search([]):
for field_name, machine_code in mapping:
if not getattr(partner, field_name, False):
continue
machine = self.env['ows.machine'].search([('code', '=', machine_code)], limit=1)
if not machine:
continue
# Änderungsverlauf durchsuchen: Wann wurde das Feld auf True gesetzt?
tracking = TrackingValue.search([
('field', '=', field_name),
('field_type', '=', 'boolean'),
('new_value_integer', '=', 1),
('mail_message_id.model', '=', 'res.partner'),
('mail_message_id.res_id', '=', partner.id),
], order='id ASC', limit=1)
date_granted = tracking.mail_message_id.date if tracking else fields.Datetime.now()
if not MachineAccess.search([('partner_id', '=', partner.id), ('machine_id', '=', machine.id)]):
MachineAccess.create({
'partner_id': partner.id,
'machine_id': machine.id,
'date_granted': date_granted,
})
count_created += 1
_logger.info(f"[OWS Migration] ✅ Maschinenfreigaben erstellt: {count_created}")
self.env.cr.commit()
@api.model
def archive_partners_without_users(self):
"""
Archiviert alle Partner (res.partner), die keine Benutzer (res.users) sind.
"""
Partner = self.env['res.partner']
User = self.env['res.users']
# IDs aller Partner, die ein Benutzerkonto haben
user_partner_ids = User.search([]).mapped('partner_id').ids
# Alle Partner ohne Benutzerkonto
partners_to_archive = Partner.search([
('id', 'not in', user_partner_ids),
('active', '=', True),
])
count = len(partners_to_archive)
partners_to_archive.write({'active': False})
for p in partners_to_archive:
_logger.debug(f"[OWS] Archiviert Partner: {p.name} (ID {p.id})")
_logger.info(f"[OWS] Archiviert {count} Partner ohne Benutzerkonto.")
self.env.cr.commit()
class OwsUser(models.Model):
_name = 'ows.user'
_description = 'OWS: Benutzerdaten'
_table = 'ows_user'
_logger.info("✅ ows_user geladen")
partner_id = fields.Many2one(
'res.partner',
required=True,
ondelete='cascade',
string='Kontakt'
)
birthday = fields.Date('Geburtstag')
rfid_card = fields.Text('RFID Card ID')
security_briefing = fields.Boolean('Haftungsausschluss', default=False)
security_id = fields.Text('Haftungsausschluss ID')
_sql_constraints = [
('partner_unique', 'unique(partner_id)', 'Jeder Partner darf nur einen OWS-Datensatz haben.')
]
AVAILABLE_COLORS = [
('#000000', 'schwarz'),
('#ff0000', 'Rot'),
('#E91E63', 'Pink'),
('#9C27B0', 'Lila'),
('#3F51B5', 'Indigo'),
('#0000ff', 'Blau'),
('#008000', 'Grün'),
('#ffff00', 'Gelb'),
('#FF9800', 'Orange'),
('#795548', 'Braun'),
('#ffffff', 'Weiss'),
]
class OwsMachineArea(models.Model):
_name = 'ows.machine.area'
_table = 'ows_machine_area'
_description = 'OWS: Maschinenbereich'
_order = 'name'
name = fields.Char(string="Name", required=True, translate=True)
color_hex = fields.Selection(
selection=AVAILABLE_COLORS,
string="Farbe (Hex)",
required=True,
)
color_hex_value = fields.Char(
string="Farbcode",
compute='_compute_color_hex_value',
store=False
)
color_name = fields.Char(
string="Farbname",
compute='_compute_color_name',
store=False
)
@api.depends('color_hex')
def _compute_color_hex_value(self):
for rec in self:
rec.color_hex_value = rec.color_hex or ''
@api.depends('color_hex')
def _compute_color_name(self):
label_dict = dict(AVAILABLE_COLORS)
for rec in self:
rec.color_name = label_dict.get(rec.color_hex, 'Unbekannt')
class OwsMachine(models.Model):
_name = 'ows.machine'
_table = 'ows_machine'
_description = 'OWS: Maschine'
name = fields.Char(required=True, translate=True)
code = fields.Char(required=True, help="Eindeutiger Kurzcode, z.B. 'lasercutter'")
category = fields.Selection([
('green', 'Kategorie 1: grün'),
('yellow', 'Kategorie 2: gelb'),
('red', 'Kategorie 3: rot'),
], string="Sicherheitskategorie", required=True, default='red', help="Sicherheitsrelevante Maschinenkategorie:\n"
"- grün: keine Einweisungspflicht\n"
"- gelb: empfohlene Einweisung\n"
"- rot: Einweisung zwingend erforderlich")
category_icon = fields.Char(string="Kategorie-Symbol", compute="_compute_category_icon", store=False)
@api.depends('category')
def _compute_category_icon(self):
for rec in self:
icon_map = {
'green': '🟢',
'yellow': '🟡',
'red': '🔴',
}
rec.category_icon = icon_map.get(rec.category, '')
description = fields.Text()
active = fields.Boolean(default=True)
area_id = fields.Many2one('ows.machine.area', string='Bereich', help="Bereich, in dem die Maschine oder das Gerät steht.")
product_ids = fields.One2many('ows.machine.product', 'machine_id', string="Nutzungsprodukte")
product_names = fields.Char(string="Liste der Nutzungsprodukte", compute="_compute_product_using_names", store=False,)
training_ids = fields.One2many('ows.machine.training', 'machine_id', string="Einweisungsprodukte")
training_names = fields.Char(string="Liste der Einweisungsprodukte", compute="_compute_product_training_names", store=False,)
storage_location = fields.Char(string="Lagerort", help="Lagerort der Maschine oder des Geräts.")
purchase_price = fields.Float(string="Kaufpreis", help="Kaufpreis der Maschine oder des Geräts.")
purchase_date = fields.Date(string="Kaufdatum", help="Kaufdatum der Maschine oder des Geräts.")
@api.depends('product_ids.product_id.name')
def _compute_product_using_names(self):
for machine in self:
names = machine.product_ids.mapped('product_id.name')
machine.product_names = ", ".join(names)
@api.depends('training_ids.training_id.name')
def _compute_product_training_names(self):
for machine in self:
names = machine.training_ids.mapped('training_id.name')
machine.training_names = ", ".join(names)
_sql_constraints = [
('code_unique', 'unique(code)', 'Maschinencode muss eindeutig sein.')
]
def name_get(self):
return [(rec.id, f"{rec.name} ({rec.code})") for rec in self]
@api.model
def get_access_list_grouped(self, partner_id):
"""
Gibt eine gruppierte Liste von Maschinenzugängen für einen bestimmten Partner zurück. Diese Funktion wird in
Odoo POS Frontend verwendet um die Ansicht zu erzeugen auf Welche Maschinen der Partner Zugriff hat.
Für einen gegebenen Partner (über die partner_id) werden alle Maschinenbereiche (areas) abgefragt.
Für jeden Bereich wird geprüft, auf welche Maschinen der Partner Zugriff hat. Das Ergebnis wird
als Liste von Bereichen mit jeweils zugehörigen Maschinen und Zugriffsstatus zurückgegeben.
Zusätzlich werden sicherheitsrelevante Informationen des Partners (wie Sicherheitsunterweisung,
Sicherheits-ID, RFID-Karte und Geburtstag) aus dem zugehörigen ows_user ermittelt und mitgeliefert.
Args:
partner_id (int): Die ID des Partners, für den die Zugriffsübersicht erstellt werden soll.
Returns:
dict: Ein Dictionary mit folgenden Schlüsseln:
- 'access_by_area': Liste von Bereichen mit Maschinen und Zugriffsstatus.
- 'security_briefing': Sicherheitsunterweisung des Nutzers (bool oder False).
- 'security_id': Sicherheits-ID des Nutzers (str oder '').
- 'rfid_card': RFID-Kartennummer des Nutzers (str oder '').
- 'birthday': Geburtstag des Nutzers (str oder '').
"""
partner = self.env['res.partner'].browse(partner_id)
areas = self.env['ows.machine.area'].search([], order="name")
_logger.info("Access RPC called with partner_id=%s", partner_id)
access_by_area = []
for area in areas:
machines = self.search([('area_id', '=', area.id), ('category', '=', 'red')], order="name")
machine_list = []
for machine in machines:
has_access = bool(self.env['ows.machine.access'].search([
('partner_id', '=', partner_id),
('machine_id', '=', machine.id),
], limit=1))
machine_list.append({
'name': machine.name,
'has_access': has_access,
})
if machine_list:
access_by_area.append({
'area': area.name,
'color_hex': area.color_hex or '#000000',
'machines': machine_list
})
user = partner.ows_user_id[:1]
return {
'access_by_area': access_by_area,
'security_briefing': user.security_briefing if user else False,
'security_id': user.security_id if user else '',
'rfid_card': user.rfid_card if user else '',
'birthday': user.birthday if user else '',
}
class OwsMachineAccess(models.Model):
_name = 'ows.machine.access'
_table = 'ows_machine_access'
_description = 'OWS: Maschinenfreigabe'
_order = 'partner_id, machine_id'
partner_id = fields.Many2one('res.partner', required=True, ondelete='cascade')
machine_id = fields.Many2one('ows.machine', required=True)
date_granted = fields.Date(default=fields.Date.today)
date_expiry = fields.Date(string="Ablaufdatum")
granted_by_pos = fields.Boolean(default=True)
_sql_constraints = [
('partner_machine_unique', 'unique(partner_id, machine_id)', 'Der Kunde hat diese Freigabe bereits.')
]
class OwsMachineProduct(models.Model):
_name = 'ows.machine.product'
_table = 'ows_machine_product'
_description = 'OWS: Zuordnung Produkt der Nutzung zu der Maschine'
product_id = fields.Many2one('product.product', required=True, domain=[('available_in_pos', '=', True)], ondelete='cascade')
machine_id = fields.Many2one('ows.machine', required=True, ondelete='cascade')
class OwsMachineTraining(models.Model):
_name = 'ows.machine.training'
_table = 'ows_machine_training'
_description = 'OWS: Zuordnung Produkt der Einweisung zu der Maschine'
training_id = fields.Many2one('product.product', required=True, domain=[('available_in_pos', '=', True)], ondelete='cascade')
machine_id = fields.Many2one('ows.machine', required=True, ondelete='cascade')