# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. from decimal import Decimal from sql import Literal, Null from sql.aggregate import Sum from sql.conditionals import Coalesce from trytond import backend from trytond.i18n import gettext from trytond.model import ModelSQL, ValueMixin, fields from trytond.modules.company.model import ( CompanyMultiValueMixin, CompanyValueMixin) from trytond.modules.currency.fields import Monetary from trytond.pool import Pool, PoolMeta from trytond.pyson import Eval, If from trytond.tools import grouped_slice, reduce_ids from trytond.transaction import Transaction from trytond.modules.ncf_manager.ncf_manager import NCF_TYPE import logging logger = logging.getLogger(__name__) def _backend_name(): name = backend.name return name() if callable(name) else name customer_ncf_type = fields.Selection( NCF_TYPE, "Customer NCF Type", sort=False ) supplier_ncf_type = fields.Selection( NCF_TYPE, "Supplier NCF Type", sort=False ) class Party(metaclass=PoolMeta): __name__ = 'party.party' customer_ncf_type = fields.MultiValue(customer_ncf_type) supplier_ncf_type = fields.MultiValue(supplier_ncf_type) @classmethod def multivalue_model(cls, field): pool = Pool() if field in {'customer_ncf_type', 'supplier_ncf_type'}: return pool.get('party.party.ncf_type') return super(Party, cls).multivalue_model(field) @classmethod def default_customer_ncf_type(cls, **pattern): pool = Pool() Configuration = pool.get('account.configuration') config = Configuration(1) return config.get_multivalue( 'default_customer_ncf_type', **pattern) @classmethod def default_supplier_ncf_type(cls, **pattern): pool = Pool() Configuration = pool.get('account.configuration') config = Configuration(1) return config.get_multivalue( 'default_supplier_ncf_type', **pattern) class PartyNcfType(ModelSQL, ValueMixin): "Party Ncf Type" __name__ = 'party.party.ncf_type' party = fields.Many2One( 'party.party', "Party", ondelete='CASCADE') customer_ncf_type = customer_ncf_type supplier_ncf_type = supplier_ncf_type @classmethod def __register__(cls, module_name): """ Migrate legacy party NCF fields safely. Old structure: party_party_ncf_type.customer_ncf_type = ncf.manager.sequence.id party_party_ncf_type.supplier_ncf_type = ncf.manager.sequence.id New structure: party_party_ncf_type.customer_ncf_type = NCF type code party_party_ncf_type.supplier_ncf_type = NCF type code The old integer columns must be renamed before super().__register__(), otherwise Tryton tries to convert int4 -> varchar directly. """ cls._ncf_prepare_legacy_party_columns() super().__register__(module_name) cls._ncf_migrate_party_ncf_types_from_legacy_columns() @classmethod def _ncf_column_exists(cls, cursor, table_name, column_name): if _backend_name() == 'sqlite': cursor.execute(f'PRAGMA table_info("{table_name}")') return any(row[1] == column_name for row in cursor.fetchall()) cursor.execute(""" SELECT 1 FROM information_schema.columns WHERE table_name = %s AND column_name = %s """, (table_name, column_name)) return cursor.fetchone() is not None @classmethod def _ncf_column_udt_name(cls, cursor, table_name, column_name): if _backend_name() == 'sqlite': cursor.execute(f'PRAGMA table_info("{table_name}")') for row in cursor.fetchall(): if row[1] == column_name: return (row[2] or '').lower() return None cursor.execute(""" SELECT udt_name FROM information_schema.columns WHERE table_name = %s AND column_name = %s """, (table_name, column_name)) row = cursor.fetchone() return row[0] if row else None @classmethod def _ncf_prepare_legacy_party_columns(cls): """ Rename old integer columns before Tryton creates the new Selection columns. """ cursor = Transaction().connection.cursor() table_name = 'party_party_ncf_type' columns = ( ('customer_ncf_type', 'customer_ncf_type_legacy_sequence_id'), ('supplier_ncf_type', 'supplier_ncf_type_legacy_sequence_id'), ) for old_column, legacy_column in columns: if not cls._ncf_column_exists(cursor, table_name, old_column): continue if cls._ncf_column_exists(cursor, table_name, legacy_column): continue udt_name = cls._ncf_column_udt_name( cursor, table_name, old_column) if udt_name in {'int2', 'int4', 'int8', 'integer'}: logger.info( "Renaming legacy %s.%s integer column to %s before " "Tryton creates the new Selection column", table_name, old_column, legacy_column, ) cursor.execute(f""" ALTER TABLE {table_name} RENAME COLUMN {old_column} TO {legacy_column} """) @classmethod def _ncf_migrate_party_ncf_types_from_legacy_columns(cls): """ Convert legacy party NCF sequence IDs into real NCF type values. Example: customer_ncf_type_legacy_sequence_id = 1 ncf_manager_sequence.id = 1 ncf_manager_sequence.ncf_type = '31' Result: customer_ncf_type = '31' """ cursor = Transaction().connection.cursor() table_name = 'party_party_ncf_type' columns = ( ('customer_ncf_type', 'customer_ncf_type_legacy_sequence_id'), ('supplier_ncf_type', 'supplier_ncf_type_legacy_sequence_id'), ) for new_column, legacy_column in columns: if not cls._ncf_column_exists(cursor, table_name, legacy_column): logger.info( "No legacy column %s found in %s. Skipping migration.", legacy_column, table_name, ) continue logger.info( "Migrating party field %s from legacy sequence ID to " "fiscal NCF type code", new_column, ) cursor.execute(f""" UPDATE {table_name} AS party_ncf SET {new_column} = seq.ncf_type FROM ncf_manager_sequence AS seq WHERE party_ncf.{legacy_column} IS NOT NULL AND party_ncf.{legacy_column} = seq.id AND seq.ncf_type IS NOT NULL AND seq.ncf_type <> '' AND ( party_ncf.{new_column} IS NULL OR party_ncf.{new_column} = '' OR party_ncf.{new_column} <> seq.ncf_type ) """) logger.info( "Migrated %s rows for party field %s", cursor.rowcount, new_column, ) cursor.execute(f""" SELECT COUNT(*) FROM {table_name} AS party_ncf LEFT JOIN ncf_manager_sequence AS seq ON party_ncf.{legacy_column} = seq.id WHERE party_ncf.{legacy_column} IS NOT NULL AND seq.id IS NULL """) unmatched = cursor.fetchone()[0] if unmatched: logger.warning( "%s rows in %s have legacy values in %s that do not " "match any ncf_manager_sequence.id", unmatched, table_name, legacy_column, )