A R R R Associates
227 lines
9.1 KiB
Python
227 lines
9.1 KiB
Python
"""services phase s1: categories, recurrence, sort order, applicability flags
|
|
|
|
Revision ID: 20260422_services_phase_s1
|
|
Revises: 20260420_services_catalogue_firm_toggle
|
|
Create Date: 2026-04-22 20:00:00.000000
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "20260422_services_phase_s1"
|
|
down_revision = "20260420_services_catalogue_firm_toggle"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _inspector():
|
|
return sa.inspect(op.get_bind())
|
|
|
|
|
|
def _has_table(name: str) -> bool:
|
|
return name in _inspector().get_table_names()
|
|
|
|
|
|
def _has_column(table_name: str, column_name: str) -> bool:
|
|
try:
|
|
cols = _inspector().get_columns(table_name)
|
|
except Exception:
|
|
return False
|
|
return any(col.get("name") == column_name for col in cols)
|
|
|
|
|
|
def _has_index(table_name: str, index_name: str) -> bool:
|
|
try:
|
|
indexes = _inspector().get_indexes(table_name)
|
|
except Exception:
|
|
return False
|
|
return any(ix.get("name") == index_name for ix in indexes)
|
|
|
|
|
|
def _create_index_if_missing(table_name: str, index_name: str, columns, unique: bool = False) -> None:
|
|
if not _has_index(table_name, index_name):
|
|
op.create_index(index_name, table_name, columns, unique=unique)
|
|
|
|
|
|
def _normalize_code(name: str) -> str:
|
|
text = (name or "").strip().upper()
|
|
out = []
|
|
last_sep = False
|
|
for ch in text:
|
|
if ch.isalnum():
|
|
out.append(ch)
|
|
last_sep = False
|
|
else:
|
|
if not last_sep:
|
|
out.append("_")
|
|
last_sep = True
|
|
code = "".join(out).strip("_")
|
|
return code or "UNCATEGORIZED"
|
|
|
|
|
|
def upgrade():
|
|
conn = op.get_bind()
|
|
|
|
if not _has_table("service_categories"):
|
|
op.create_table(
|
|
"service_categories",
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True, nullable=False),
|
|
sa.Column("code", sa.String(length=50), nullable=True),
|
|
sa.Column("name", sa.String(length=100), nullable=True),
|
|
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
|
|
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
|
|
)
|
|
|
|
if _has_table("service_categories"):
|
|
if not _has_column("service_categories", "code"):
|
|
op.add_column("service_categories", sa.Column("code", sa.String(length=50), nullable=True))
|
|
if not _has_column("service_categories", "name"):
|
|
op.add_column("service_categories", sa.Column("name", sa.String(length=100), nullable=True))
|
|
if not _has_column("service_categories", "sort_order"):
|
|
op.add_column("service_categories", sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"))
|
|
if not _has_column("service_categories", "is_active"):
|
|
op.add_column("service_categories", sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()))
|
|
|
|
conn.execute(sa.text(
|
|
"UPDATE service_categories SET name = COALESCE(name, 'Category ' || id) WHERE name IS NULL"
|
|
))
|
|
|
|
rows = conn.execute(sa.text(
|
|
"SELECT id, name FROM service_categories WHERE (code IS NULL OR TRIM(code) = '')"
|
|
)).fetchall()
|
|
used_codes = set()
|
|
existing_code_rows = conn.execute(sa.text(
|
|
"SELECT code FROM service_categories WHERE code IS NOT NULL AND TRIM(code) <> ''"
|
|
)).fetchall()
|
|
for row in existing_code_rows:
|
|
used_codes.add(row[0])
|
|
|
|
for row in rows:
|
|
category_id = row[0]
|
|
name = row[1] or f"Category {category_id}"
|
|
base_code = _normalize_code(name)
|
|
code = base_code
|
|
n = 2
|
|
while code in used_codes:
|
|
code = f"{base_code}_{n}"
|
|
n += 1
|
|
used_codes.add(code)
|
|
conn.execute(
|
|
sa.text("UPDATE service_categories SET code = :code, name = COALESCE(name, :name) WHERE id = :id"),
|
|
{"code": code, "name": name, "id": category_id},
|
|
)
|
|
|
|
conn.execute(sa.text("UPDATE service_categories SET sort_order = 0 WHERE sort_order IS NULL"))
|
|
conn.execute(sa.text("UPDATE service_categories SET is_active = TRUE WHERE is_active IS NULL"))
|
|
|
|
_create_index_if_missing("service_categories", "ix_service_categories_code", ["code"])
|
|
_create_index_if_missing("service_categories", "ix_service_categories_name", ["name"])
|
|
_create_index_if_missing("service_categories", "ix_service_categories_sort_order", ["sort_order"])
|
|
|
|
if _has_table("service_catalogues"):
|
|
additions = [
|
|
("category_id", sa.Integer(), True, None),
|
|
("recurrence_type", sa.String(length=50), True, None),
|
|
("sort_order", sa.Integer(), False, "0"),
|
|
("applicable_individual", sa.Boolean(), False, sa.false()),
|
|
("applicable_proprietorship", sa.Boolean(), False, sa.false()),
|
|
("applicable_partnership", sa.Boolean(), False, sa.false()),
|
|
("applicable_llp", sa.Boolean(), False, sa.false()),
|
|
("applicable_company", sa.Boolean(), False, sa.false()),
|
|
("applicable_trust", sa.Boolean(), False, sa.false()),
|
|
("applicable_society", sa.Boolean(), False, sa.false()),
|
|
]
|
|
|
|
for name, typ, nullable, default in additions:
|
|
if not _has_column("service_catalogues", name):
|
|
kwargs = {"nullable": nullable}
|
|
if default is not None:
|
|
kwargs["server_default"] = default
|
|
op.add_column("service_catalogues", sa.Column(name, typ, **kwargs))
|
|
|
|
conn.execute(sa.text("UPDATE service_catalogues SET sort_order = 0 WHERE sort_order IS NULL"))
|
|
for flag in [
|
|
"applicable_individual",
|
|
"applicable_proprietorship",
|
|
"applicable_partnership",
|
|
"applicable_llp",
|
|
"applicable_company",
|
|
"applicable_trust",
|
|
"applicable_society",
|
|
]:
|
|
if _has_column("service_catalogues", flag):
|
|
conn.execute(sa.text(f"UPDATE service_catalogues SET {flag} = FALSE WHERE {flag} IS NULL"))
|
|
|
|
_create_index_if_missing("service_catalogues", "ix_service_catalogues_category_id", ["category_id"])
|
|
_create_index_if_missing("service_catalogues", "ix_service_catalogues_recurrence_type", ["recurrence_type"])
|
|
_create_index_if_missing("service_catalogues", "ix_service_catalogues_sort_order", ["sort_order"])
|
|
|
|
if _has_column("service_catalogues", "category"):
|
|
rows = conn.execute(sa.text(
|
|
"""
|
|
SELECT DISTINCT TRIM(category) AS category
|
|
FROM service_catalogues
|
|
WHERE category IS NOT NULL AND TRIM(category) <> ''
|
|
ORDER BY TRIM(category)
|
|
"""
|
|
)).fetchall()
|
|
|
|
next_sort = conn.execute(sa.text("SELECT COALESCE(MAX(sort_order), 0) FROM service_categories")).scalar() or 0
|
|
|
|
for row in rows:
|
|
category_name = row[0]
|
|
existing = conn.execute(
|
|
sa.text("SELECT id FROM service_categories WHERE name = :name ORDER BY id LIMIT 1"),
|
|
{"name": category_name},
|
|
).fetchone()
|
|
if not existing:
|
|
base_code = _normalize_code(category_name)
|
|
code = base_code
|
|
n = 2
|
|
while conn.execute(
|
|
sa.text("SELECT 1 FROM service_categories WHERE code = :code"),
|
|
{"code": code},
|
|
).fetchone():
|
|
code = f"{base_code}_{n}"
|
|
n += 1
|
|
next_sort += 1
|
|
conn.execute(
|
|
sa.text(
|
|
"""
|
|
INSERT INTO service_categories (code, name, sort_order, is_active)
|
|
VALUES (:code, :name, :sort_order, TRUE)
|
|
"""
|
|
),
|
|
{"code": code, "name": category_name, "sort_order": next_sort},
|
|
)
|
|
|
|
if _has_column("service_catalogues", "category_id"):
|
|
rows = conn.execute(sa.text(
|
|
"""
|
|
SELECT id, category
|
|
FROM service_catalogues
|
|
WHERE category_id IS NULL
|
|
AND category IS NOT NULL
|
|
AND TRIM(category) <> ''
|
|
"""
|
|
)).fetchall()
|
|
|
|
for row in rows:
|
|
catalogue_id = row[0]
|
|
category_name = row[1].strip()
|
|
match = conn.execute(
|
|
sa.text("SELECT id FROM service_categories WHERE name = :name ORDER BY id LIMIT 1"),
|
|
{"name": category_name},
|
|
).fetchone()
|
|
if match:
|
|
conn.execute(
|
|
sa.text("UPDATE service_catalogues SET category_id = :category_id WHERE id = :catalogue_id"),
|
|
{"category_id": match[0], "catalogue_id": catalogue_id},
|
|
)
|
|
|
|
|
|
def downgrade():
|
|
pass
|