Multicaster/init_db.py at main · nicolasumaras/Multicaster · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
#!/usr/bin/env python3
"""
Database initialization script for Multicaster application
"""

from flask import Flask
from flask_migrate import Migrate
from models import db, User, Branch, Switch, SwitchPort, UserRole, AuthType, PortSpeed
from auth import AuthService
from services.ip_service import IPAllocationService
from services.port_service import PortAllocationService
from config import Config
import os

def create_app():
    """Create Flask app for database initialization"""
    app = Flask(__name__)
    app.config.from_object(Config)
    db.init_app(app)
    return app

def init_database():
    """Initialize database with tables and sample data"""
    app = create_app()

    with app.app_context():
        # Create all tables
        db.create_all()

        # Create admin user
        admin_user = User.query.filter_by(username='admin').first()
        if not admin_user:
            admin_user = User(
                username='admin',
                email=app.config['ADMIN_EMAIL'],
                password_hash=AuthService.hash_password(app.config['ADMIN_PASSWORD']),
                role=UserRole.ADMIN,
                auth_type=AuthType.LOCAL,
                is_active=True
            )
            db.session.add(admin_user)
            print("Created admin user")

        # Create sample branch
        sample_branch = Branch.query.filter_by(code='HQ').first()
        if not sample_branch:
            sample_branch = Branch(
                name='Headquarters',
                code='HQ',
                address='123 Main St, City, State 12345',
                vlan_range='100-199',
                # Control Network
                unicast_control_ranges=['10.1.0.0/24', '10.2.0.0/24'],
                # Media Main Network
                unicast_media_main_ranges=['10.1.1.0/24', '10.2.1.0/24'],
                multicast_media_main_video_ranges=['224.1.0.0/20', '224.2.0.0/20'],
                multicast_media_main_audio1_ranges=['224.1.16.0/20', '224.2.16.0/20'],
                multicast_media_main_audio2_ranges=['224.1.17.0/20', '224.2.17.0/20'],
                multicast_media_main_audio3_ranges=['224.1.18.0/20', '224.2.18.0/20'],
                multicast_media_main_audio4_ranges=['224.1.19.0/20', '224.2.19.0/20'],
                multicast_media_main_data_ranges=['224.1.32.0/20', '224.2.32.0/20'],
                # Media Backup Network
                unicast_media_backup_ranges=['10.1.2.0/24', '10.2.2.0/24'],
                multicast_media_backup_video_ranges=['224.1.48.0/20', '224.2.48.0/20'],
                multicast_media_backup_audio1_ranges=['224.1.64.0/20', '224.2.64.0/20'],
                multicast_media_backup_audio2_ranges=['224.1.65.0/20', '224.2.65.0/20'],
                multicast_media_backup_audio3_ranges=['224.1.66.0/20', '224.2.66.0/20'],
                multicast_media_backup_audio4_ranges=['224.1.67.0/20', '224.2.67.0/20'],
                multicast_media_backup_data_ranges=['224.1.80.0/20', '224.2.80.0/20']
            )
            db.session.add(sample_branch)
            db.session.flush()  # Get branch ID
        else:
            # Update existing branch with new range structure
            sample_branch.unicast_control_ranges = ['10.1.0.0/24', '10.2.0.0/24']
            sample_branch.unicast_media_main_ranges = ['10.1.1.0/24', '10.2.1.0/24']
            sample_branch.multicast_media_main_video_ranges = ['224.1.0.0/20', '224.2.0.0/20']
            sample_branch.multicast_media_main_audio1_ranges = ['224.1.16.0/20', '224.2.16.0/20']
            sample_branch.multicast_media_main_audio2_ranges = ['224.1.17.0/20', '224.2.17.0/20']
            sample_branch.multicast_media_main_audio3_ranges = ['224.1.18.0/20', '224.2.18.0/20']
            sample_branch.multicast_media_main_audio4_ranges = ['224.1.19.0/20', '224.2.19.0/20']
            sample_branch.multicast_media_main_data_ranges = ['224.1.32.0/20', '224.2.32.0/20']
            sample_branch.unicast_media_backup_ranges = ['10.1.2.0/24', '10.2.2.0/24']
            sample_branch.multicast_media_backup_video_ranges = ['224.1.48.0/20', '224.2.48.0/20']
            sample_branch.multicast_media_backup_audio1_ranges = ['224.1.64.0/20', '224.2.64.0/20']
            sample_branch.multicast_media_backup_audio2_ranges = ['224.1.65.0/20', '224.2.65.0/20']
            sample_branch.multicast_media_backup_audio3_ranges = ['224.1.66.0/20', '224.2.66.0/20']
            sample_branch.multicast_media_backup_audio4_ranges = ['224.1.67.0/20', '224.2.67.0/20']
            sample_branch.multicast_media_backup_data_ranges = ['224.1.80.0/20', '224.2.80.0/20']
            db.session.flush()

            # Initialize IP ranges
            print("Initializing IP ranges for sample branch...")
            IPAllocationService.initialize_ip_range(sample_branch.id, '224.1.0.0/24', 'multicast')
            IPAllocationService.initialize_ip_range(sample_branch.id, '10.1.0.0/24', 'unicast')
            IPAllocationService.initialize_ip_range(sample_branch.id, '10.2.0.0/26', 'unicast')  # Second unicast range

            # Initialize specific multicast ranges for testing
            print("Initializing specific multicast ranges...")
            IPAllocationService.initialize_ip_range(sample_branch.id, '224.1.0.0/26', 'multicast', 'video')  # Video
            IPAllocationService.initialize_ip_range(sample_branch.id, '224.1.16.0/26', 'multicast', 'audio') # Audio
            IPAllocationService.initialize_ip_range(sample_branch.id, '224.1.32.0/26', 'multicast', 'data') # Data

            print("Created sample branch with IP ranges")

        # Create sample switch
        sample_switch = Switch.query.filter_by(name='Core-Switch-01').first()
        if not sample_switch and sample_branch:
            sample_switch = Switch(
                name='Core-Switch-01',
                model='Catalyst 9500',
                manufacturer='Cisco',
                ip_address='10.1.0.10',
                management_ip='192.168.1.10',
                location='Data Center Rack A1',
                branch_id=sample_branch.id,
                is_managed=False
            )
            db.session.add(sample_switch)
            db.session.flush()  # Get switch ID

            # Create sample ports
            print("Creating sample switch ports...")

            # 48x 10G ports
            port_config = [
                {'start': 1, 'count': 48, 'speed': '10G'},
                {'start': 49, 'count': 4, 'speed': '100G'}
            ]

            PortAllocationService.bulk_create_ports(sample_switch.id, port_config)

            print("Created sample switch with ports")

        # Create sample users for different roles
        sample_users = [
            {
                'username': 'manager',
                'email': 'manager@company.com',
                'password': 'manager123',
                'role': UserRole.MANAGER
            },
            {
                'username': 'operator',
                'email': 'operator@company.com',
                'password': 'operator123',
                'role': UserRole.OPERATOR
            },
            {
                'username': 'viewer',
                'email': 'viewer@company.com',
                'password': 'viewer123',
                'role': UserRole.VIEWER
            }
        ]

        for user_data in sample_users:
            existing_user = User.query.filter_by(username=user_data['username']).first()
            if not existing_user:
                user = User(
                    username=user_data['username'],
                    email=user_data['email'],
                    password_hash=AuthService.hash_password(user_data['password']),
                    role=user_data['role'],
                    auth_type=AuthType.LOCAL,
                    is_active=True
                )
                db.session.add(user)
                print(f"Created sample user: {user_data['username']}")

        db.session.commit()
        print("Database initialization completed successfully!")

        # Print summary
        print("\n=== Database Summary ===")
        print(f"Branches: {Branch.query.count()}")
        print(f"Switches: {Switch.query.count()}")
        print(f"Ports: {SwitchPort.query.count()}")
        print(f"Users: {User.query.count()}")
        try:
            ip_count = db.session.execute(db.text("SELECT COUNT(*) FROM ip_allocations")).scalar()
            print(f"IP Allocations: {ip_count or 0}")
        except:
            print("IP Allocations: 0")

        print("\n=== Default Credentials ===")
        print("Admin: admin / admin123")
        print("Manager: manager / manager123")
        print("Operator: operator / operator123")
        print("Viewer: viewer / viewer123")

if __name__ == '__main__':
    init_database()