100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
"""Material aliases — substitution/alias system for material name resolution
|
|
|
|
Revision ID: 020
|
|
Revises: 019
|
|
Create Date: 2026-03-02
|
|
"""
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
revision = "020"
|
|
down_revision = "019"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
# Create material_aliases table
|
|
op.create_table(
|
|
"material_aliases",
|
|
sa.Column("id", UUID(as_uuid=True), primary_key=True, default=uuid.uuid4),
|
|
sa.Column(
|
|
"material_id",
|
|
UUID(as_uuid=True),
|
|
sa.ForeignKey("materials.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
),
|
|
sa.Column("alias", sa.String(300), nullable=False),
|
|
sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
|
)
|
|
|
|
# Case-insensitive unique index on alias
|
|
op.create_index(
|
|
"uq_material_aliases_alias_lower",
|
|
"material_aliases",
|
|
[sa.text("lower(alias)")],
|
|
unique=True,
|
|
)
|
|
|
|
# Index on material_id for FK lookups
|
|
op.create_index(
|
|
"ix_material_aliases_material_id",
|
|
"material_aliases",
|
|
["material_id"],
|
|
)
|
|
|
|
# Seed aliases from naming_scheme.xlsx Materialmapping data
|
|
_seed_aliases()
|
|
|
|
|
|
def _seed_aliases() -> None:
|
|
from app.data.material_alias_seeds import MATERIAL_ALIAS_SEEDS
|
|
|
|
conn = op.get_bind()
|
|
|
|
for entry in MATERIAL_ALIAS_SEEDS:
|
|
material_name = entry["material_name"]
|
|
|
|
# Look up material by name
|
|
result = conn.execute(
|
|
sa.text("SELECT id FROM materials WHERE name = :name"),
|
|
{"name": material_name},
|
|
)
|
|
row = result.fetchone()
|
|
if not row:
|
|
# Material not seeded yet, skip
|
|
continue
|
|
|
|
material_id = row[0]
|
|
|
|
for alias_str in entry["aliases"]:
|
|
# Skip if alias already exists (case-insensitive)
|
|
existing = conn.execute(
|
|
sa.text("SELECT id FROM material_aliases WHERE lower(alias) = lower(:alias)"),
|
|
{"alias": alias_str},
|
|
)
|
|
if existing.fetchone():
|
|
continue
|
|
|
|
conn.execute(
|
|
sa.text(
|
|
"INSERT INTO material_aliases (id, material_id, alias, created_at) "
|
|
"VALUES (:id, :material_id, :alias, :created_at)"
|
|
),
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"material_id": str(material_id),
|
|
"alias": alias_str,
|
|
"created_at": datetime.utcnow(),
|
|
},
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
op.drop_index("ix_material_aliases_material_id", table_name="material_aliases")
|
|
op.drop_index("uq_material_aliases_alias_lower", table_name="material_aliases")
|
|
op.drop_table("material_aliases")
|