Initial commit
This commit is contained in:
952
simple_app.py
Normal file
952
simple_app.py
Normal file
@@ -0,0 +1,952 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
簡化版夥伴對齊系統 - 使用 SQLite
|
||||
無需 MySQL，開箱即用
|
||||
"""
|
||||
|
||||
from flask import Flask, render_template, request, jsonify
|
||||
from flask_cors import CORS
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from datetime import datetime, date
|
||||
import json
|
||||
import os
|
||||
import csv
|
||||
import io
|
||||
|
||||
# 創建 Flask 應用程式
|
||||
app = Flask(__name__)
|
||||
|
||||
# 配置
|
||||
app.config['SECRET_KEY'] = 'dev-secret-key-for-testing'
|
||||
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///partner_alignment.db'
|
||||
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||||
|
||||
# 初始化擴展
|
||||
db = SQLAlchemy(app)
|
||||
CORS(app, origins=['http://localhost:5000', 'http://127.0.0.1:5000'])
|
||||
|
||||
# 數據模型
|
||||
class User(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
username = db.Column(db.String(80), unique=True, nullable=False)
|
||||
email = db.Column(db.String(120), unique=True, nullable=False)
|
||||
full_name = db.Column(db.String(100), nullable=False)
|
||||
department = db.Column(db.String(50), nullable=False)
|
||||
position = db.Column(db.String(50), nullable=False)
|
||||
password_hash = db.Column(db.String(255), nullable=False)
|
||||
is_active = db.Column(db.Boolean, default=True)
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
|
||||
class Capability(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
name = db.Column(db.String(100), nullable=False)
|
||||
l1_description = db.Column(db.Text)
|
||||
l2_description = db.Column(db.Text)
|
||||
l3_description = db.Column(db.Text)
|
||||
l4_description = db.Column(db.Text)
|
||||
l5_description = db.Column(db.Text)
|
||||
is_active = db.Column(db.Boolean, default=True)
|
||||
|
||||
class DepartmentCapability(db.Model):
|
||||
"""部門與能力項目的關聯表"""
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
department = db.Column(db.String(50), nullable=False)
|
||||
capability_id = db.Column(db.Integer, db.ForeignKey('capability.id'), nullable=False)
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
|
||||
# 建立唯一索引，確保同一部門不會重複選擇同一能力項目
|
||||
__table_args__ = (db.UniqueConstraint('department', 'capability_id', name='uq_dept_capability'),)
|
||||
|
||||
class Assessment(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
|
||||
department = db.Column(db.String(50), nullable=False)
|
||||
position = db.Column(db.String(50), nullable=False)
|
||||
employee_name = db.Column(db.String(100))
|
||||
assessment_data = db.Column(db.Text)
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
|
||||
class StarFeedback(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
evaluator_name = db.Column(db.String(100), nullable=False)
|
||||
evaluatee_name = db.Column(db.String(100), nullable=False)
|
||||
evaluatee_department = db.Column(db.String(50), nullable=False)
|
||||
evaluatee_position = db.Column(db.String(50), nullable=False)
|
||||
situation = db.Column(db.Text, nullable=False)
|
||||
task = db.Column(db.Text, nullable=False)
|
||||
action = db.Column(db.Text, nullable=False)
|
||||
result = db.Column(db.Text, nullable=False)
|
||||
score = db.Column(db.Integer, nullable=False)
|
||||
points_earned = db.Column(db.Integer, nullable=False)
|
||||
feedback_date = db.Column(db.Date, default=date.today)
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
|
||||
class EmployeePoint(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
employee_name = db.Column(db.String(100), nullable=False)
|
||||
department = db.Column(db.String(50), nullable=False)
|
||||
position = db.Column(db.String(50), nullable=False)
|
||||
total_points = db.Column(db.Integer, default=0)
|
||||
monthly_points = db.Column(db.Integer, default=0)
|
||||
|
||||
# 路由
|
||||
@app.route('/')
|
||||
def index():
|
||||
return render_template('index.html')
|
||||
|
||||
@app.route('/api/auth/login', methods=['POST'])
|
||||
def login():
|
||||
"""用戶登入"""
|
||||
try:
|
||||
data = request.get_json()
|
||||
username = data.get('username')
|
||||
password = data.get('password')
|
||||
|
||||
if not username or not password:
|
||||
return jsonify({'error': '用戶名和密碼不能為空'}), 400
|
||||
|
||||
# 查找用戶
|
||||
user = User.query.filter_by(username=username).first()
|
||||
|
||||
if not user:
|
||||
return jsonify({'error': '用戶名或密碼錯誤'}), 401
|
||||
|
||||
# 簡化版：直接比較密碼（不安全，僅用於開發）
|
||||
# 注意：生產環境必須使用密碼哈希
|
||||
if user.password_hash != password:
|
||||
return jsonify({'error': '用戶名或密碼錯誤'}), 401
|
||||
|
||||
if not user.is_active:
|
||||
return jsonify({'error': '帳號已被停用'}), 403
|
||||
|
||||
# 返回用戶信息和簡單令牌
|
||||
return jsonify({
|
||||
'access_token': f'token_{user.id}_{datetime.now().timestamp()}',
|
||||
'refresh_token': f'refresh_{user.id}_{datetime.now().timestamp()}',
|
||||
'user': {
|
||||
'id': user.id,
|
||||
'username': user.username,
|
||||
'email': user.email,
|
||||
'full_name': user.full_name,
|
||||
'department': user.department,
|
||||
'position': user.position
|
||||
}
|
||||
}), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': f'登入失敗: {str(e)}'}), 500
|
||||
|
||||
@app.route('/api/auth/register', methods=['POST'])
|
||||
def register():
|
||||
"""用戶註冊"""
|
||||
try:
|
||||
data = request.get_json()
|
||||
|
||||
# 驗證必填欄位
|
||||
required_fields = ['username', 'email', 'password', 'full_name', 'department', 'position', 'employee_id']
|
||||
for field in required_fields:
|
||||
if not data.get(field):
|
||||
return jsonify({'error': f'{field} 是必填欄位'}), 400
|
||||
|
||||
# 檢查用戶名是否已存在
|
||||
if User.query.filter_by(username=data['username']).first():
|
||||
return jsonify({'error': '用戶名已存在'}), 409
|
||||
|
||||
# 檢查郵箱是否已存在
|
||||
if User.query.filter_by(email=data['email']).first():
|
||||
return jsonify({'error': '郵箱已被註冊'}), 409
|
||||
|
||||
# 創建新用戶
|
||||
user = User(
|
||||
username=data['username'],
|
||||
email=data['email'],
|
||||
full_name=data['full_name'],
|
||||
department=data['department'],
|
||||
position=data['position'],
|
||||
employee_id=data['employee_id'],
|
||||
password_hash=data['password'], # 簡化版：直接存儲密碼
|
||||
is_active=True
|
||||
)
|
||||
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'message': '註冊成功',
|
||||
'user': {
|
||||
'id': user.id,
|
||||
'username': user.username,
|
||||
'email': user.email,
|
||||
'full_name': user.full_name
|
||||
}
|
||||
}), 201
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({'error': f'註冊失敗: {str(e)}'}), 500
|
||||
|
||||
@app.route('/api/auth/protected', methods=['GET'])
|
||||
def protected():
|
||||
"""受保護的路由（用於驗證令牌）"""
|
||||
# 簡化版：不做實際驗證，返回默認用戶信息
|
||||
# 在實際應用中，這裡應該驗證 JWT 令牌並返回對應用戶信息
|
||||
return jsonify({
|
||||
'message': 'Access granted',
|
||||
'logged_in_as': 'admin', # 默認用戶
|
||||
'roles': ['admin'] # 默認角色
|
||||
}), 200
|
||||
|
||||
@app.route('/api/capabilities', methods=['GET'])
|
||||
def get_capabilities():
|
||||
"""獲取所有啟用的能力項目"""
|
||||
capabilities = Capability.query.filter_by(is_active=True).all()
|
||||
return jsonify({
|
||||
'capabilities': [{
|
||||
'id': cap.id,
|
||||
'name': cap.name,
|
||||
'l1_description': cap.l1_description,
|
||||
'l2_description': cap.l2_description,
|
||||
'l3_description': cap.l3_description,
|
||||
'l4_description': cap.l4_description,
|
||||
'l5_description': cap.l5_description
|
||||
} for cap in capabilities]
|
||||
})
|
||||
|
||||
@app.route('/api/department-capabilities/<department>', methods=['GET'])
|
||||
def get_department_capabilities(department):
|
||||
"""獲取特定部門選擇的能力項目"""
|
||||
# 查詢該部門已選擇的能力項目ID
|
||||
dept_caps = DepartmentCapability.query.filter_by(department=department).all()
|
||||
capability_ids = [dc.capability_id for dc in dept_caps]
|
||||
|
||||
# 獲取對應的能力項目詳細信息
|
||||
capabilities = Capability.query.filter(
|
||||
Capability.id.in_(capability_ids),
|
||||
Capability.is_active == True
|
||||
).all()
|
||||
|
||||
return jsonify({
|
||||
'department': department,
|
||||
'capabilities': [{
|
||||
'id': cap.id,
|
||||
'name': cap.name,
|
||||
'l1_description': cap.l1_description,
|
||||
'l2_description': cap.l2_description,
|
||||
'l3_description': cap.l3_description,
|
||||
'l4_description': cap.l4_description,
|
||||
'l5_description': cap.l5_description
|
||||
} for cap in capabilities]
|
||||
})
|
||||
|
||||
@app.route('/api/department-capabilities/<department>', methods=['POST'])
|
||||
def set_department_capabilities(department):
|
||||
"""設定部門的能力項目（部門主管使用）"""
|
||||
data = request.get_json()
|
||||
capability_ids = data.get('capability_ids', [])
|
||||
|
||||
if not capability_ids:
|
||||
return jsonify({'error': '請至少選擇一個能力項目'}), 400
|
||||
|
||||
try:
|
||||
# 先刪除該部門現有的所有能力項目關聯
|
||||
DepartmentCapability.query.filter_by(department=department).delete()
|
||||
|
||||
# 添加新的能力項目關聯
|
||||
for cap_id in capability_ids:
|
||||
dept_cap = DepartmentCapability(
|
||||
department=department,
|
||||
capability_id=cap_id
|
||||
)
|
||||
db.session.add(dept_cap)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'message': f'{department}部門的能力項目設定成功',
|
||||
'capability_count': len(capability_ids)
|
||||
}), 200
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/api/capabilities/import-csv', methods=['POST'])
|
||||
def import_capabilities_csv():
|
||||
"""從 CSV 檔案匯入能力項目"""
|
||||
if 'file' not in request.files:
|
||||
return jsonify({'error': '未上傳檔案'}), 400
|
||||
|
||||
file = request.files['file']
|
||||
|
||||
if file.filename == '':
|
||||
return jsonify({'error': '檔案名稱為空'}), 400
|
||||
|
||||
if not file.filename.endswith('.csv'):
|
||||
return jsonify({'error': '請上傳 CSV 檔案'}), 400
|
||||
|
||||
try:
|
||||
# 讀取 CSV 檔案
|
||||
stream = io.StringIO(file.stream.read().decode("UTF-8"), newline=None)
|
||||
csv_reader = csv.DictReader(stream)
|
||||
|
||||
imported_count = 0
|
||||
skipped_count = 0
|
||||
errors = []
|
||||
|
||||
for row_num, row in enumerate(csv_reader, start=2): # start=2 因為第1行是標題
|
||||
try:
|
||||
# 驗證必要欄位
|
||||
if not row.get('name'):
|
||||
errors.append(f"第 {row_num} 行：能力名稱不能為空")
|
||||
skipped_count += 1
|
||||
continue
|
||||
|
||||
# 檢查是否已存在相同名稱的能力項目
|
||||
existing = Capability.query.filter_by(name=row['name']).first()
|
||||
|
||||
if existing:
|
||||
# 更新現有能力項目
|
||||
existing.l1_description = row.get('l1_description', '')
|
||||
existing.l2_description = row.get('l2_description', '')
|
||||
existing.l3_description = row.get('l3_description', '')
|
||||
existing.l4_description = row.get('l4_description', '')
|
||||
existing.l5_description = row.get('l5_description', '')
|
||||
existing.is_active = True
|
||||
else:
|
||||
# 建立新的能力項目
|
||||
capability = Capability(
|
||||
name=row['name'],
|
||||
l1_description=row.get('l1_description', ''),
|
||||
l2_description=row.get('l2_description', ''),
|
||||
l3_description=row.get('l3_description', ''),
|
||||
l4_description=row.get('l4_description', ''),
|
||||
l5_description=row.get('l5_description', ''),
|
||||
is_active=True
|
||||
)
|
||||
db.session.add(capability)
|
||||
|
||||
imported_count += 1
|
||||
|
||||
except Exception as e:
|
||||
errors.append(f"第 {row_num} 行發生錯誤：{str(e)}")
|
||||
skipped_count += 1
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'message': f'匯入完成：成功 {imported_count} 筆，跳過 {skipped_count} 筆',
|
||||
'imported_count': imported_count,
|
||||
'skipped_count': skipped_count,
|
||||
'errors': errors
|
||||
}), 200
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({'error': f'匯入失敗：{str(e)}'}), 500
|
||||
|
||||
@app.route('/api/assessments', methods=['POST'])
|
||||
def create_assessment():
|
||||
data = request.get_json()
|
||||
|
||||
# 驗證必填欄位
|
||||
required_fields = ['department', 'position', 'assessment_data']
|
||||
for field in required_fields:
|
||||
if field not in data:
|
||||
return jsonify({'error': f'Missing required field: {field}'}), 400
|
||||
|
||||
# 創建新評估
|
||||
assessment = Assessment(
|
||||
user_id=1, # 簡化版使用固定用戶ID
|
||||
department=data['department'],
|
||||
position=data['position'],
|
||||
employee_name=data.get('employee_name'),
|
||||
assessment_data=json.dumps(data['assessment_data'])
|
||||
)
|
||||
|
||||
db.session.add(assessment)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'message': '評估資料儲存成功',
|
||||
'assessment_id': assessment.id
|
||||
}), 201
|
||||
|
||||
@app.route('/api/assessments', methods=['GET'])
|
||||
def get_assessments():
|
||||
assessments = Assessment.query.order_by(Assessment.created_at.desc()).all()
|
||||
|
||||
result = []
|
||||
for assessment in assessments:
|
||||
assessment_data = json.loads(assessment.assessment_data) if isinstance(assessment.assessment_data, str) else assessment.assessment_data
|
||||
result.append({
|
||||
'id': assessment.id,
|
||||
'department': assessment.department,
|
||||
'position': assessment.position,
|
||||
'employee_name': assessment.employee_name,
|
||||
'assessment_data': assessment_data,
|
||||
'created_at': assessment.created_at.isoformat()
|
||||
})
|
||||
|
||||
return jsonify({'assessments': result})
|
||||
|
||||
@app.route('/api/star-feedbacks', methods=['POST'])
|
||||
def create_star_feedback():
|
||||
data = request.get_json()
|
||||
|
||||
# 驗證必填欄位
|
||||
required_fields = ['evaluator_name', 'evaluatee_name', 'evaluatee_department',
|
||||
'evaluatee_position', 'situation', 'task', 'action', 'result', 'score']
|
||||
for field in required_fields:
|
||||
if field not in data:
|
||||
return jsonify({'error': f'Missing required field: {field}'}), 400
|
||||
|
||||
# 計算點數
|
||||
score = data['score']
|
||||
points_earned = score * 10
|
||||
|
||||
# 創建 STAR 回饋
|
||||
feedback = StarFeedback(
|
||||
evaluator_name=data['evaluator_name'],
|
||||
evaluatee_name=data['evaluatee_name'],
|
||||
evaluatee_department=data['evaluatee_department'],
|
||||
evaluatee_position=data['evaluatee_position'],
|
||||
situation=data['situation'],
|
||||
task=data['task'],
|
||||
action=data['action'],
|
||||
result=data['result'],
|
||||
score=score,
|
||||
points_earned=points_earned
|
||||
)
|
||||
|
||||
db.session.add(feedback)
|
||||
|
||||
# 更新員工積分
|
||||
employee = EmployeePoint.query.filter_by(employee_name=data['evaluatee_name']).first()
|
||||
if employee:
|
||||
employee.total_points += points_earned
|
||||
employee.monthly_points += points_earned
|
||||
else:
|
||||
employee = EmployeePoint(
|
||||
employee_name=data['evaluatee_name'],
|
||||
department=data['evaluatee_department'],
|
||||
position=data['evaluatee_position'],
|
||||
total_points=points_earned,
|
||||
monthly_points=points_earned
|
||||
)
|
||||
db.session.add(employee)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'message': '回饋資料儲存成功',
|
||||
'points_earned': points_earned
|
||||
}), 201
|
||||
|
||||
@app.route('/api/star-feedbacks', methods=['GET'])
|
||||
def get_star_feedbacks():
|
||||
feedbacks = StarFeedback.query.order_by(StarFeedback.created_at.desc()).all()
|
||||
|
||||
result = []
|
||||
for feedback in feedbacks:
|
||||
result.append({
|
||||
'id': feedback.id,
|
||||
'evaluator_name': feedback.evaluator_name,
|
||||
'evaluatee_name': feedback.evaluatee_name,
|
||||
'evaluatee_department': feedback.evaluatee_department,
|
||||
'evaluatee_position': feedback.evaluatee_position,
|
||||
'situation': feedback.situation,
|
||||
'task': feedback.task,
|
||||
'action': feedback.action,
|
||||
'result': feedback.result,
|
||||
'score': feedback.score,
|
||||
'points_earned': feedback.points_earned,
|
||||
'feedback_date': feedback.feedback_date.isoformat(),
|
||||
'created_at': feedback.created_at.isoformat()
|
||||
})
|
||||
|
||||
return jsonify({'feedbacks': result})
|
||||
|
||||
@app.route('/api/dashboard/me', methods=['GET'])
|
||||
def get_dashboard_data():
|
||||
"""獲取個人儀表板數據"""
|
||||
# 模擬用戶數據（簡化版）
|
||||
user_points = EmployeePoint.query.first()
|
||||
|
||||
if not user_points:
|
||||
return jsonify({
|
||||
'points_summary': {
|
||||
'total_points': 0,
|
||||
'monthly_points': 0,
|
||||
'department_rank': 0,
|
||||
'total_rank': 0
|
||||
},
|
||||
'recent_activities': [],
|
||||
'achievements': [],
|
||||
'performance_data': []
|
||||
})
|
||||
|
||||
# 計算排名
|
||||
total_employees = EmployeePoint.query.count()
|
||||
better_employees = EmployeePoint.query.filter(EmployeePoint.total_points > user_points.total_points).count()
|
||||
total_rank = better_employees + 1
|
||||
|
||||
# 模擬最近活動
|
||||
recent_activities = [
|
||||
{
|
||||
'type': 'assessment',
|
||||
'title': '完成能力評估',
|
||||
'description': '對溝通能力進行了評估',
|
||||
'points': 20,
|
||||
'created_at': '2024-01-15T10:30:00Z'
|
||||
},
|
||||
{
|
||||
'type': 'feedback',
|
||||
'title': '收到STAR回饋',
|
||||
'description': '來自同事的正面回饋',
|
||||
'points': 15,
|
||||
'created_at': '2024-01-14T14:20:00Z'
|
||||
}
|
||||
]
|
||||
|
||||
# 模擬成就
|
||||
achievements = [
|
||||
{
|
||||
'name': '評估達人',
|
||||
'description': '完成10次能力評估',
|
||||
'icon': 'clipboard-check'
|
||||
},
|
||||
{
|
||||
'name': '回饋專家',
|
||||
'description': '提供5次STAR回饋',
|
||||
'icon': 'star-fill'
|
||||
}
|
||||
]
|
||||
|
||||
# 模擬績效數據
|
||||
performance_data = [
|
||||
{'month': '1月', 'points': 50},
|
||||
{'month': '2月', 'points': 75},
|
||||
{'month': '3月', 'points': 60},
|
||||
{'month': '4月', 'points': 90},
|
||||
{'month': '5月', 'points': 80}
|
||||
]
|
||||
|
||||
return jsonify({
|
||||
'points_summary': {
|
||||
'total_points': user_points.total_points,
|
||||
'monthly_points': user_points.monthly_points,
|
||||
'department_rank': 1,
|
||||
'total_rank': total_rank
|
||||
},
|
||||
'recent_activities': recent_activities,
|
||||
'achievements': achievements,
|
||||
'performance_data': performance_data
|
||||
})
|
||||
|
||||
@app.route('/api/rankings/total', methods=['GET'])
|
||||
def get_total_rankings():
|
||||
department = request.args.get('department')
|
||||
limit = request.args.get('limit', 50, type=int)
|
||||
|
||||
# 構建查詢
|
||||
query = EmployeePoint.query
|
||||
if department:
|
||||
query = query.filter(EmployeePoint.department == department)
|
||||
|
||||
employees = query.order_by(EmployeePoint.total_points.desc()).limit(limit).all()
|
||||
total_count = query.count()
|
||||
|
||||
rankings = []
|
||||
for rank, employee in enumerate(employees, 1):
|
||||
# 計算百分位數
|
||||
percentile = ((total_count - rank + 1) / total_count * 100) if total_count > 0 else 0
|
||||
|
||||
rankings.append({
|
||||
'rank': rank,
|
||||
'employee_name': employee.employee_name,
|
||||
'department': employee.department,
|
||||
'position': employee.position,
|
||||
'total_points': employee.total_points,
|
||||
'monthly_points': employee.monthly_points,
|
||||
'percentile': round(percentile, 1),
|
||||
'tier': get_tier_by_percentile(percentile)
|
||||
})
|
||||
|
||||
return jsonify({
|
||||
'rankings': rankings,
|
||||
'total_count': total_count,
|
||||
'department_filter': department
|
||||
})
|
||||
|
||||
@app.route('/api/rankings/advanced', methods=['GET'])
|
||||
def get_advanced_rankings():
|
||||
"""高級排名系統，包含更多統計信息"""
|
||||
department = request.args.get('department')
|
||||
position = request.args.get('position')
|
||||
min_points = request.args.get('min_points', 0, type=int)
|
||||
max_points = request.args.get('max_points', type=int)
|
||||
|
||||
# 構建查詢
|
||||
query = EmployeePoint.query
|
||||
if department:
|
||||
query = query.filter(EmployeePoint.department == department)
|
||||
if position:
|
||||
query = query.filter(EmployeePoint.position.like(f'%{position}%'))
|
||||
if min_points:
|
||||
query = query.filter(EmployeePoint.total_points >= min_points)
|
||||
if max_points:
|
||||
query = query.filter(EmployeePoint.total_points <= max_points)
|
||||
|
||||
employees = query.order_by(EmployeePoint.total_points.desc()).all()
|
||||
total_count = len(employees)
|
||||
|
||||
if total_count == 0:
|
||||
return jsonify({
|
||||
'rankings': [],
|
||||
'statistics': {},
|
||||
'filters': {
|
||||
'department': department,
|
||||
'position': position,
|
||||
'min_points': min_points,
|
||||
'max_points': max_points
|
||||
}
|
||||
})
|
||||
|
||||
# 計算統計信息
|
||||
points_list = [emp.total_points for emp in employees]
|
||||
avg_points = sum(points_list) / len(points_list)
|
||||
median_points = sorted(points_list)[len(points_list) // 2]
|
||||
max_points_val = max(points_list)
|
||||
min_points_val = min(points_list)
|
||||
|
||||
rankings = []
|
||||
for rank, employee in enumerate(employees, 1):
|
||||
percentile = ((total_count - rank + 1) / total_count * 100) if total_count > 0 else 0
|
||||
|
||||
rankings.append({
|
||||
'rank': rank,
|
||||
'employee_name': employee.employee_name,
|
||||
'department': employee.department,
|
||||
'position': employee.position,
|
||||
'total_points': employee.total_points,
|
||||
'monthly_points': employee.monthly_points,
|
||||
'percentile': round(percentile, 1),
|
||||
'tier': get_tier_by_percentile(percentile),
|
||||
'vs_average': round(employee.total_points - avg_points, 1),
|
||||
'vs_median': round(employee.total_points - median_points, 1)
|
||||
})
|
||||
|
||||
return jsonify({
|
||||
'rankings': rankings,
|
||||
'statistics': {
|
||||
'total_count': total_count,
|
||||
'average_points': round(avg_points, 1),
|
||||
'median_points': median_points,
|
||||
'max_points': max_points_val,
|
||||
'min_points': min_points_val,
|
||||
'standard_deviation': round(calculate_standard_deviation(points_list), 1)
|
||||
},
|
||||
'filters': {
|
||||
'department': department,
|
||||
'position': position,
|
||||
'min_points': min_points,
|
||||
'max_points': max_points
|
||||
}
|
||||
})
|
||||
|
||||
def get_tier_by_percentile(percentile):
|
||||
"""根據百分位數返回等級"""
|
||||
if percentile >= 95:
|
||||
return {'name': '大師', 'color': 'danger', 'icon': 'crown'}
|
||||
elif percentile >= 85:
|
||||
return {'name': '專家', 'color': 'warning', 'icon': 'star-fill'}
|
||||
elif percentile >= 70:
|
||||
return {'name': '熟練', 'color': 'info', 'icon': 'award'}
|
||||
elif percentile >= 50:
|
||||
return {'name': '良好', 'color': 'success', 'icon': 'check-circle'}
|
||||
else:
|
||||
return {'name': '基礎', 'color': 'secondary', 'icon': 'circle'}
|
||||
|
||||
def calculate_standard_deviation(data):
|
||||
"""計算標準差"""
|
||||
if len(data) <= 1:
|
||||
return 0
|
||||
|
||||
mean = sum(data) / len(data)
|
||||
variance = sum((x - mean) ** 2 for x in data) / (len(data) - 1)
|
||||
return variance ** 0.5
|
||||
|
||||
@app.route('/api/notifications', methods=['GET'])
|
||||
def get_notifications():
|
||||
"""獲取用戶通知"""
|
||||
# 模擬通知數據
|
||||
notifications = [
|
||||
{
|
||||
'id': 1,
|
||||
'type': 'achievement',
|
||||
'title': '🎉 恭喜獲得新成就！',
|
||||
'message': '您已完成10次能力評估，獲得「評估達人」徽章',
|
||||
'is_read': False,
|
||||
'created_at': '2024-01-15T10:30:00Z',
|
||||
'icon': 'trophy-fill',
|
||||
'color': 'warning'
|
||||
},
|
||||
{
|
||||
'id': 2,
|
||||
'type': 'ranking',
|
||||
'title': '📈 排名更新',
|
||||
'message': '您的總排名上升了3位，目前排名第5名',
|
||||
'is_read': False,
|
||||
'created_at': '2024-01-14T15:20:00Z',
|
||||
'icon': 'arrow-up-circle',
|
||||
'color': 'success'
|
||||
},
|
||||
{
|
||||
'id': 3,
|
||||
'type': 'feedback',
|
||||
'title': '⭐ 收到新回饋',
|
||||
'message': '同事張三為您提供了STAR回饋，請查看詳情',
|
||||
'is_read': True,
|
||||
'created_at': '2024-01-13T09:15:00Z',
|
||||
'icon': 'star-fill',
|
||||
'color': 'info'
|
||||
},
|
||||
{
|
||||
'id': 4,
|
||||
'type': 'system',
|
||||
'title': '🔔 系統通知',
|
||||
'message': '新的評估項目已上線，歡迎體驗新功能',
|
||||
'is_read': True,
|
||||
'created_at': '2024-01-12T14:00:00Z',
|
||||
'icon': 'bell-fill',
|
||||
'color': 'primary'
|
||||
}
|
||||
]
|
||||
|
||||
return jsonify({
|
||||
'notifications': notifications,
|
||||
'unread_count': len([n for n in notifications if not n['is_read']])
|
||||
})
|
||||
|
||||
@app.route('/api/notifications/<int:notification_id>/read', methods=['POST'])
|
||||
def mark_notification_read(notification_id):
|
||||
"""標記通知為已讀"""
|
||||
# 在實際應用中，這裡會更新數據庫
|
||||
return jsonify({'success': True, 'message': '通知已標記為已讀'})
|
||||
|
||||
@app.route('/api/notifications/read-all', methods=['POST'])
|
||||
def mark_all_notifications_read():
|
||||
"""標記所有通知為已讀"""
|
||||
# 在實際應用中，這裡會更新數據庫
|
||||
return jsonify({'success': True, 'message': '所有通知已標記為已讀'})
|
||||
|
||||
@app.route('/api/admin/users', methods=['GET'])
|
||||
def get_admin_users():
|
||||
"""獲取所有用戶（管理員功能）"""
|
||||
users = User.query.all()
|
||||
|
||||
users_data = []
|
||||
for user in users:
|
||||
users_data.append({
|
||||
'id': user.id,
|
||||
'username': user.username,
|
||||
'email': user.email,
|
||||
'full_name': user.full_name,
|
||||
'department': user.department,
|
||||
'position': user.position,
|
||||
'employee_id': user.employee_id,
|
||||
'is_active': user.is_active,
|
||||
'created_at': user.created_at.isoformat() if user.created_at else None
|
||||
})
|
||||
|
||||
return jsonify({'users': users_data})
|
||||
|
||||
@app.route('/api/admin/users/<int:user_id>', methods=['PUT'])
|
||||
def update_user(user_id):
|
||||
"""更新用戶信息（管理員功能）"""
|
||||
user = User.query.get_or_404(user_id)
|
||||
data = request.get_json()
|
||||
|
||||
if 'is_active' in data:
|
||||
user.is_active = data['is_active']
|
||||
if 'department' in data:
|
||||
user.department = data['department']
|
||||
if 'position' in data:
|
||||
user.position = data['position']
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'success': True, 'message': '用戶信息已更新'})
|
||||
|
||||
@app.route('/api/admin/statistics', methods=['GET'])
|
||||
def get_admin_statistics():
|
||||
"""獲取管理員統計信息"""
|
||||
total_users = User.query.count()
|
||||
active_users = User.query.filter_by(is_active=True).count()
|
||||
total_assessments = Assessment.query.count()
|
||||
total_feedbacks = StarFeedback.query.count()
|
||||
|
||||
# 部門統計
|
||||
department_stats = db.session.query(
|
||||
User.department,
|
||||
db.func.count(User.id).label('count')
|
||||
).group_by(User.department).all()
|
||||
|
||||
# 積分統計
|
||||
points_stats = db.session.query(
|
||||
db.func.avg(EmployeePoint.total_points).label('avg_points'),
|
||||
db.func.max(EmployeePoint.total_points).label('max_points'),
|
||||
db.func.min(EmployeePoint.total_points).label('min_points')
|
||||
).first()
|
||||
|
||||
return jsonify({
|
||||
'total_users': total_users,
|
||||
'active_users': active_users,
|
||||
'total_assessments': total_assessments,
|
||||
'total_feedbacks': total_feedbacks,
|
||||
'department_stats': [{'department': d[0], 'count': d[1]} for d in department_stats],
|
||||
'points_stats': {
|
||||
'average': round(points_stats.avg_points or 0, 1),
|
||||
'maximum': points_stats.max_points or 0,
|
||||
'minimum': points_stats.min_points or 0
|
||||
}
|
||||
})
|
||||
|
||||
def create_sample_data():
|
||||
"""創建樣本數據"""
|
||||
# 創建能力項目
|
||||
capabilities = [
|
||||
Capability(
|
||||
name='溝通能力',
|
||||
l1_description='基本溝通',
|
||||
l2_description='有效溝通',
|
||||
l3_description='專業溝通',
|
||||
l4_description='領導溝通',
|
||||
l5_description='戰略溝通'
|
||||
),
|
||||
Capability(
|
||||
name='技術能力',
|
||||
l1_description='基礎技術',
|
||||
l2_description='熟練技術',
|
||||
l3_description='專業技術',
|
||||
l4_description='專家技術',
|
||||
l5_description='大師技術'
|
||||
),
|
||||
Capability(
|
||||
name='領導能力',
|
||||
l1_description='自我管理',
|
||||
l2_description='團隊協作',
|
||||
l3_description='團隊領導',
|
||||
l4_description='部門領導',
|
||||
l5_description='戰略領導'
|
||||
)
|
||||
]
|
||||
|
||||
for cap in capabilities:
|
||||
existing = Capability.query.filter_by(name=cap.name).first()
|
||||
if not existing:
|
||||
db.session.add(cap)
|
||||
|
||||
# 創建測試帳號
|
||||
test_accounts = [
|
||||
{
|
||||
'username': 'admin',
|
||||
'email': 'admin@company.com',
|
||||
'full_name': '系統管理員',
|
||||
'department': 'IT',
|
||||
'position': '系統管理員',
|
||||
'password_hash': 'admin123' # 簡化版直接存儲密碼
|
||||
},
|
||||
{
|
||||
'username': 'hr_manager',
|
||||
'email': 'hr@company.com',
|
||||
'full_name': 'HR主管',
|
||||
'department': 'HR',
|
||||
'position': '人力資源主管',
|
||||
'password_hash': 'hr123'
|
||||
},
|
||||
{
|
||||
'username': 'user',
|
||||
'email': 'user@company.com',
|
||||
'full_name': '一般用戶',
|
||||
'department': 'IT',
|
||||
'position': '軟體工程師',
|
||||
'password_hash': 'user123'
|
||||
}
|
||||
]
|
||||
|
||||
for account in test_accounts:
|
||||
existing_user = User.query.filter_by(username=account['username']).first()
|
||||
if not existing_user:
|
||||
user = User(**account)
|
||||
db.session.add(user)
|
||||
print(f"創建測試帳號: {account['username']}")
|
||||
|
||||
# 創建樣本積分數據
|
||||
sample_points_data = [
|
||||
{
|
||||
'employee_name': '系統管理員',
|
||||
'department': 'IT',
|
||||
'position': '系統管理員',
|
||||
'total_points': 150,
|
||||
'monthly_points': 50
|
||||
},
|
||||
{
|
||||
'employee_name': 'HR主管',
|
||||
'department': 'HR',
|
||||
'position': '人力資源主管',
|
||||
'total_points': 120,
|
||||
'monthly_points': 40
|
||||
},
|
||||
{
|
||||
'employee_name': '一般用戶',
|
||||
'department': 'IT',
|
||||
'position': '軟體工程師',
|
||||
'total_points': 80,
|
||||
'monthly_points': 30
|
||||
}
|
||||
]
|
||||
|
||||
for points_data in sample_points_data:
|
||||
existing = EmployeePoint.query.filter_by(employee_name=points_data['employee_name']).first()
|
||||
if not existing:
|
||||
points = EmployeePoint(**points_data)
|
||||
db.session.add(points)
|
||||
|
||||
db.session.commit()
|
||||
print("測試帳號和樣本數據創建完成")
|
||||
|
||||
if __name__ == '__main__':
|
||||
with app.app_context():
|
||||
# 創建所有數據庫表
|
||||
db.create_all()
|
||||
print("數據庫表創建成功！")
|
||||
|
||||
# 創建樣本數據
|
||||
create_sample_data()
|
||||
|
||||
print("=" * 60)
|
||||
print("夥伴對齊系統已啟動！")
|
||||
print("=" * 60)
|
||||
print("[WEB] 訪問地址: http://localhost:5000")
|
||||
print()
|
||||
print("[ACCOUNT] 測試帳號資訊:")
|
||||
print(" 管理員: admin / admin123")
|
||||
print(" HR主管: hr_manager / hr123")
|
||||
print(" 一般用戶: user / user123")
|
||||
print()
|
||||
print("[FEATURES] 功能包括:")
|
||||
print("- 能力評估系統")
|
||||
print("- STAR 回饋系統")
|
||||
print("- 排名系統")
|
||||
print("- 數據導出")
|
||||
print()
|
||||
print("[TIP] 提示: 登入頁面會顯示測試帳號資訊和快速登入按鈕")
|
||||
print("=" * 60)
|
||||
|
||||
# 啟動應用程式
|
||||
app.run(debug=True, host='0.0.0.0', port=5000)
|
||||
Reference in New Issue
Block a user