shopdb-flask/shopdb/cli/__init__.py

"""Flask CLI commands."""

import click
from flask.cli import with_appcontext


@click.group('db-utils')
def db_cli():
    """Database utility commands."""
    pass


@db_cli.command('create-all')
@with_appcontext
def create_all():
    """Create all database tables."""
    from shopdb.extensions import db

    db.create_all()
    click.echo(click.style("All tables created.", fg='green'))


@db_cli.command('preflight')
@with_appcontext
def preflight():
    """Check install prerequisites before `flask db upgrade`.

    Reports what must be fixed/installed (Python, required env, DB connectivity,
    and the MySQL 5.6 utf8mb4 index flags). Exits non-zero if anything blocks.
    """
    import sys
    import os
    from sqlalchemy import text
    from flask import current_app
    from shopdb.extensions import db

    failures = []
    warnings = []

    def ok(msg):
        click.echo(click.style('  PASS ', fg='green') + msg)

    def fail(msg, fix):
        failures.append((msg, fix))
        click.echo(click.style('  FAIL ', fg='red') + msg)
        click.echo(click.style('       fix: ', fg='red') + fix)

    def warn(msg, fix):
        warnings.append((msg, fix))
        click.echo(click.style('  WARN ', fg='yellow') + msg)
        click.echo(click.style('       ', fg='yellow') + fix)

    click.echo(click.style('ShopDB preflight', bold=True))

    # Python
    py = sys.version_info
    if py >= (3, 9):
        ok(f'Python {py.major}.{py.minor}.{py.micro}')
    else:
        fail(f'Python {py.major}.{py.minor} is too old',
             'Install Python 3.9 or newer.')

    # Required config
    for key in ('SECRET_KEY', 'JWT_SECRET_KEY', 'DATABASE_URL'):
        value = current_app.config.get(key) or os.environ.get(key)
        if value and 'change' not in str(value).lower() and 'dev-' not in str(value).lower():
            ok(f'{key} is set')
        else:
            fail(f'{key} is missing or a dev default',
                 f'Set {key} in .env (64+ random chars for the secrets).')

    # DB connectivity + MySQL index prerequisites
    try:
        version = db.session.execute(text('SELECT VERSION()')).scalar()
        ok(f'Database reachable (MySQL {version})')
    except Exception as exc:
        fail('Cannot connect to the database', f'Check DATABASE_URL / server. ({exc})')
        version = ''

    if version and version.startswith('5.6'):
        variables = {}
        for name in ('innodb_large_prefix', 'innodb_file_format', 'innodb_file_per_table'):
            try:
                row = db.session.execute(text(f"SHOW VARIABLES LIKE '{name}'")).fetchone()
                variables[name] = row[1] if row else None
            except Exception:
                variables[name] = None
        needed = {'innodb_large_prefix': 'ON', 'innodb_file_format': 'Barracuda',
                  'innodb_file_per_table': 'ON'}
        bad = [n for n, want in needed.items() if str(variables.get(n)).lower() != want.lower()]
        if bad:
            fail('MySQL 5.6 index flags not set: ' + ', '.join(bad),
                 'Add to my.cnf [mysqld]: innodb_file_per_table=1, '
                 'innodb_file_format=Barracuda, innodb_large_prefix=1 (then restart). '
                 'Otherwise `flask db upgrade` fails with error 1071.')
        else:
            ok('MySQL 5.6 index flags OK (Barracuda + large_prefix)')
    elif version:
        ok('MySQL 5.7+/8.0 - no extra index flags needed')

    click.echo('')
    if failures:
        click.echo(click.style(f'{len(failures)} blocker(s). Fix them before installing.', fg='red', bold=True))
        raise SystemExit(1)
    click.echo(click.style('All prerequisites met.' + (f' ({len(warnings)} warning(s))' if warnings else ''), fg='green', bold=True))


@db_cli.command('drop-all')
@click.confirmation_option(prompt='This will delete ALL data. Are you sure?')
@with_appcontext
def drop_all():
    """Drop all database tables."""
    from shopdb.extensions import db

    db.drop_all()
    click.echo(click.style("All tables dropped.", fg='yellow'))


@click.group('seed')
def seed_cli():
    """Database seeding commands."""
    pass


@seed_cli.command('reference-data')
@with_appcontext
def seed_reference_data():
    """Seed reference data (machine types, statuses, etc.)."""
    from shopdb.extensions import db
    from shopdb.core.models import MachineType, OperatingSystem, AssetStatus, LocationType
    from shopdb.core.models.relationship import RelationshipType

    # Machine types
    machine_types = [
        {'machinetype': 'CNC Mill', 'category': 'Equipment', 'description': 'CNC Milling Machine'},
        {'machinetype': 'CNC Lathe', 'category': 'Equipment', 'description': 'CNC Lathe'},
        {'machinetype': 'CMM', 'category': 'Equipment', 'description': 'Coordinate Measuring Machine'},
        {'machinetype': 'EDM', 'category': 'Equipment', 'description': 'Electrical Discharge Machine'},
        {'machinetype': 'Grinder', 'category': 'Equipment', 'description': 'Grinding Machine'},
        {'machinetype': 'Inspection Station', 'category': 'Equipment', 'description': 'Inspection Station'},
        {'machinetype': 'Desktop PC', 'category': 'PC', 'description': 'Desktop Computer'},
        {'machinetype': 'Laptop', 'category': 'PC', 'description': 'Laptop Computer'},
        {'machinetype': 'Shopfloor PC', 'category': 'PC', 'description': 'Shopfloor Computer'},
        {'machinetype': 'Server', 'category': 'Network', 'description': 'Server'},
        {'machinetype': 'Switch', 'category': 'Network', 'description': 'Network Switch'},
        {'machinetype': 'Access Point', 'category': 'Network', 'description': 'Wireless Access Point'},
    ]

    for mt_data in machine_types:
        existing = MachineType.query.filter_by(machinetype=mt_data['machinetype']).first()
        if not existing:
            mt = MachineType(**mt_data)
            db.session.add(mt)

    # Asset statuses (canonical set - the asset model is the contract)
    asset_statuses = [
        {'status': 'In Use', 'description': 'Currently in use', 'color': '#28a745'},
        {'status': 'Inventory', 'description': 'In inventory', 'color': '#17a2b8'},
        {'status': 'In Repair', 'description': 'Being repaired', 'color': '#ffc107'},
        {'status': 'Retired', 'description': 'No longer in use', 'color': '#6c757d'},
        {'status': 'Returned', 'description': 'Returned to vendor or owner', 'color': '#fd7e14'},
        {'status': 'Warrantied', 'description': 'Under warranty service', 'color': '#20c997'},
        {'status': 'Lost', 'description': 'Lost or missing', 'color': '#dc3545'},
    ]

    for s_data in asset_statuses:
        existing = AssetStatus.query.filter_by(status=s_data['status']).first()
        if not existing:
            db.session.add(AssetStatus(isactive=True, **s_data))
        elif existing.isactive is not True:
            existing.isactive = True

    # Location types (ADR-001)
    location_types = ['section', 'cell', 'subcell', 'operation', 'meetingroom',
                      'lab', 'office', 'storage', 'hallway', 'networkcloset',
                      'building']
    for lt in location_types:
        if not LocationType.query.filter_by(locationtype=lt).first():
            db.session.add(LocationType(locationtype=lt, isactive=True))

    # Operating systems
    os_list = [
        {'osname': 'Windows 10', 'osversion': '10.0'},
        {'osname': 'Windows 11', 'osversion': '11.0'},
        {'osname': 'Windows Server 2019', 'osversion': '2019'},
        {'osname': 'Windows Server 2022', 'osversion': '2022'},
        {'osname': 'Linux', 'osversion': 'Various'},
    ]

    for os_data in os_list:
        existing = OperatingSystem.query.filter_by(osname=os_data['osname']).first()
        if not existing:
            os_obj = OperatingSystem(**os_data)
            db.session.add(os_obj)

    # Connection types (pre-1.0 legacy; kept for backward compat with
    # existing relationship rows. New ADR-001 code reasons about the three
    # canonical types below via free-text label.)
    connection_types = [
        {'relationshiptype': 'Serial Cable', 'description': 'RS-232 or similar serial connection'},
        {'relationshiptype': 'Direct Ethernet', 'description': 'Direct network cable (airgapped)'},
        {'relationshiptype': 'USB', 'description': 'USB connection'},
        {'relationshiptype': 'WiFi', 'description': 'Wireless network connection'},
        {'relationshiptype': 'Dualpath', 'description': 'Redundant/failover network path'},
    ]

    for ct_data in connection_types:
        existing = RelationshipType.query.filter_by(relationshiptype=ct_data['relationshiptype']).first()
        if not existing:
            ct = RelationshipType(**ct_data)
            db.session.add(ct)

    # ADR-001 canonical relationship types. Created first without propagation
    # FKs, then patched with propagatesthroughid since `controls` points at
    # `partof` (same table). All three are idempotent.
    #
    # MySQL collation is case-insensitive by default, which would let a
    # legacy capitalized row (e.g. "Controls") match the lowercase
    # "controls" check and skip the insert. Force binary comparison via
    # collate so the three ADR-001 types stay distinct from any legacy
    # rows with the same spelling but different case.
    from sqlalchemy import func, literal
    def _lookup_binary(name):
        dialect = db.engine.dialect.name
        if dialect == 'mysql':
            return RelationshipType.query.filter(
                func.binary(RelationshipType.relationshiptype) == literal(name)
            ).first()
        return RelationshipType.query.filter_by(relationshiptype=name).first()

    adr_types = [
        {'relationshiptype': 'partof', 'description': 'Composition / sub-assembly (ADR-001)'},
        {'relationshiptype': 'controls', 'description': 'Operational authority over another asset (ADR-001)'},
        {'relationshiptype': 'connectedto', 'description': 'Network or data link without authority (ADR-001)'},
    ]
    for at in adr_types:
        if not _lookup_binary(at['relationshiptype']):
            db.session.add(RelationshipType(**at))
    db.session.flush()

    # Wire `controls` -> `partof` propagation rail. partof + connectedto stay
    # null (no propagation).
    partof = _lookup_binary('partof')
    controls = _lookup_binary('controls')
    if partof and controls and controls.propagatesthroughid != partof.relationshiptypeid:
        controls.propagatesthroughid = partof.relationshiptypeid

    # Default-printer link: a PC asset -> its default printer asset. Read by the
    # printer-installer endpoint (parity with classic apipcdefaultprinter.asp).
    # Attribute-style edge, not a position rail, so no propagation.
    if not _lookup_binary('defaultprinter'):
        db.session.add(RelationshipType(
            relationshiptype='defaultprinter',
            description='PC to its default printer (installer preselect, ADR-001)'
        ))

    db.session.commit()
    click.echo(click.style("Reference data seeded.", fg='green'))


@seed_cli.command('test-user')
@with_appcontext
def seed_test_user():
    """Create a test admin user."""
    from shopdb.extensions import db
    from shopdb.core.models import User, Role
    from werkzeug.security import generate_password_hash

    # Create admin role if not exists
    admin_role = Role.query.filter_by(rolename='admin').first()
    if not admin_role:
        admin_role = Role(rolename='admin', description='Administrator')
        db.session.add(admin_role)

    # Create test user
    test_user = User.query.filter_by(username='admin').first()
    if not test_user:
        test_user = User(
            username='admin',
            email='admin@localhost',
            passwordhash=generate_password_hash('admin123'),
            isactive=True
        )
        test_user.roles.append(admin_role)
        db.session.add(test_user)
        db.session.commit()
        click.echo(click.style("Test user created: admin / admin123", fg='green'))
    else:
        click.echo(click.style("Test user already exists", fg='yellow'))


@seed_cli.command('admin')
@click.option('--username', required=True, help='Admin login username')
@click.option('--email', required=True, help='Admin email address')
@click.option('--password', default=None,
              help='Admin password. Omit to generate a strong one and print it once.')
@with_appcontext
def seed_admin(username, email, password):
    """Create the first admin user for a new site.

    Password is generated and printed ONCE if not supplied. Store it safely.
    """
    import secrets
    from shopdb.extensions import db
    from shopdb.core.models import User, Role
    from werkzeug.security import generate_password_hash

    if User.query.filter_by(username=username).first():
        click.echo(click.style(f'User "{username}" already exists.', fg='yellow'))
        return

    admin_role = Role.query.filter_by(rolename='admin').first()
    if not admin_role:
        admin_role = Role(rolename='admin', description='Administrator')
        db.session.add(admin_role)

    generated = password is None
    if generated:
        password = secrets.token_urlsafe(12)

    user = User(username=username, email=email,
                passwordhash=generate_password_hash(password), isactive=True)
    user.roles.append(admin_role)
    db.session.add(user)
    db.session.commit()

    click.echo(click.style(f'Admin "{username}" created.', fg='green'))
    if generated:
        click.echo(click.style('=' * 52, fg='cyan'))
        click.echo(click.style(f'  Password: {password}', fg='cyan', bold=True))
        click.echo(click.style('  Store this now - it will not be shown again.', fg='cyan'))
        click.echo(click.style('=' * 52, fg='cyan'))


@seed_cli.command('permissions')
@with_appcontext
def seed_permissions():
    """Seed predefined permissions."""
    from shopdb.extensions import db
    from shopdb.core.models import Permission

    created = Permission.seed()
    db.session.commit()
    click.echo(click.style(f"{created} permissions created.", fg='green'))


@seed_cli.command('settings')
@with_appcontext
def seed_settings():
    """Seed default system settings."""
    from shopdb.extensions import db
    from shopdb.core.models import Setting
    from shopdb.core.api.settings import build_default_settings

    defaults = build_default_settings()

    created = 0
    for d in defaults:
        if not Setting.query.filter_by(key=d['key']).first():
            setting = Setting(**d)
            db.session.add(setting)
            created += 1

    db.session.commit()
    click.echo(click.style(f"{created} default settings created.", fg='green'))