from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import declarative_base from sqlalchemy import text import os BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) DATABASE_URL = f"sqlite+aiosqlite:///{os.path.join(BASE_DIR, 'billing.db')}" engine = create_async_engine(DATABASE_URL, echo=True) AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) Base = declarative_base() async def get_db(): async with AsyncSessionLocal() as session: yield session async def init_db(): from app.models.models import Base async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # create_all does not add columns to the existing SQLite database. columns = { "provisioning_status": "VARCHAR DEFAULT 'not_started'", "mikrotik_provisioning_status": "VARCHAR DEFAULT 'pending'", "olt_provisioning_status": "VARCHAR DEFAULT 'pending'", "provisioning_message": "TEXT", "provisioned_at": "DATETIME", "is_archived": "BOOLEAN DEFAULT 0", "play_type": "VARCHAR DEFAULT '1play'", "jatuh_tempo": "DATETIME", "service_profile": "VARCHAR", "provisioning_mode": "VARCHAR DEFAULT 'private_pppoe'", "pppoe_username": "VARCHAR", "pppoe_password": "VARCHAR", "onu_id": "INTEGER", "wan_vlan": "INTEGER DEFAULT 175", "service_vlan": "INTEGER DEFAULT 175", "pppoe_vlan": "INTEGER DEFAULT 100", "hotspot_vlan": "INTEGER DEFAULT 150", "iptv_vlan": "INTEGER DEFAULT 200", "olt_vlan_profile": "VARCHAR DEFAULT 'bnbv1009'", # Added after the original database was created; create_all() # does not alter existing SQLite tables. "odp_id": "INTEGER", "olt_vendor": "VARCHAR DEFAULT 'ZTE'", "olt_model": "VARCHAR DEFAULT 'C320'", "service_package_id": "INTEGER", "service_profile_id": "INTEGER", "service_mode": "VARCHAR DEFAULT 'private_pppoe'", "management_pppoe_vlan": "INTEGER", "odp_port": "INTEGER", "ont_management_mode": "VARCHAR DEFAULT 'none'", "management_pppoe_username": "VARCHAR", "management_pppoe_password": "VARCHAR", "ont_sn_mode": "VARCHAR DEFAULT 'manual'", "onu_id_mode": "VARCHAR DEFAULT 'manual'", "olt_port_mode": "VARCHAR DEFAULT 'manual'", "ont_brand_mode": "VARCHAR DEFAULT 'manual'", "ont_rx_power": "VARCHAR", "ont_tx_power": "VARCHAR", "hardware_discovery_status": "VARCHAR DEFAULT 'unavailable'", "odp_latitude": "VARCHAR", "odp_longitude": "VARCHAR", "customer_code": "VARCHAR", "nik": "VARCHAR", "pppoe_secret_revealed_at": "DATETIME", "pppoe_secret_version": "INTEGER DEFAULT 1", "service_modes_json": "TEXT DEFAULT '[\"pppoe\"]'", } existing = {row[1] for row in (await conn.execute(text("PRAGMA table_info(customers)"))).fetchall()} for name, definition in columns.items(): if name not in existing: await conn.execute(text(f"ALTER TABLE customers ADD COLUMN {name} {definition}")) # Existing rows receive a deterministic, permanent payment code. rows = (await conn.execute(text("SELECT id FROM customers WHERE customer_code IS NULL OR customer_code = ''"))).fetchall() for (customer_id,) in rows: await conn.execute(text("UPDATE customers SET customer_code = :code WHERE id = :id"), {"code": f"PLG-{customer_id:07d}", "id": customer_id}) await conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_customers_customer_code ON customers(customer_code)")) await conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_customers_nik_unique ON customers(nik) WHERE nik IS NOT NULL")) await conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_customers_pppoe_username_unique ON customers(pppoe_username) WHERE pppoe_username IS NOT NULL AND pppoe_username <> ''")) await conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_customers_ont_sn_unique ON customers(ont_sn) WHERE ont_sn IS NOT NULL AND ont_sn <> ''")) # A completed installation must not remain suspended merely because no # invoice exists yet. Suspension is only a billing consequence of an # overdue unpaid invoice. await conn.execute(text(""" UPDATE customers SET status = 'active', provisioning_message = 'Internet aktif; status dipulihkan oleh rekonsiliasi billing.' WHERE status = 'suspended' AND provisioning_status = 'provisioned' AND NOT EXISTS ( SELECT 1 FROM invoices WHERE invoices.customer_id = customers.id AND invoices.status <> 'paid' AND invoices.due_date <= CURRENT_TIMESTAMP ) """)) voucher_columns = { "used_at": "DATETIME", "expires_at": "DATETIME", } existing_voucher = {row[1] for row in (await conn.execute(text("PRAGMA table_info(vouchers)"))).fetchall()} for name, definition in voucher_columns.items(): if name not in existing_voucher: await conn.execute(text(f"ALTER TABLE vouchers ADD COLUMN {name} {definition}")) existing_agents = {row[1] for row in (await conn.execute(text("PRAGMA table_info(voucher_agents)"))).fetchall()} if "commission_rate" not in existing_agents: await conn.execute(text("ALTER TABLE voucher_agents ADD COLUMN commission_rate INTEGER DEFAULT 0")) # Multi-vendor device registry and sanitized provisioning-audit migration. table_columns = { "customers": { "olt_device_id": "INTEGER", "mikrotik_device_id": "INTEGER", "olt_adapter_key": "VARCHAR", "olt_fields_json": "TEXT", "pppoe_vlan": "INTEGER DEFAULT 100", "hotspot_vlan": "INTEGER DEFAULT 150", }, "service_packages": { "pppoe_vlan": "INTEGER DEFAULT 100", "hotspot_vlan": "INTEGER DEFAULT 150", }, "mikrotik_devices": { "area": "VARCHAR", "status": "VARCHAR DEFAULT 'registered'", "credential_id": "INTEGER", }, "olt_devices": { "location": "VARCHAR", "pop_name": "VARCHAR", "management_protocol": "VARCHAR DEFAULT 'ssh'", "api_port": "INTEGER", "status": "VARCHAR DEFAULT 'registered'", "credential_id": "INTEGER", }, "odps": {"mikrotik_device_id": "INTEGER"}, "provisioning_jobs": { "vendor": "VARCHAR", "adapter_key": "VARCHAR", "olt_device_id": "INTEGER", "mikrotik_device_id": "INTEGER", "payload_summary": "TEXT", "executed_at": "DATETIME", }, } for table, additions in table_columns.items(): existing_table_columns = {row[1] for row in (await conn.execute(text(f"PRAGMA table_info({table})"))).fetchall()} for name, definition in additions.items(): if name not in existing_table_columns: await conn.execute(text(f"ALTER TABLE {table} ADD COLUMN {name} {definition}"))