PinHung
f690604c4a
Some checks failed
CI/CD Pipeline / test (3.10) (push) Has been cancelled
CI/CD Pipeline / test (3.11) (push) Has been cancelled
CI/CD Pipeline / test (3.8) (push) Has been cancelled
CI/CD Pipeline / test (3.9) (push) Has been cancelled
CI/CD Pipeline / security (push) Has been cancelled
CI/CD Pipeline / build (push) Has been cancelled
CI/CD Pipeline / deploy-staging (push) Has been cancelled
CI/CD Pipeline / deploy-production (push) Has been cancelled
954 lines
33 KiB
Python
954 lines
33 KiB
Python
#!/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)
|
||
employee_id = db.Column(db.String(50), unique=True)
|
||
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)
|