283 lines
10 KiB
Python
283 lines
10 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.
|
||
|
|
import logging
|
||
|
|
|
||
|
|
from trytond import backend
|
||
|
|
from trytond.model import fields, ModelSQL
|
||
|
|
from trytond.modules.company.model import CompanyValueMixin
|
||
|
|
from trytond.pool import PoolMeta, Pool
|
||
|
|
from trytond.transaction import Transaction
|
||
|
|
from trytond.modules.ncf_manager.ncf_manager import NCF_TYPE
|
||
|
|
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
def _backend_name():
|
||
|
|
name = backend.name
|
||
|
|
return name() if callable(name) else name
|
||
|
|
|
||
|
|
|
||
|
|
class Configuration(metaclass=PoolMeta):
|
||
|
|
'Account Configuration'
|
||
|
|
__name__ = 'account.configuration'
|
||
|
|
|
||
|
|
default_customer_ncf_type = fields.MultiValue(fields.Selection(
|
||
|
|
NCF_TYPE,
|
||
|
|
"Default Customer NCF Type",
|
||
|
|
sort=False,
|
||
|
|
help="Default customer NCF type for new parties."
|
||
|
|
))
|
||
|
|
|
||
|
|
default_supplier_ncf_type = fields.MultiValue(fields.Selection(
|
||
|
|
NCF_TYPE,
|
||
|
|
"Default Supplier NCF Type",
|
||
|
|
sort=False,
|
||
|
|
help="Default supplier NCF type for new parties."
|
||
|
|
))
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def multivalue_model(cls, field):
|
||
|
|
pool = Pool()
|
||
|
|
if field in {'default_customer_ncf_type', 'default_supplier_ncf_type'}:
|
||
|
|
return pool.get('account.configuration.default_ncf_type')
|
||
|
|
return super(Configuration, cls).multivalue_model(field)
|
||
|
|
|
||
|
|
|
||
|
|
class DefaultNcfType(ModelSQL, CompanyValueMixin):
|
||
|
|
"Account Configuration Default NCF Type"
|
||
|
|
__name__ = 'account.configuration.default_ncf_type'
|
||
|
|
|
||
|
|
default_customer_ncf_type = fields.Selection(
|
||
|
|
NCF_TYPE,
|
||
|
|
"Default Customer NCF Type",
|
||
|
|
sort=False
|
||
|
|
)
|
||
|
|
|
||
|
|
default_supplier_ncf_type = fields.Selection(
|
||
|
|
NCF_TYPE,
|
||
|
|
"Default Supplier NCF Type",
|
||
|
|
sort=False
|
||
|
|
)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def __register__(cls, module_name):
|
||
|
|
"""
|
||
|
|
Register the model and migrate legacy configuration values safely.
|
||
|
|
|
||
|
|
Old structure:
|
||
|
|
account_configuration_default_ncf_type.default_customer_ncf_type
|
||
|
|
account_configuration_default_ncf_type.default_supplier_ncf_type
|
||
|
|
|
||
|
|
could be stored as integer Many2One values pointing to:
|
||
|
|
ncf.manager.sequence.id
|
||
|
|
|
||
|
|
New structure:
|
||
|
|
both fields are Selection values and must store the fiscal NCF type
|
||
|
|
code directly, for example: '31', '32', '41', '43', '99'.
|
||
|
|
|
||
|
|
The integer columns must be renamed before super().__register__(),
|
||
|
|
otherwise Tryton tries to migrate int4 -> varchar directly and the
|
||
|
|
later SQL migration fails when comparing an integer column with ''.
|
||
|
|
"""
|
||
|
|
cls._ncf_prepare_legacy_configuration_columns()
|
||
|
|
super().__register__(module_name)
|
||
|
|
cls._ncf_migrate_configuration_defaults()
|
||
|
|
|
||
|
|
@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_configuration_columns(cls):
|
||
|
|
"""
|
||
|
|
Rename old integer columns before Tryton creates the new Selection
|
||
|
|
varchar columns.
|
||
|
|
"""
|
||
|
|
cursor = Transaction().connection.cursor()
|
||
|
|
table_name = 'account_configuration_default_ncf_type'
|
||
|
|
|
||
|
|
columns = (
|
||
|
|
(
|
||
|
|
'default_customer_ncf_type',
|
||
|
|
'default_customer_ncf_type_legacy_sequence_id',
|
||
|
|
),
|
||
|
|
(
|
||
|
|
'default_supplier_ncf_type',
|
||
|
|
'default_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_configuration_defaults(cls):
|
||
|
|
"""
|
||
|
|
Convert old configuration values from ncf.manager.sequence.id to
|
||
|
|
ncf.manager.sequence.ncf_type.
|
||
|
|
"""
|
||
|
|
cursor = Transaction().connection.cursor()
|
||
|
|
table_name = 'account_configuration_default_ncf_type'
|
||
|
|
|
||
|
|
columns = (
|
||
|
|
(
|
||
|
|
'default_customer_ncf_type',
|
||
|
|
'default_customer_ncf_type_legacy_sequence_id',
|
||
|
|
),
|
||
|
|
(
|
||
|
|
'default_supplier_ncf_type',
|
||
|
|
'default_supplier_ncf_type_legacy_sequence_id',
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
valid_ncf_type_codes = tuple(
|
||
|
|
value for value, _label in NCF_TYPE if value
|
||
|
|
)
|
||
|
|
|
||
|
|
for new_column, legacy_column in columns:
|
||
|
|
if cls._ncf_column_exists(cursor, table_name, legacy_column):
|
||
|
|
logger.info(
|
||
|
|
"Migrating account configuration field %s from legacy "
|
||
|
|
"ncf.manager.sequence id column to NCF type code",
|
||
|
|
new_column,
|
||
|
|
)
|
||
|
|
|
||
|
|
cursor.execute(f"""
|
||
|
|
UPDATE {table_name} AS config
|
||
|
|
SET {new_column} = seq.ncf_type
|
||
|
|
FROM ncf_manager_sequence AS seq
|
||
|
|
WHERE config.{legacy_column} IS NOT NULL
|
||
|
|
AND config.{legacy_column} = seq.id
|
||
|
|
AND seq.ncf_type IS NOT NULL
|
||
|
|
AND seq.ncf_type <> ''
|
||
|
|
AND (
|
||
|
|
config.{new_column} IS NULL
|
||
|
|
OR config.{new_column} = ''
|
||
|
|
OR config.{new_column} <> seq.ncf_type
|
||
|
|
)
|
||
|
|
""")
|
||
|
|
|
||
|
|
logger.info(
|
||
|
|
"Migrated %s rows for account configuration field %s",
|
||
|
|
cursor.rowcount,
|
||
|
|
new_column,
|
||
|
|
)
|
||
|
|
|
||
|
|
cursor.execute(f"""
|
||
|
|
SELECT COUNT(*)
|
||
|
|
FROM {table_name} AS config
|
||
|
|
LEFT JOIN ncf_manager_sequence AS seq
|
||
|
|
ON config.{legacy_column} = seq.id
|
||
|
|
WHERE config.{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,
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Fallback for databases where Tryton already managed to create the
|
||
|
|
# column as varchar but old numeric sequence ids are still stored as
|
||
|
|
# text. This intentionally skips already-valid NCF type codes to
|
||
|
|
# avoid converting a value such as '31' again.
|
||
|
|
if not cls._ncf_column_exists(cursor, table_name, new_column):
|
||
|
|
continue
|
||
|
|
if _backend_name() == 'sqlite':
|
||
|
|
logger.info(
|
||
|
|
"Skipping PostgreSQL-style text fallback migration for "
|
||
|
|
"%s.%s on SQLite",
|
||
|
|
table_name,
|
||
|
|
new_column,
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
|
||
|
|
udt_name = cls._ncf_column_udt_name(cursor, table_name, new_column)
|
||
|
|
if udt_name not in {'varchar', 'text', 'bpchar'}:
|
||
|
|
logger.info(
|
||
|
|
"Skipping in-place migration for %s.%s because column "
|
||
|
|
"type is %s",
|
||
|
|
table_name,
|
||
|
|
new_column,
|
||
|
|
udt_name,
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
|
||
|
|
logger.info(
|
||
|
|
"Checking account configuration field %s for legacy numeric "
|
||
|
|
"values stored as text",
|
||
|
|
new_column,
|
||
|
|
)
|
||
|
|
|
||
|
|
cursor.execute(f"""
|
||
|
|
UPDATE {table_name} AS config
|
||
|
|
SET {new_column} = seq.ncf_type
|
||
|
|
FROM ncf_manager_sequence AS seq
|
||
|
|
WHERE config.{new_column} IS NOT NULL
|
||
|
|
AND config.{new_column} <> ''
|
||
|
|
AND config.{new_column} ~ '^[0-9]+$'
|
||
|
|
AND config.{new_column} != ALL(%s)
|
||
|
|
AND config.{new_column}::integer = seq.id
|
||
|
|
AND seq.ncf_type IS NOT NULL
|
||
|
|
AND seq.ncf_type <> ''
|
||
|
|
""", (list(valid_ncf_type_codes),))
|
||
|
|
|
||
|
|
logger.info(
|
||
|
|
"Migrated %s text rows for account configuration field %s",
|
||
|
|
cursor.rowcount,
|
||
|
|
new_column,
|
||
|
|
)
|