import os import json import uuid import logging from typing import Dict import requests from flask import Blueprint, request, jsonify, session, make_response, current_app, render_template, redirect, url_for from http import HTTPStatus # 导入数据库模型 from models import get_or_create_user, update_user_login_info, User, db # 认证蓝图 auth_bp = Blueprint('auth', __name__, url_prefix='/api/auth') logger = logging.getLogger(__name__) def get_dt_config() -> Dict[str, str]: """获取钉钉配置""" return { 'client_id': os.getenv('DINGTALK_CLIENT_ID', ''), 'client_secret': os.getenv('DINGTALK_CLIENT_SECRET', ''), 'redirect_uri': os.getenv('DINGTALK_REDIRECT_URI', '') } def dt_enabled() -> bool: """检查钉钉是否已配置""" cfg = get_dt_config() return all([cfg['client_id'], cfg['client_secret'], cfg['redirect_uri']]) def check_database_connection() -> bool: """检查数据库连接是否正常""" try: with current_app.app_context(): # 尝试执行一个简单的查询 from sqlalchemy import text db.session.execute(text('SELECT 1')) return True except Exception as e: logger.error(f'数据库连接检查失败: {e}') return False @auth_bp.route('/ping', methods=['GET']) def ping(): """健康检查接口""" return jsonify({ 'success': True, 'message': 'auth ok', 'dingtalk_enabled': dt_enabled() }) @auth_bp.route('/login_error') def login_error(): """登录错误页面""" error_type = request.args.get('error', 'unknown') error_message = request.args.get('message', '') error_code = request.args.get('code', '') return render_template('login_error.html', error_type=error_type, error_message=error_message, error_code=error_code) @auth_bp.route('/dingtalk/login_url', methods=['GET']) def dingtalk_login_url(): """生成钉钉登录URL""" if not dt_enabled(): return jsonify({'success': False, 'error': 'DingTalk未配置'}), HTTPStatus.BAD_REQUEST try: state = uuid.uuid4().hex session['dt_oauth_state'] = state cfg = get_dt_config() auth_url = ( 'https://login.dingtalk.com/oauth2/auth' f'?client_id={cfg["client_id"]}' f'&response_type=code' f'&scope=openid%20profile%20Contact.User.Read' f'&redirect_uri={requests.utils.quote(cfg["redirect_uri"], safe="")}' f'&state={state}' f'&prompt=consent' ) return jsonify({'success': True, 'url': auth_url}) except Exception as e: logger.error(f'生成钉钉登录URL失败: {e}') return jsonify({'success': False, 'error': '生成登录链接失败'}), HTTPStatus.INTERNAL_SERVER_ERROR @auth_bp.route('/dingtalk/callback', methods=['GET']) def dingtalk_oauth_callback(): """钉钉OAuth回调处理""" if not dt_enabled(): return redirect(url_for('auth.login_error', error='config_error', message='钉钉OAuth未正确配置，请联系管理员')) # 检查数据库连接 if not check_database_connection(): return redirect(url_for('auth.login_error', error='database_error', message='数据库连接失败，请稍后重试', code='DATABASE_CONNECTION_FAILED')) try: # 验证参数 code = request.args.get('code') state = request.args.get('state') if not code or not state or state != session.get('dt_oauth_state'): return redirect(url_for('auth.login_error', error='oauth_failed', message='OAuth认证参数无效或已过期', code='INVALID_PARAMS')) # 获取access token cfg = get_dt_config() try: token_resp = requests.post( 'https://api.dingtalk.com/v1.0/oauth2/userAccessToken', json={ 'clientId': cfg['client_id'], 'clientSecret': cfg['client_secret'], 'code': code, 'grantType': 'authorization_code' }, timeout=15 ) except requests.exceptions.Timeout: logger.error('获取access token超时') return redirect(url_for('auth.login_error', error='timeout', message='请求超时，请稍后重试', code='TOKEN_TIMEOUT')) except requests.exceptions.RequestException as e: logger.error(f'获取access token网络错误: {e}') return redirect(url_for('auth.login_error', error='network_error', message='网络连接异常，请检查网络后重试', code='NETWORK_ERROR')) if token_resp.status_code != 200: logger.error(f'获取access token失败: {token_resp.status_code}') return redirect(url_for('auth.login_error', error='token_failed', message=f'获取访问令牌失败 (HTTP {token_resp.status_code})', code=f'TOKEN_HTTP_{token_resp.status_code}')) token_data = token_resp.json() access_token = token_data.get('accessToken') if not access_token: logger.error(f"access_token为空: {token_data}") return redirect(url_for('auth.login_error', error='token_failed', message='获取到的访问令牌无效', code='EMPTY_TOKEN')) # 获取用户信息 user_info = get_dingtalk_user_info(access_token) if not user_info: return redirect(url_for('auth.login_error', error='userinfo_failed', message='无法获取用户信息，请确保钉钉账户信息完整', code='USERINFO_FAILED')) user_id = user_info.get('openid') or str(uuid.uuid4()) # 保存或更新用户信息到数据库 try: # 获取客户端IP client_ip = request.headers.get('X-Forwarded-For', request.remote_addr) if client_ip and ',' in client_ip: client_ip = client_ip.split(',')[0].strip() # 准备用户数据 user_display_name = user_info.get('name', '钉钉用户') user_data = { 'username': f"dingtalk_{user_id[:8]}", # 系统内部唯一标识 'nickname': user_display_name, # 用户显示名称 'avatar_url': user_info.get('avatar', ''), 'email': user_info.get('email', ''), 'phone': user_info.get('mobile', '') or None, # 空字符串转换为None 'dingtalk_id': user_id, 'dingtalk_unionid': user_info.get('openid', ''), 'openid': user_info.get('openid', ''), 'status': 'active' } # 获取或创建用户 db_user = get_or_create_user('dingtalk', **user_data) # 更新登录信息 update_user_login_info(db_user, client_ip) # 更新session中的用户信息，包含数据库ID user_info['db_user_id'] = db_user.user_id except Exception as e: logger.error(f'保存用户信息到数据库失败: {e}', exc_info=True) # 数据库操作失败，返回错误页面 return redirect(url_for('auth.login_error', error='database_error', message=f'用户信息保存失败: {str(e)}', code='DATABASE_SAVE_FAILED')) # 生成JWT token jwt_token = generate_jwt_token(user_id, user_info) if not jwt_token: return redirect(url_for('auth.login_error', error='jwt_failed', message='生成登录凭证失败，系统可能正在维护中', code='JWT_GENERATION_FAILED')) # 保存到session session['auth_token'] = jwt_token session['user_info'] = user_info # 直接跳转到首页，通过URL参数传递用户信息 import urllib.parse user_info_json = urllib.parse.quote(json.dumps(user_info, ensure_ascii=False)) redirect_url = f"/?login_success=true&token={jwt_token}&user_info={user_info_json}" return redirect(redirect_url) except Exception as e: logger.error(f'钉钉OAuth回调处理失败: {e}', exc_info=True) return redirect(url_for('auth.login_error', error='unknown', message=f'登录过程中发生未知错误: {str(e)}', code='CALLBACK_EXCEPTION')) def get_dingtalk_user_info(access_token: str) -> Dict: """获取钉钉用户信息""" try: # 使用正确的钉钉API端点 headers = { 'x-acs-dingtalk-access-token': access_token, 'Content-Type': 'application/json' } # 钉钉官方推荐的获取用户信息API api_url = 'https://api.dingtalk.com/v1.0/contact/users/me' resp = requests.get(api_url, headers=headers, timeout=15) if resp.status_code == 200: data = resp.json() # 检查是否有错误码 if not data.get('errcode') or data.get('errcode') == 0: user_info = parse_dingtalk_user_data(data) if user_info: return user_info else: logger.warning(f'钉钉API返回错误: {data.get("errcode")} - {data.get("errmsg")}') else: logger.warning(f'钉钉API调用失败: HTTP {resp.status_code}') # 如果API调用失败，返回默认用户信息 logger.warning('钉钉API调用失败，使用默认用户信息') return get_default_user_info() except Exception as e: logger.warning(f'获取钉钉用户信息失败: {e}') return get_default_user_info() def parse_dingtalk_user_data(data: Dict) -> Dict: """解析钉钉用户数据""" try: # 尝试多种可能的字段名 name_fields = ['nick', 'nickname', 'realName', 'name', 'nickName', 'userid', 'user_id'] name = None for field in name_fields: if data.get(field) and data.get(field).strip(): name = data.get(field).strip() break if not name: name = '钉钉用户' # 获取用户ID openid = (data.get('openId') or data.get('openid') or data.get('unionId') or data.get('unionid') or data.get('userid') or data.get('user_id') or str(uuid.uuid4())) # 获取头像 avatar_fields = ['avatarUrl', 'avatar_url', 'avatar', 'headimgurl', 'avatarUrl'] avatar = None for field in avatar_fields: if data.get(field) and data.get(field).strip(): avatar = data.get(field).strip() break # 如果头像URL是相对路径，转换为完整URL if avatar and not avatar.startswith(('http://', 'https://')): if avatar.startswith('/'): avatar = f'https://oapi.dingtalk.com{avatar}' else: avatar = f'https://oapi.dingtalk.com/{avatar}' # 获取其他信息 email = data.get('email') or data.get('mail') or '' mobile = data.get('mobile') or data.get('phone') or '' job_number = data.get('jobNumber') or data.get('job_number') or '' title = data.get('title') or data.get('position') or '' dept_name = data.get('deptName') or data.get('department_name') or '' result = { 'name': name, 'nickname': name, # 添加nickname字段，确保前端能正确显示 'openid': openid, 'avatar': avatar or '', 'email': email, 'mobile': mobile, 'job_number': job_number, 'title': title, 'dept_name': dept_name, 'login_type': 'dingtalk' } return result except Exception as e: logger.error(f'解析钉钉用户数据失败: {e}') return None def get_default_user_info() -> Dict: """获取默认用户信息""" return { 'name': '钉钉用户', 'nickname': '钉钉用户', # 添加nickname字段 'openid': str(uuid.uuid4()), 'avatar': '', 'email': '', 'mobile': '', 'job_number': '', 'title': '', 'dept_name': '', 'login_type': 'dingtalk' } def generate_jwt_token(user_id: str, user_info: Dict) -> str: """生成JWT token""" try: if not hasattr(current_app, 'auth_middleware'): logger.error('auth_middleware未配置') return None auth_middleware = current_app.auth_middleware if not auth_middleware: logger.error('auth_middleware为空') return None # 确保user_info包含必要的字段 if not user_info.get('name') and not user_info.get('nickname'): user_info['name'] = '钉钉用户' user_info['nickname'] = '钉钉用户' token = auth_middleware.generate_token(user_id, user_info) return token except Exception as e: logger.error(f'生成JWT token失败: {e}', exc_info=True) return None @auth_bp.route('/logout', methods=['POST']) def logout(): """退出登录""" try: session.clear() return jsonify({'success': True, 'message': '退出登录成功'}) except Exception as e: logger.error(f'退出登录失败: {e}') return jsonify({'success': False, 'error': '退出登录失败'}), HTTPStatus.INTERNAL_SERVER_ERROR @auth_bp.route('/verify', methods=['GET']) def verify_token(): """验证令牌""" try: token = request.headers.get('Authorization', '').replace('Bearer ', '') if not token: return jsonify({'success': False, 'error': '缺少令牌'}), HTTPStatus.UNAUTHORIZED auth_middleware = current_app.auth_middleware payload = auth_middleware.verify_token(token) return jsonify({ 'success': True, 'user_id': payload['user_id'], 'user_info': payload['user_info'] }) except ValueError as e: return jsonify({'success': False, 'error': str(e)}), HTTPStatus.UNAUTHORIZED except Exception as e: logger.error(f'验证令牌失败: {e}') return jsonify({'success': False, 'error': '验证失败'}), HTTPStatus.INTERNAL_SERVER_ERROR @auth_bp.route('/user/info', methods=['GET']) def get_user_info(): """获取用户信息""" try: token = request.headers.get('Authorization', '').replace('Bearer ', '') if not token: return jsonify({'success': False, 'error': '缺少令牌'}), HTTPStatus.UNAUTHORIZED auth_middleware = current_app.auth_middleware payload = auth_middleware.verify_token(token) return jsonify({'success': True, 'user_info': payload['user_info']}) except ValueError as e: return jsonify({'success': False, 'error': str(e)}), HTTPStatus.UNAUTHORIZED except Exception as e: logger.error(f'获取用户信息失败: {e}') return jsonify({'success': False, 'error': '获取用户信息失败'}), HTTPStatus.INTERNAL_SERVER_ERROR @auth_bp.route('/refresh', methods=['POST']) def refresh_token(): """刷新JWT令牌""" try: token = request.headers.get('Authorization', '').replace('Bearer ', '') if not token: logger.warning('刷新token请求缺少Authorization头') return jsonify({ 'success': False, 'error': '缺少令牌', 'code': 'MISSING_TOKEN' }), HTTPStatus.UNAUTHORIZED auth_middleware = current_app.auth_middleware try: # 验证当前token payload = auth_middleware.verify_token(token) user_id = payload['user_id'] user_info = payload['user_info'] # 生成新的token new_token = auth_middleware.generate_token(user_id, user_info) logger.info(f'Token刷新成功，用户ID: {user_id}') return jsonify({ 'success': True, 'token': new_token, 'message': 'Token刷新成功' }) except ValueError as e: logger.warning(f'Token验证失败: {str(e)}') # Token已过期或无效，尝试从session获取用户信息 user_info = session.get('user_info') if user_info: user_id = user_info.get('openid') or str(uuid.uuid4()) new_token = auth_middleware.generate_token(user_id, user_info) logger.info(f'从session刷新token成功，用户ID: {user_id}') return jsonify({ 'success': True, 'token': new_token, 'message': 'Token刷新成功' }) else: logger.warning('Token已过期且session中无用户信息，需要重新登录') return jsonify({ 'success': False, 'error': 'Token已过期且无法刷新，请重新登录', 'code': 'TOKEN_EXPIRED_NO_SESSION' }), HTTPStatus.UNAUTHORIZED except Exception as e: logger.error(f'刷新Token失败: {e}', exc_info=True) return jsonify({ 'success': False, 'error': '刷新Token失败', 'code': 'REFRESH_ERROR' }), HTTPStatus.INTERNAL_SERVER_ERROR