PotatoMaxwell/mxpic_EDA
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Multiuser system initiated with log recording

Browse Source
This commit is contained in:
PotatoMaxwell
2026-05-27 14:27:31 +08:00
parent 2b4bb5ea64
commit 1c2a0647cb
9 changed files with 619 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/database.py
+125 -4
View File
@@ -2,12 +2,20 @@
import sqlite3
import os
from werkzeug.security import generate_password_hash
from datetime import datetime
# Save the database in the backend folder
DB_FILE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "database", "mxpic_data.db"))
def connect_db():
conn = sqlite3.connect(DB_FILE, timeout=20)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=20000")
return conn
def init_db():
conn = sqlite3.connect(DB_FILE)
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
conn = connect_db()
cursor = conn.cursor()
# Create Users Table
@@ -18,25 +26,138 @@ def init_db():
password_hash TEXT NOT NULL
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
username TEXT NOT NULL,
action TEXT NOT NULL,
project TEXT,
cell TEXT,
detail TEXT,
ip_address TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id)
)
''')
cursor.execute("PRAGMA table_info(users)")
existing_columns = {row[1] for row in cursor.fetchall()}
migrations = {
"created_at": "ALTER TABLE users ADD COLUMN created_at TEXT",
"credits": "ALTER TABLE users ADD COLUMN credits INTEGER NOT NULL DEFAULT 0",
"occupation": "ALTER TABLE users ADD COLUMN occupation TEXT NOT NULL DEFAULT 'intern'",
}
for column, statement in migrations.items():
if column not in existing_columns:
cursor.execute(statement)
now = datetime.utcnow().strftime("%Y-%m-%d")
cursor.execute("UPDATE users SET created_at = ? WHERE created_at IS NULL OR created_at = ''", (now,))
# Insert a test user if the table is empty
# Insert default users for local multi-account development.
cursor.execute("SELECT * FROM users WHERE username = 'admin'")
if not cursor.fetchone():
test_hash = generate_password_hash("123456")
cursor.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)", ("admin", test_hash))
cursor.execute(
"INSERT INTO users (username, password_hash, created_at, credits, occupation) VALUES (?, ?, ?, ?, ?)",
("admin", test_hash, now, 0, "principle engineer")
)
print("Test user created. Username: admin | Password: 123456")
cursor.execute("SELECT * FROM users WHERE username = 'engineer'")
if not cursor.fetchone():
test_hash = generate_password_hash("123456")
cursor.execute(
"INSERT INTO users (username, password_hash, created_at, credits, occupation) VALUES (?, ?, ?, ?, ?)",
("engineer", test_hash, now, 0, "junior engineer")
)
print("Second test user created. Username: engineer | Password: 123456")
conn.commit()
conn.close()
def get_user(username):
conn = sqlite3.connect(DB_FILE)
conn = connect_db()
cursor = conn.cursor()
cursor.execute("SELECT id, username, password_hash FROM users WHERE username = ?", (username,))
user = cursor.fetchone()
conn.close()
return user
def get_user_profile(user_id):
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
"SELECT id, username, created_at, credits, occupation FROM users WHERE id = ?",
(user_id,)
)
user = cursor.fetchone()
conn.close()
return user
def get_user_auth_by_id(user_id):
conn = connect_db()
cursor = conn.cursor()
cursor.execute("SELECT id, username, password_hash FROM users WHERE id = ?", (user_id,))
user = cursor.fetchone()
conn.close()
return user
def update_user_occupation(user_id, occupation):
conn = connect_db()
cursor = conn.cursor()
cursor.execute("UPDATE users SET occupation = ? WHERE id = ?", (occupation, user_id))
conn.commit()
conn.close()
def update_user_password(user_id, password):
conn = connect_db()
cursor = conn.cursor()
cursor.execute("UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), user_id))
conn.commit()
conn.close()
def add_user_log(user_id, username, action, project=None, cell=None, detail=None, ip_address=None):
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
'''
INSERT INTO user_logs (user_id, username, action, project, cell, detail, ip_address, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''',
(
user_id,
username,
action,
project,
cell,
detail,
ip_address,
datetime.utcnow().isoformat(timespec="seconds") + "Z"
)
)
conn.commit()
conn.close()
def list_user_logs(user_id, limit=200):
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
'''
SELECT id, action, project, cell, detail, ip_address, created_at
FROM user_logs
WHERE user_id = ?
ORDER BY id DESC
LIMIT ?
''',
(user_id, limit)
)
rows = cursor.fetchall()
conn.close()
return rows
if __name__ == "__main__":
init_db()
print("Database initialized successfully.")
backend/server.py
+132 -3
View File
@@ -25,7 +25,12 @@ DATABASE_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..', 'database'))
app = Flask(__name__, template_folder=FRONTEND_DIR, static_folder=FRONTEND_DIR)
app.secret_key = 'super_secret_mxpic_key'
app.secret_key = os.environ.get('MXPIC_SECRET_KEY', 'change_me_for_intranet_deployment')
app.config.update(
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SAMESITE='Lax',
SESSION_COOKIE_SECURE=os.environ.get('MXPIC_COOKIE_SECURE', '0').lower() in {'1', 'true', 'yes'},
)
app.json.sort_keys = False
database.init_db()
@@ -40,6 +45,31 @@ def login_required_json(view_func):
return wrapper
def request_ip():
forwarded_for = request.headers.get('X-Forwarded-For', '')
if forwarded_for:
return forwarded_for.split(',')[0].strip()
return request.remote_addr
def record_action(action, project=None, cell=None, detail=None):
if 'user_id' not in session:
return
if isinstance(detail, (dict, list)):
detail = json.dumps(detail, ensure_ascii=False)
elif detail is not None:
detail = str(detail)
database.add_user_log(
session.get('user_id'),
session.get('username', 'unknown'),
action,
project=safe_name(project, '') if project else None,
cell=safe_name(cell, '') if cell else None,
detail=detail,
ip_address=request_ip()
)
def safe_name(value, fallback):
"""Keep user/project/cell names filesystem-friendly without changing display names."""
value = (value or '').strip()
@@ -187,6 +217,7 @@ def login():
if user and check_password_hash(user[2], password):
session['user_id'] = user[0]
session['username'] = user[1]
record_action('login')
return redirect(url_for('dashboard'))
else:
return render_template('login.html', error="Invalid username or password")
@@ -211,10 +242,16 @@ def canvas():
@app.route('/logout')
def logout():
"""Clear session and return to login."""
record_action('logout')
session.clear()
return redirect(url_for('home'))
@app.route('/api/health')
def health_check():
return jsonify({"status": "ok", "service": "mxpic_eda"})
@app.route('/api/technologies', methods=['GET'])
@login_required_json
def list_technologies():
@@ -239,6 +276,89 @@ def list_technologies():
return jsonify({"technologies": technologies})
@app.route('/api/profile', methods=['GET', 'PATCH'])
@login_required_json
def account_profile():
occupations = {'intern', 'senior engineer', 'junior engineer', 'principle engineer'}
user_id = session.get('user_id')
if request.method == 'PATCH':
data = request.get_json(silent=True) or {}
occupation = (data.get('occupation') or '').strip().lower()
if occupation not in occupations:
return jsonify({"error": "Invalid occupation"}), 400
database.update_user_occupation(user_id, occupation)
record_action('profile.update_occupation', detail={"occupation": occupation})
profile = database.get_user_profile(user_id)
if not profile:
return jsonify({"error": "Profile not found"}), 404
return jsonify({
"id": f"mx-{int(profile[0]):06d}",
"username": profile[1],
"created_at": profile[2],
"credits": profile[3] or 0,
"occupation": profile[4] or "intern",
"occupations": sorted(occupations)
})
@app.route('/api/profile/password', methods=['POST'])
@login_required_json
def change_password():
data = request.get_json(silent=True) or {}
current_password = data.get('current_password') or ''
new_password = data.get('new_password') or ''
if len(new_password) < 6:
return jsonify({"error": "New password must be at least 6 characters"}), 400
user = database.get_user_auth_by_id(session.get('user_id'))
if not user or not check_password_hash(user[2], current_password):
return jsonify({"error": "Current password is incorrect"}), 400
database.update_user_password(user[0], new_password)
record_action('profile.change_password')
return jsonify({"message": "Password updated"})
@app.route('/api/logs', methods=['GET', 'POST'])
@login_required_json
def user_logs():
if request.method == 'POST':
data = request.get_json(silent=True) or {}
action = safe_name(data.get('action'), '')
if not action:
return jsonify({"error": "Action is required"}), 400
record_action(
action,
project=data.get('project'),
cell=data.get('cell'),
detail=data.get('detail')
)
return jsonify({"message": "logged"})
try:
limit = min(500, max(1, int(request.args.get('limit', 200))))
except ValueError:
limit = 200
rows = database.list_user_logs(session.get('user_id'), limit=limit)
return jsonify({
"logs": [
{
"id": row[0],
"action": row[1],
"project": row[2],
"cell": row[3],
"detail": row[4],
"ip_address": row[5],
"created_at": row[6]
}
for row in rows
]
})
@app.route('/api/projects', methods=['GET'])
@login_required_json
def list_projects():
@@ -291,6 +411,7 @@ def create_project():
"name": project_name,
"technology": technology
})
record_action('project.create', project=project_name, detail={"technology": technology})
return jsonify({"name": project_name, "technology": technology}), 201
@@ -337,6 +458,7 @@ def delete_project(project_name):
return jsonify({"error": "Project not found"}), 404
shutil.rmtree(target)
record_action('project.delete', project=project_name)
return jsonify({"message": "deleted", "project": safe_name(project_name, 'project_1')})
@@ -350,8 +472,10 @@ def rename_cell(project_name, cell_name):
if not target.startswith(project_dir + os.sep):
return jsonify({"error": "Invalid cell path"}), 400
if not os.path.exists(target):
record_action('canvas.delete_missing', project=project_name, cell=cell)
return jsonify({"message": "already deleted", "cell": cell})
os.remove(target)
record_action('canvas.delete', project=project_name, cell=cell)
return jsonify({"message": "deleted", "cell": cell})
data = request.get_json(silent=True) or {}
@@ -367,6 +491,7 @@ def rename_cell(project_name, cell_name):
if os.path.exists(old_path):
os.rename(old_path, new_path)
record_action('canvas.rename', project=project_name, cell=new_cell, detail={"old_cell": old_cell})
return jsonify({"message": "renamed", "old_cell": old_cell, "cell": new_cell})
@@ -387,6 +512,7 @@ def save_layout():
with open(save_path, 'w', encoding='utf-8') as f:
f.write(content)
record_action('layout.save', project=project, cell=cell, detail={"bytes": len(content)})
return jsonify({
"message": "successfully saved",
"project": project,
@@ -433,8 +559,11 @@ def getCompImg(component_name):
return jsonify({"error": "No image found"}), 404
if __name__ == '__main__':
print("Starting mxpic EDA Server on http://127.0.0.1:3000")
app.run(host='127.0.0.1', port=3000, debug=True)
host = os.environ.get('MXPIC_HOST', '0.0.0.0')
port = int(os.environ.get('MXPIC_PORT', '3000'))
debug = os.environ.get('MXPIC_DEBUG', '0').lower() in {'1', 'true', 'yes'}
print(f"Starting mxpic EDA Server on http://{host}:{port}")
app.run(host=host, port=port, debug=debug, threaded=True)