jude-cehua/auth_routes.py

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