481 lines
15 KiB
Python
481 lines
15 KiB
Python
# AccManager Backend Server (Python Alternative)
|
|
# Requirements: Python 3.8+, Flask, pyodbc
|
|
|
|
from flask import Flask, request, jsonify, render_template
|
|
from flask_cors import CORS
|
|
from datetime import datetime
|
|
import pyodbc
|
|
import json
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
|
|
# SQL Server Connection Settings
|
|
DB_SERVER = os.getenv('DB_SERVER', '172.20.235.176')
|
|
DB_USER = os.getenv('DB_USER', 'sa')
|
|
DB_PASSWORD = os.getenv('DB_PASSWORD', 'robotics@2020')
|
|
DB_NAME = os.getenv('DB_NAME', 'AccManager')
|
|
|
|
# Connection string
|
|
CONNECTION_STRING = f'''
|
|
Driver={{ODBC Driver 17 for SQL Server}};
|
|
Server={DB_SERVER};
|
|
Database={DB_NAME};
|
|
UID={DB_USER};
|
|
PWD={DB_PASSWORD};
|
|
TrustServerCertificate=yes;
|
|
'''
|
|
|
|
# Initialize database connection
|
|
def get_db_connection():
|
|
try:
|
|
conn = pyodbc.connect(CONNECTION_STRING)
|
|
return conn
|
|
except Exception as e:
|
|
print(f"Database connection error: {e}")
|
|
return None
|
|
|
|
# Create tables
|
|
def create_tables():
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return False
|
|
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# Users table
|
|
cursor.execute('''
|
|
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Users')
|
|
BEGIN
|
|
CREATE TABLE Users (
|
|
UserId INT PRIMARY KEY IDENTITY(1,1),
|
|
Username NVARCHAR(50) UNIQUE NOT NULL,
|
|
Password NVARCHAR(255) NOT NULL,
|
|
Email NVARCHAR(100),
|
|
FullName NVARCHAR(100),
|
|
Role NVARCHAR(50) NOT NULL,
|
|
Status NVARCHAR(20) DEFAULT 'Active',
|
|
CreatedDate DATETIME DEFAULT GETDATE(),
|
|
LastLogin DATETIME,
|
|
IsActive BIT DEFAULT 1
|
|
)
|
|
END
|
|
''')
|
|
|
|
# Applications table
|
|
cursor.execute('''
|
|
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Applications')
|
|
BEGIN
|
|
CREATE TABLE Applications (
|
|
AppId INT PRIMARY KEY IDENTITY(1,1),
|
|
Name NVARCHAR(100) NOT NULL,
|
|
Type NVARCHAR(50),
|
|
Status NVARCHAR(20) DEFAULT 'online',
|
|
Icon NVARCHAR(50),
|
|
Description NVARCHAR(500),
|
|
CreatedDate DATETIME DEFAULT GETDATE(),
|
|
UpdatedDate DATETIME DEFAULT GETDATE()
|
|
)
|
|
END
|
|
''')
|
|
|
|
# Accounts table
|
|
cursor.execute('''
|
|
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Accounts')
|
|
BEGIN
|
|
CREATE TABLE Accounts (
|
|
AccountId INT PRIMARY KEY IDENTITY(1,1),
|
|
UserId INT NOT NULL,
|
|
AppId INT NOT NULL,
|
|
AccountUsername NVARCHAR(100),
|
|
AccountPassword NVARCHAR(255),
|
|
Email NVARCHAR(100),
|
|
AccessLevel NVARCHAR(50),
|
|
Status NVARCHAR(20) DEFAULT 'Active',
|
|
Notes NVARCHAR(MAX),
|
|
CreatedDate DATETIME DEFAULT GETDATE(),
|
|
UpdatedDate DATETIME DEFAULT GETDATE(),
|
|
FOREIGN KEY (UserId) REFERENCES Users(UserId) ON DELETE CASCADE,
|
|
FOREIGN KEY (AppId) REFERENCES Applications(AppId) ON DELETE CASCADE
|
|
)
|
|
END
|
|
''')
|
|
|
|
# AuditLog table
|
|
cursor.execute('''
|
|
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'AuditLog')
|
|
BEGIN
|
|
CREATE TABLE AuditLog (
|
|
LogId INT PRIMARY KEY IDENTITY(1,1),
|
|
UserId INT,
|
|
Action NVARCHAR(50),
|
|
TableName NVARCHAR(50),
|
|
RecordId INT,
|
|
OldValue NVARCHAR(MAX),
|
|
NewValue NVARCHAR(MAX),
|
|
Timestamp DATETIME DEFAULT GETDATE(),
|
|
FOREIGN KEY (UserId) REFERENCES Users(UserId)
|
|
)
|
|
END
|
|
''')
|
|
|
|
conn.commit()
|
|
|
|
# Insert admin user
|
|
cursor.execute('''
|
|
IF NOT EXISTS (SELECT * FROM Users WHERE Username = 'admin')
|
|
INSERT INTO Users (Username, Password, Email, FullName, Role, IsActive)
|
|
VALUES ('admin', 'admin', 'admin@accmanager.local', 'Administrator', 'admin', 1)
|
|
''')
|
|
|
|
# Insert sample applications
|
|
cursor.execute('''
|
|
IF (SELECT COUNT(*) FROM Applications) = 0
|
|
BEGIN
|
|
INSERT INTO Applications (Name, Type, Status, Icon, Description)
|
|
VALUES
|
|
('AWS', 'Cloud', 'online', 'cloud', 'Amazon Web Services'),
|
|
('GitHub', 'VCS', 'online', 'code', 'GitHub - Version Control'),
|
|
('Google Workspace', 'Collaboration', 'online', 'mail', 'Google Workspace'),
|
|
('Nginx Proxy', 'Infra', 'offline', 'dns', 'Nginx Web Server')
|
|
END
|
|
''')
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
conn.close()
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Table creation error: {e}")
|
|
return False
|
|
|
|
# ==========================================
|
|
# Routes: Health Check
|
|
# ==========================================
|
|
|
|
@app.route('/api/health', methods=['GET'])
|
|
def health():
|
|
conn = get_db_connection()
|
|
if conn:
|
|
conn.close()
|
|
return jsonify({"status": "OK", "database": "Connected"}), 200
|
|
else:
|
|
return jsonify({"status": "ERROR", "database": "Disconnected"}), 500
|
|
|
|
# ==========================================
|
|
# Routes: Authentication
|
|
# ==========================================
|
|
|
|
@app.route('/api/auth/login', methods=['POST'])
|
|
def login():
|
|
try:
|
|
data = request.get_json()
|
|
username = data.get('username')
|
|
password = data.get('password')
|
|
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT UserId, Username, Email, FullName, Role, Status
|
|
FROM Users
|
|
WHERE Username = ? AND Password = ? AND IsActive = 1
|
|
''', (username, password))
|
|
|
|
user = cursor.fetchone()
|
|
|
|
if user:
|
|
cursor.execute('UPDATE Users SET LastLogin = GETDATE() WHERE UserId = ?', user[0])
|
|
conn.commit()
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"message": "Login successful",
|
|
"user": {
|
|
"UserId": user[0],
|
|
"username": user[1],
|
|
"email": user[2],
|
|
"fullname": user[3],
|
|
"role": user[4],
|
|
"status": user[5]
|
|
}
|
|
}), 200
|
|
else:
|
|
return jsonify({
|
|
"success": False,
|
|
"message": "Invalid username or password"
|
|
}), 401
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ==========================================
|
|
# Routes: Users
|
|
# ==========================================
|
|
|
|
@app.route('/api/users', methods=['GET'])
|
|
def get_users():
|
|
try:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT UserId, Username, Email, FullName, Role, Status, CreatedDate
|
|
FROM Users
|
|
ORDER BY CreatedDate DESC
|
|
''')
|
|
|
|
users = []
|
|
for row in cursor.fetchall():
|
|
users.append({
|
|
"UserId": row[0],
|
|
"Username": row[1],
|
|
"Email": row[2],
|
|
"FullName": row[3],
|
|
"Role": row[4],
|
|
"Status": row[5],
|
|
"CreatedDate": str(row[6])
|
|
})
|
|
|
|
return jsonify({"success": True, "data": users}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
@app.route('/api/users/<int:user_id>', methods=['GET'])
|
|
def get_user(user_id):
|
|
try:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT * FROM Users WHERE UserId = ?', (user_id,))
|
|
|
|
user = cursor.fetchone()
|
|
|
|
if user:
|
|
return jsonify({
|
|
"success": True,
|
|
"data": {"UserId": user[0], "Username": user[1]}
|
|
}), 200
|
|
else:
|
|
return jsonify({"success": False, "message": "User not found"}), 404
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
@app.route('/api/users', methods=['POST'])
|
|
def create_user():
|
|
try:
|
|
data = request.get_json()
|
|
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO Users (Username, Password, Email, FullName, Role, IsActive)
|
|
VALUES (?, ?, ?, ?, ?, 1)
|
|
''', (data['username'], data['password'], data['email'],
|
|
data['fullname'], data['role']))
|
|
|
|
conn.commit()
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"message": "User created"
|
|
}), 201
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ==========================================
|
|
# Routes: Applications
|
|
# ==========================================
|
|
|
|
@app.route('/api/applications', methods=['GET'])
|
|
def get_applications():
|
|
try:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT * FROM Applications ORDER BY Name')
|
|
|
|
apps = []
|
|
for row in cursor.fetchall():
|
|
apps.append({
|
|
"AppId": row[0],
|
|
"Name": row[1],
|
|
"Type": row[2],
|
|
"Status": row[3],
|
|
"Icon": row[4],
|
|
"Description": row[5]
|
|
})
|
|
|
|
return jsonify({"success": True, "data": apps}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ==========================================
|
|
# Routes: Accounts
|
|
# ==========================================
|
|
|
|
@app.route('/api/accounts/user/<int:user_id>', methods=['GET'])
|
|
def get_accounts(user_id):
|
|
try:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT a.*, app.Name, app.Type
|
|
FROM Accounts a
|
|
JOIN Applications app ON a.AppId = app.AppId
|
|
WHERE a.UserId = ?
|
|
ORDER BY a.CreatedDate DESC
|
|
''', (user_id,))
|
|
|
|
accounts = []
|
|
for row in cursor.fetchall():
|
|
accounts.append({
|
|
"AccountId": row[0],
|
|
"UserId": row[1],
|
|
"AppId": row[2],
|
|
"AccountUsername": row[3],
|
|
"AccountPassword": row[4],
|
|
"AppName": row[12],
|
|
"AppType": row[13]
|
|
})
|
|
|
|
return jsonify({"success": True, "data": accounts}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ==========================================
|
|
# Routes: Database Info
|
|
# ==========================================
|
|
|
|
@app.route('/api/database/info', methods=['GET'])
|
|
def database_info():
|
|
try:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
return jsonify({"success": False, "message": "Database connection failed"}), 500
|
|
|
|
cursor = conn.cursor()
|
|
|
|
# Get tables
|
|
cursor.execute('''
|
|
SELECT TABLE_NAME
|
|
FROM INFORMATION_SCHEMA.TABLES
|
|
WHERE TABLE_SCHEMA = 'dbo'
|
|
ORDER BY TABLE_NAME
|
|
''')
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
|
|
# Get statistics
|
|
cursor.execute('SELECT COUNT(*) FROM Users')
|
|
user_count = cursor.fetchone()[0]
|
|
|
|
cursor.execute('SELECT COUNT(*) FROM Applications')
|
|
app_count = cursor.fetchone()[0]
|
|
|
|
cursor.execute('SELECT COUNT(*) FROM Accounts')
|
|
account_count = cursor.fetchone()[0]
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"database": "AccManager",
|
|
"server": DB_SERVER,
|
|
"tables": tables,
|
|
"statistics": {
|
|
"users": user_count,
|
|
"applications": app_count,
|
|
"accounts": account_count
|
|
}
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({"success": False, "message": str(e)}), 500
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ==========================================
|
|
# Error Handlers
|
|
# ==========================================
|
|
|
|
@app.errorhandler(404)
|
|
def not_found(error):
|
|
return jsonify({"error": "Not found"}), 404
|
|
|
|
@app.errorhandler(500)
|
|
def server_error(error):
|
|
return jsonify({"error": "Internal server error"}), 500
|
|
|
|
# ==========================================
|
|
# Startup
|
|
# ==========================================
|
|
|
|
if __name__ == '__main__':
|
|
print("\n========================================")
|
|
print("AccManager Backend Server (Python)")
|
|
print("========================================")
|
|
print(f"Server: {DB_SERVER}")
|
|
print(f"Database: {DB_NAME}")
|
|
print(f"User: {DB_USER}")
|
|
|
|
# Initialize database
|
|
print("\nInitializing database...")
|
|
if create_tables():
|
|
print("✓ Database tables created/verified")
|
|
else:
|
|
print("✗ Database initialization failed")
|
|
|
|
print("\n✓ Server running on http://localhost:5000")
|
|
print("\nAPI Endpoints:")
|
|
print(" /api/health")
|
|
print(" /api/auth/login")
|
|
print(" /api/users")
|
|
print(" /api/applications")
|
|
print(" /api/accounts/user/<id>")
|
|
print(" /api/database/info")
|
|
print("========================================\n")
|
|
|
|
app.run(debug=True, host='localhost', port=5000)
|