249 lines
8.1 KiB
Python
249 lines
8.1 KiB
Python
|
# 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,
|
||
|
|
)
|