import secrets from functools import wraps from flask import request, jsonify from .database import get_db def generate_token(): return secrets.token_urlsafe(32) def admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'success': False, 'error': '请先登录'}), 401 token = auth_header.split(' ')[1] conn = get_db() cursor = conn.cursor() cursor.execute(''' SELECT a.* FROM admin_users a JOIN admin_sessions s ON a.id = s.admin_id WHERE s.token = ? AND s.expires_at > datetime('now') ''', (token,)) admin = cursor.fetchone() conn.close() if not admin: return jsonify({'success': False, 'error': '登录已过期,请重新登录'}), 401 request.admin = dict(admin) return f(*args, **kwargs) return decorated_function def get_user_from_token(token): if not token: return None conn = get_db() cursor = conn.cursor() cursor.execute(''' SELECT u.* FROM users u JOIN sessions s ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > datetime('now') AND u.deleted_at IS NULL ''', (token,)) row = cursor.fetchone() conn.close() if not row: return None user = dict(row) user['location'] = build_location(user) return user def get_user_id_from_token(token): if not token: return None conn = get_db() cursor = conn.cursor() cursor.execute(''' SELECT u.id FROM users u JOIN sessions s ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > datetime('now') AND u.deleted_at IS NULL ''', (token,)) row = cursor.fetchone() conn.close() return row['id'] if row else None def get_admin_from_token(token=None, user_id=None): """ 通过token或user_id检查是否是管理员 如果是管理员返回True,否则返回False """ conn = get_db() cursor = conn.cursor() try: if token: cursor.execute(''' SELECT a.id FROM admin_users a JOIN admin_sessions s ON a.id = s.admin_id WHERE s.token = ? AND s.expires_at > datetime('now') ''', (token,)) elif user_id: cursor.execute(''' SELECT a.id FROM admin_users a WHERE a.id = ? ''', (user_id,)) else: return False row = cursor.fetchone() return row is not None finally: conn.close() def build_location(data, include_detail=False): location_parts = [] if data.get('province'): location_parts.append(data['province']) if data.get('city'): location_parts.append(data['city']) if data.get('district'): location_parts.append(data['district']) if data.get('street'): location_parts.append(data['street']) if include_detail and data.get('detail_location'): location_parts.append(data['detail_location']) return '/'.join(location_parts) if location_parts else None