# app/auth/routes.py from flask import render_template, redirect, url_for, flash, request, jsonify, session, current_app, send_file from flask_login import login_user, logout_user, login_required, current_user from urllib.parse import urlparse, urljoin from datetime import datetime, timedelta import random import string import json import os import logging import hashlib import hmac import base64 from functools import wraps from sqlalchemy.exc import IntegrityError, SQLAlchemyError from io import BytesIO from typing import Optional from flask import Response, render_template, flash, redirect, url_for from app.models import User from app.extensions import db from app.auth import auth_bp from app.auth.forms import * from app.extensions import db from app.models import User, Project, ProjectEmployee from app.helpers import get_dashboard_url, redirect_to_dashboard, get_role_persian from app.auth import auth_bp from app.models import * import phonenumbers from phonenumbers import carrier, geocoder, timezone from app.models import User, Notification import re from pytz import timezone as pytz_timezone # تنظیم لاگر logger = logging.getLogger(__name__) # ============================================ # ثابت‌ها و تنظیمات # ============================================ TEMP_PASSWORD_EXPIRY_MINUTES = 5 MAX_LOGIN_ATTEMPTS = 5 LOGIN_ATTEMPT_WINDOW = 15 # دقیقه GPS_MAX_DISTANCE = 50 # حداکثر ۵۰ متر فاصله مجاز FACE_MATCH_THRESHOLD = 0.85 # ۸۵% تطابق چهره # ============================================ # توابع کمکی # ============================================ def send_sms(phone, code, template="verify"): """ارسال کد تایید به شماره موبایل""" logger.info(f"ارسال کد تایید به شماره {phone} - کد: {code} - قالب: {template}") return True def send_password_via_sms(phone, password): """ارسال رمز عبور از طریق پیامک""" logger.info(f"ارسال رمز عبور به {phone}: {password}") return True def send_temp_password_via_sms(phone, password): """ارسال رمز موقت از طریق پیامک""" message = f"🔑 رمز موقت شما: {password}\n⏱ این رمز به مدت {TEMP_PASSWORD_EXPIRY_MINUTES} دقیقه معتبر است." logger.info(f"ارسال رمز موقت به {phone} - رمز: {password}") print(f"=========================================") print(f"📱 ارسال رمز موقت به {phone}") print(f"🔑 رمز موقت: {password}") print(f"📝 پیام: {message}") print(f"=========================================") return True def generate_16_digit_code(): """تولید کد ۱۶ رقمی یکتا""" return ''.join(random.choices(string.digits, k=16)) def generate_temp_password(): """تولید رمز موقت ۶ رقمی عددی""" return ''.join(random.choices(string.digits, k=6)) def is_safe_url(target): """ بررسی امن بودن URL برای ریدایرکت Args: target: URL مورد نظر برای بررسی Returns: bool: True اگر URL امن باشد، False در غیر این صورت """ if not target: return False # اگر target یک رشته خالی یا None باشد if not isinstance(target, str): return False # جلوگیری از ریدایرکت به سایت‌های خارجی ref_url = urlparse(request.host_url) test_url = urlparse(urljoin(request.host_url, target)) # بررسی پروتکل و دامنه return (test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc) def verify_face(face_data, user_id=None): """تشخیص چهره با استفاده از سرویس تشخیص چهره""" if not face_data: return False logger.info(f"بررسی چهره برای {'کاربر موقت' if not user_id else f'کاربر {user_id}'}") try: # بررسی وجود داده if len(face_data) < 100: # حداقل طول داده return False # در محیط واقعی، اینجا باید سرویس تشخیص چهره فراخوانی شود # مثلاً با استفاده از AWS Rekognition، Face++، یا OpenCV # بررسی تطابق با چهره ذخیره‌شده if user_id: user = User.query.get(user_id) if user and user.face_encoding: # در محیط واقعی، اینجا تطابق چهره انجام می‌شود # از کتابخانه face_recognition یا خدمات ابری استفاده کنید return True return True # در حالت شبیه‌سازی، همیشه تایید می‌شود except Exception as e: logger.error(f"خطا در تشخیص چهره: {e}") return False def verify_gps(lat, lon, project_id): """بررسی موقعیت GPS در محدوده پروژه""" if not lat or not lon: return False project = Project.query.get(project_id) if not project: return False if not project.gps_bounds: logger.info(f"محدوده GPS برای پروژه {project_id} تعریف نشده است - تایید خودکار") return True try: bounds = project.gps_bounds north = bounds.get('north') south = bounds.get('south') east = bounds.get('east') west = bounds.get('west') if north and south and east and west: is_inside = (south <= lat <= north) and (west <= lon <= east) logger.info(f"بررسی GPS: lat={lat}, lon={lon} - نتیجه: {is_inside}") return is_inside except Exception as e: logger.error(f"خطا در بررسی GPS: {e}") return True def calculate_gps_distance(lat1, lon1, lat2, lon2): """ محاسبه فاصله بین دو نقطه GPS با استفاده از فرمول هاورسین Args: lat1, lon1: مختصات نقطه اول (پروژه) lat2, lon2: مختصات نقطه دوم (کاربر) Returns: float: فاصله به متر """ from math import radians, sin, cos, sqrt, atan2 # شعاع زمین به متر R = 6371000 # تبدیل درجه به رادیان lat1_rad = radians(lat1) lat2_rad = radians(lat2) delta_lat = radians(lat2 - lat1) delta_lon = radians(lon2 - lon1) # فرمول هاورسین a = sin(delta_lat / 2) ** 2 + cos(lat1_rad) * cos(lat2_rad) * sin(delta_lon / 2) ** 2 c = 2 * atan2(sqrt(a), sqrt(1 - a)) distance = R * c return distance def get_user_by_phone(phone): """دریافت کاربر بر اساس شماره تلفن""" return User.query.filter_by(phone=phone).first() def get_project_by_code(project_code): """دریافت پروژه بر اساس کد ۱۶ رقمی""" return Project.query.filter_by(project_code=project_code).first() def clear_temp_password(user): """پاک کردن رمز موقت کاربر""" if user: user.temp_password = None user.temp_password_expires = None user.is_temp_password = False def handle_db_error(e, rollback=True): """مدیریت خطاهای دیتابیس""" if rollback: db.session.rollback() logger.error(f"خطای دیتابیس: {str(e)}") flash('خطا در ارتباط با دیتابیس. لطفاً مجدداً تلاش کنید.', 'danger') return False def login_user_handler(user, remember=True): """ تابع جامع برای ورود کاربر با مدیریت خطا """ try: if not user.is_active: flash('حساب کاربری شما غیرفعال شده است.', 'danger') return False, None if user.role == 'manager': if user.manager_request_status == 'pending': flash('درخواست شما در انتظار تایید ادمین است.', 'warning') return False, None elif user.manager_request_status == 'rejected': flash('درخواست شما رد شده است. با پشتیبانی تماس بگیرید.', 'danger') return False, None clear_temp_password(user) user.last_login = datetime.utcnow() db.session.commit() login_user(user, remember=remember) session.permanent = True session.modified = True logger.info(f"ورود موفق: {user.username} (نقش: {user.role}, ID: {user.id})") return True, user.role except SQLAlchemyError as e: db.session.rollback() logger.error(f"خطای دیتابیس در ورود کاربر: {e}") flash('خطا در ورود به سیستم. لطفاً مجدداً تلاش کنید.', 'danger') return False, None except Exception as e: logger.error(f"خطای پیش‌بینی‌نشده در ورود کاربر: {e}") flash('خطا در ورود به سیستم. لطفاً مجدداً تلاش کنید.', 'danger') return False, None # ============================================ # دکوریتورهای سفارشی # ============================================ def role_required(allowed_roles): """دکوریتور برای محدود کردن دسترسی بر اساس نقش""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not current_user.is_authenticated: flash('لطفاً وارد شوید.', 'warning') return redirect(url_for('auth.login')) if current_user.role not in allowed_roles: flash('شما دسترسی به این بخش را ندارید.', 'danger') return redirect(url_for('auth.dashboard')) return f(*args, **kwargs) return decorated_function return decorator # ============================================ # مسیرهای ورود # ============================================ @auth_bp.route('/login', methods=['GET']) def login(): """صفحه ورود عمومی - انتخاب نقش""" if current_user.is_authenticated: return redirect_to_dashboard(current_user) return render_template('auth/login.html') #===== # ============================================ # تنظیمات زمان # ============================================ # تنظیم منطقه زمانی ایران IRAN_TZ = pytz_timezone('Asia/Tehran') def get_current_time(): """دریافت زمان فعلی با منطقه زمانی ایران""" return datetime.now(IRAN_TZ) # ============================================ # ورود به صفحه درخواست تبلیغ (بدون نیاز به ثبت‌نام) # ============================================ @auth_bp.route('/advertisement-request-login', methods=['GET', 'POST']) def advertisement_request_login(): """ ورود با شماره تلفن و رمز موقت برای دسترسی به صفحه درخواست تبلیغ کاربر با هر شماره تلفنی می‌تواند وارد شود و در صورت عدم وجود، به‌صورت خودکار ثبت‌نام می‌شود """ # اگر کاربر لاگین است، مستقیماً به صفحه درخواست تبلیغ برود if current_user.is_authenticated: return redirect(url_for('public.advertisement_request')) form = AdvertisementLoginForm() if form.validate_on_submit(): phone_number = form.phone_number.data otp_code = form.otp_code.data # اگر کد OTP وارد شده باشد، اعتبارسنجی می‌کنیم if otp_code: # بررسی وجود کاربر با این شماره تلفن user = User.query.filter_by(phone=phone_number).first() # اگر کاربر وجود نداشت، به‌صورت خودکار ایجاد می‌کنیم if not user: user = create_user_automatically(phone_number) if not user: flash('خطا در ایجاد حساب کاربری. لطفاً مجدداً تلاش کنید.', 'danger') return render_template('auth/advertisement_request_login.html', form=form) # بررسی کد OTP در جلسه if 'otp_code' not in session or 'otp_phone' not in session: flash('لطفاً ابتدا کد تایید را دریافت کنید.', 'warning') return render_template('auth/advertisement_request_login.html', form=form) # ✅ بررسی انقضای کد با زمان‌های timezone-aware if 'otp_expiry' in session: # تبدیل زمان ذخیره شده در جلسه به timezone-aware otp_expiry = session['otp_expiry'] if isinstance(otp_expiry, str): # اگر به صورت رشته ذخیره شده بود، به datetime تبدیل می‌کنیم try: otp_expiry = datetime.fromisoformat(otp_expiry) except: otp_expiry = None # اگر otp_expiry وجود دارد و منقضی شده است if otp_expiry and get_current_time() > otp_expiry: flash('کد تایید منقضی شده است. لطفاً مجدداً درخواست کنید.', 'danger') session.pop('otp_code', None) session.pop('otp_phone', None) session.pop('otp_expiry', None) return render_template('auth/advertisement_request_login.html', form=form) # بررسی تطابق کد if session.get('otp_code') != otp_code: flash('کد تایید وارد شده صحیح نیست. لطفاً مجدداً تلاش کنید.', 'danger') return render_template('auth/advertisement_request_login.html', form=form) # بررسی تطابق شماره تلفن if session.get('otp_phone') != phone_number: flash('شماره تلفن با کد تایید مطابقت ندارد.', 'danger') return render_template('auth/advertisement_request_login.html', form=form) # ورود کاربر login_user(user) session.pop('otp_code', None) session.pop('otp_phone', None) session.pop('otp_expiry', None) # بررسی اینکه آیا کاربر جدید ایجاد شده است if hasattr(user, 'is_new_user') and user.is_new_user: flash(f'خوش آمدید {user.fullname}، حساب کاربری شما با موفقیت ایجاد شد. اکنون می‌توانید درخواست تبلیغ خود را ثبت کنید.', 'success') else: flash(f'خوش آمدید {user.fullname}، اکنون می‌توانید درخواست تبلیغ خود را ثبت کنید.', 'success') return redirect(url_for('public.advertisement_request')) # اگر کد OTP وارد نشده باشد، کد ارسال می‌کنیم else: # اعتبارسنجی شماره تلفن if not is_valid_phone_number(phone_number): flash('شماره تلفن وارد شده معتبر نیست.', 'danger') return render_template('auth/advertisement_request_login.html', form=form) # بررسی وجود کاربر - اگر نبود، اشکالی ندارد چون بعداً ایجاد می‌شود user = User.query.filter_by(phone=phone_number).first() # تولید کد OTP تصادفی otp_code = generate_otp_code() # ✅ ذخیره در جلسه با زمان timezone-aware session['otp_code'] = otp_code session['otp_phone'] = phone_number # ذخیره به صورت ISO format برای جلوگیری از مشکل pickle session['otp_expiry'] = (get_current_time() + timedelta(minutes=5)).isoformat() # ارسال کد از طریق SMS (در محیط توسعه، در کنسول نمایش داده می‌شود) send_otp_sms(phone_number, otp_code) if user: flash(f'کد تایید به شماره تلفن {phone_number} ارسال شد. لطفاً کد را وارد کنید.', 'success') else: flash(f'کد تایید به شماره تلفن {phone_number} ارسال شد. با وارد کردن کد، حساب کاربری شما به‌صورت خودکار ایجاد می‌شود.', 'info') return render_template('auth/advertisement_request_login.html', form=form, phone_number=phone_number, show_otp=True) return render_template('auth/advertisement_request_login.html', form=form) @auth_bp.route('/resend-advertisement-otp', methods=['POST']) def resend_advertisement_otp(): """ ارسال مجدد کد OTP برای ورود به صفحه درخواست تبلیغ """ phone_number = request.json.get('phone_number') if not phone_number: return jsonify({'success': False, 'message': 'شماره تلفن وارد نشده است.'}), 400 # اعتبارسنجی شماره تلفن if not is_valid_phone_number(phone_number): return jsonify({'success': False, 'message': 'شماره تلفن معتبر نیست.'}), 400 # تولید کد OTP جدید otp_code = generate_otp_code() # ✅ ذخیره در جلسه با زمان timezone-aware session['otp_code'] = otp_code session['otp_phone'] = phone_number session['otp_expiry'] = (get_current_time() + timedelta(minutes=5)).isoformat() # ارسال کد send_otp_sms(phone_number, otp_code) return jsonify({ 'success': True, 'message': 'کد تایید مجدداً ارسال شد.', 'phone_number': phone_number }) # ============================================ # توابع کمکی # ============================================ def generate_otp_code(length=6): """تولید کد OTP تصادفی""" return ''.join(random.choices(string.digits, k=length)) def is_valid_phone_number(phone_number): """ اعتبارسنجی شماره تلفن ایران پشتیبانی از فرمت‌های: - 09123456789 - ۹۱۲۳۴۵۶۷۸۹ (با اعداد فارسی) - +989123456789 - 00989123456789 """ try: # پاکسازی شماره تلفن phone_number = str(phone_number).strip() # تبدیل اعداد فارسی به انگلیسی persian_to_english = { '۰': '0', '۱': '1', '۲': '2', '۳': '3', '۴': '4', '۵': '5', '۶': '6', '۷': '7', '۸': '8', '۹': '9' } for persian, english in persian_to_english.items(): phone_number = phone_number.replace(persian, english) # حذف کاراکترهای غیرعددی phone_number = re.sub(r'[^\d+]', '', phone_number) # بررسی فرمت‌های مختلف patterns = [ r'^09\d{9}$', # 09123456789 r'^\+989\d{9}$', # +989123456789 r'^00989\d{9}$', # 00989123456789 r'^989\d{9}$', # 989123456789 r'^9\d{9}$' # 9123456789 ] for pattern in patterns: if re.match(pattern, phone_number): return True return False except Exception as e: print(f"Error validating phone number: {e}") return False def normalize_phone_number(phone_number): """ نرمال‌سازی شماره تلفن به فرمت استاندارد 09123456789 """ try: # تبدیل اعداد فارسی به انگلیسی persian_to_english = { '۰': '0', '۱': '1', '۲': '2', '۳': '3', '۴': '4', '۵': '5', '۶': '6', '۷': '7', '۸': '8', '۹': '9' } for persian, english in persian_to_english.items(): phone_number = phone_number.replace(persian, english) # حذف کاراکترهای غیرعددی phone_number = re.sub(r'[^\d]', '', phone_number) # نرمال‌سازی if phone_number.startswith('0098'): phone_number = '0' + phone_number[4:] elif phone_number.startswith('+98'): phone_number = '0' + phone_number[3:] elif phone_number.startswith('98'): phone_number = '0' + phone_number[2:] elif not phone_number.startswith('0') and len(phone_number) == 10: phone_number = '0' + phone_number return phone_number except Exception as e: print(f"Error normalizing phone number: {e}") return phone_number def create_user_automatically(phone_number): """ ایجاد خودکار کاربر جدید با شماره تلفن """ try: # نرمال‌سازی شماره تلفن phone_number = normalize_phone_number(phone_number) # بررسی مجدد وجود کاربر (برای جلوگیری از ایجاد تکراری) existing_user = User.query.filter_by(phone=phone_number).first() if existing_user: return existing_user # ایجاد نام کاربری از شماره تلفن username = f"user_{phone_number}" # ایجاد نام کامل پیش‌فرض fullname = f"کاربر {phone_number}" # ایجاد کاربر جدید new_user = User( username=username, fullname=fullname, phone=phone_number, role='employee', # نقش پیش‌فرض is_active=True ) # تنظیم رمز عبور تصادفی (برای امنیت) from werkzeug.security import generate_password_hash random_password = generate_otp_code(8) new_user.password_hash = generate_password_hash(random_password) # اگر فیلد is_new_user در مدل وجود دارد if hasattr(User, 'is_new_user'): new_user.is_new_user = True db.session.add(new_user) db.session.commit() print("=" * 60) print(f"✅ کاربر جدید با شماره تلفن {phone_number} ایجاد شد.") print(f"👤 نام کاربری: {username}") print(f"🔑 رمز عبور موقت: {random_password}") print("=" * 60) return new_user except Exception as e: db.session.rollback() print(f"❌ خطا در ایجاد کاربر: {str(e)}") return None def send_otp_sms(phone_number, otp_code): """ ارسال کد OTP از طریق SMS در محیط توسعه، کد در کنسول نمایش داده می‌شود """ # نرمال‌سازی شماره تلفن phone_number = normalize_phone_number(phone_number) print("=" * 60) print(f"📱 ارسال کد تایید به شماره: {phone_number}") print(f"🔑 کد تایید: {otp_code}") print("=" * 60) # ============================================ # کد واقعی ارسال SMS با Kavenegar # ============================================ # try: # from kavenegar import KavenegarAPI # api = KavenegarAPI('YOUR_API_KEY') # params = { # 'sender': '1000596446', # 'receptor': phone_number, # 'message': f'کد تایید شما: {otp_code}' # } # response = api.sms_send(params) # return True # except Exception as e: # print(f"Error sending SMS: {e}") # return False return True #=== @auth_bp.route('/admin/login', methods=['GET', 'POST']) def admin_login(): """ورود ادمین با یوزرنیم و پسورد""" if current_user.is_authenticated: return redirect_to_dashboard(current_user) form = AdminLoginForm() if form.validate_on_submit(): user = User.query.filter_by(username=form.username.data, role='admin').first() if user and user.check_password(form.password.data): success, role = login_user_handler(user, form.remember_me.data) if success: # ============================================ # اصلاح این بخش # ============================================ next_page = request.args.get('next') if next_page and is_safe_url(next_page): return redirect(next_page) return redirect_to_dashboard(user) else: flash('نام کاربری یا رمز عبور اشتباه است', 'danger') return render_template('auth/admin_login.html', form=form) @auth_bp.route('/login/manager', methods=['GET', 'POST']) def login_manager(): """ورود مدیر پروژه با یوزرنیم و پسورد""" if current_user.is_authenticated: return redirect_to_dashboard(current_user) form = ManagerLoginForm() if form.validate_on_submit(): user = User.query.filter_by(username=form.username.data, role='manager').first() if user and user.check_password(form.password.data): success, role = login_user_handler(user, form.remember_me.data) if success: # ============================================ # اصلاح این بخش # ============================================ next_page = request.args.get('next') if next_page and is_safe_url(next_page): return redirect(next_page) return redirect_to_dashboard(user) else: flash('نام کاربری یا رمز عبور اشتباه است', 'danger') return render_template('auth/login_manager.html', form=form) # ============================================ # ورود کارفرما # ============================================ @auth_bp.route('/login/employer', methods=['GET', 'POST']) def login_employer(): """ورود کارفرما با کد پروژه، شماره همراه و رمز موقت""" if current_user.is_authenticated: return redirect_to_dashboard(current_user) # مرحله 1: دریافت کد پروژه و شماره موبایل if 'employer_login_step' not in session or session.get('employer_login_step') == 0: form = EmployerLoginStep1Form() if form.validate_on_submit(): try: project = get_project_by_code(form.project_code.data) if not project: flash('کد پروژه نامعتبر است', 'danger') return render_template('auth/login_employer_step1.html', form=form) user = get_user_by_phone(form.phone.data) if not user or user.role != 'employer': flash('شماره همراه ثبت نشده است یا نقش کارفرما ندارد.', 'danger') return render_template('auth/login_employer_step1.html', form=form) if project.employer_id != user.id: flash('شما به این پروژه متصل نیستید.', 'danger') return render_template('auth/login_employer_step1.html', form=form) temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() session['employer_login_step'] = 1 session['employer_login_user_id'] = user.id session['employer_login_phone'] = form.phone.data session['employer_login_project_id'] = project.id session.modified = True send_temp_password_via_sms(form.phone.data, temp_password) flash('✅ رمز موقت به شماره شما ارسال شد.', 'success') return redirect(url_for('auth.login_employer')) except SQLAlchemyError as e: handle_db_error(e) return render_template('auth/login_employer_step1.html', form=form) return render_template('auth/login_employer_step1.html', form=form) # مرحله 2: وارد کردن رمز موقت elif session.get('employer_login_step') == 1: form = EmployerLoginStep2Form() if request.args.get('action') == 'back': session.clear() flash('به مرحله قبل بازگشتید.', 'info') return redirect(url_for('auth.login_employer')) if request.args.get('action') == 'resend': user_id = session.get('employer_login_user_id') if user_id: user = User.query.get(user_id) if user: try: temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() send_temp_password_via_sms(user.phone, temp_password) flash('✅ رمز موقت جدید ارسال شد.', 'success') return redirect(url_for('auth.login_employer')) except SQLAlchemyError as e: handle_db_error(e) flash('❌ خطا در ارسال مجدد.', 'danger') return redirect(url_for('auth.login_employer')) if form.validate_on_submit(): user_id = session.get('employer_login_user_id') if not user_id: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employer')) user = User.query.get(user_id) if not user: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employer')) if user.temp_password != form.temp_password.data: flash('❌ رمز موقت نامعتبر است.', 'danger') return render_template('auth/login_employer_step2.html', form=form) if user.temp_password_expires and user.temp_password_expires < datetime.now(): flash('⏰ رمز موقت منقضی شده است.', 'danger') session.clear() return redirect(url_for('auth.login_employer')) success, role = login_user_handler(user, True) if success: session.pop('employer_login_step', None) session.pop('employer_login_user_id', None) session.pop('employer_login_phone', None) session.pop('employer_login_project_id', None) session.modified = True flash(f'✅ خوش آمدید {user.fullname}', 'success') return redirect_to_dashboard(user) else: flash('❌ خطا در ورود به سیستم.', 'danger') return render_template('auth/login_employer_step2.html', form=form) return render_template('auth/login_employer_step2.html', form=form) # ============================================ # ✅ اضافه کردن این بخش برای مواردی که هیچ شرطی مطابقت ندارد # ============================================ else: # اگر مرحله نامعتبر است، session را پاک کن session.clear() flash('لطفاً دوباره از ابتدا شروع کنید.', 'info') return redirect(url_for('auth.login_employer')) # ============================================ # ورود عوامل اجرایی (با GPS + Face - حذف QR) # ============================================ @auth_bp.route('/login/employee', methods=['GET', 'POST']) def login_employee(): """ورود عوامل اجرایی با کد پروژه، شماره همراه، رمز موقت، GPS و چهره""" if current_user.is_authenticated: return redirect_to_dashboard(current_user) # ============================================ # مرحله 1: دریافت کد پروژه و شماره موبایل # ============================================ if session.get('employee_login_step') is None or session.get('employee_login_step') == 0: form = EmployeeLoginStep1Form() if form.validate_on_submit(): try: # بررسی پروژه project = get_project_by_code(form.project_code.data) if not project: flash('کد پروژه نامعتبر است', 'danger') return render_template('auth/login_employee_step1.html', form=form) # ✅ بازگشت Response # بررسی کاربر user = get_user_by_phone(form.phone.data) if not user or user.role != 'employee': flash('شماره همراه ثبت نشده است یا نقش عامل اجرایی ندارد.', 'danger') return render_template('auth/login_employee_step1.html', form=form) # ✅ بازگشت Response # بررسی اتصال به پروژه project_employee = ProjectEmployee.query.filter_by( project_id=project.id, employee_id=user.id ).first() if not project_employee: flash('شما به این پروژه متصل نیستید.', 'danger') return render_template('auth/login_employee_step1.html', form=form) # ✅ بازگشت Response # تولید و ارسال رمز موقت temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() # ذخیره اطلاعات در session session['employee_login_step'] = 1 session['employee_login_user_id'] = user.id session['employee_login_phone'] = form.phone.data session['employee_login_project_id'] = project.id session['employee_login_project_employee_id'] = project_employee.id # ذخیره موقعیت GPS پروژه در session if project.gps_lat and project.gps_lon: session['project_gps_lat'] = project.gps_lat session['project_gps_lon'] = project.gps_lon session['project_gps_radius'] = project.gps_radius or 100 else: session['project_gps_lat'] = None session['project_gps_lon'] = None session['project_gps_radius'] = 100 session.modified = True send_temp_password_via_sms(form.phone.data, temp_password) flash('✅ رمز موقت به شماره شما ارسال شد.', 'success') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response except SQLAlchemyError as e: handle_db_error(e) return render_template('auth/login_employee_step1.html', form=form) # ✅ بازگشت Response return render_template('auth/login_employee_step1.html', form=form) # ✅ بازگشت Response # ============================================ # مرحله 2: رمز موقت + GPS # ============================================ elif session.get('employee_login_step') == 1: form = EmployeeLoginStep2Form() if request.args.get('action') == 'back': session.clear() flash('به مرحله قبل بازگشتید.', 'info') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response if request.args.get('action') == 'resend': user_id = session.get('employee_login_user_id') if user_id: user = User.query.get(user_id) if user: try: temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() send_temp_password_via_sms(user.phone, temp_password) flash('✅ رمز موقت جدید ارسال شد.', 'success') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response except SQLAlchemyError as e: handle_db_error(e) flash('❌ خطا در ارسال مجدد.', 'danger') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response if form.validate_on_submit(): user_id = session.get('employee_login_user_id') if not user_id: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response user = User.query.get(user_id) if not user: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response # بررسی رمز موقت if user.temp_password != form.temp_password.data: flash('❌ رمز موقت نامعتبر است.', 'danger') return render_template('auth/login_employee_step2.html', form=form) # ✅ بازگشت Response if user.temp_password_expires and user.temp_password_expires < datetime.now(): flash('⏰ رمز موقت منقضی شده است.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response # ============================================ # دریافت GPS از مرورگر # ============================================ lat = request.form.get('lat', type=float) lon = request.form.get('lon', type=float) gps_accuracy = request.form.get('accuracy', type=float) if not lat or not lon: flash('📍 لطفاً موقعیت GPS خود را ارسال کنید.', 'danger') return render_template('auth/login_employee_step2.html', form=form) # ✅ بازگشت Response # دریافت موقعیت GPS پروژه project_lat = session.get('project_gps_lat') project_lon = session.get('project_gps_lon') project_radius = session.get('project_gps_radius', 100) if not project_lat or not project_lon: flash('⚠️ موقعیت GPS برای این پروژه تعریف نشده است. لطفاً با مدیر پروژه تماس بگیرید.', 'warning') return render_template('auth/login_employee_step2.html', form=form) # ✅ بازگشت Response # مقایسه GPS کاربر با موقعیت پروژه distance_to_project = calculate_gps_distance( project_lat, project_lon, lat, lon ) is_inside_project = distance_to_project <= project_radius logger.info( f"📍 مقایسه GPS کاربر {user.id} با پروژه {session.get('employee_login_project_id')}: " f"موقعیت پروژه=({project_lat:.6f}, {project_lon:.6f}), " f"موقعیت کاربر=({lat:.6f}, {lon:.6f}), " f"فاصله={distance_to_project:.1f}متر, " f"شعاع مجاز={project_radius}متر, " f"در محدوده={is_inside_project}" ) if not is_inside_project: flash( f'📍 موقعیت GPS شما در محدوده پروژه نیست. ' f'فاصله تا پروژه: {distance_to_project:.1f} متر ' f'(حداکثر مجاز: {project_radius} متر)', 'danger' ) return render_template('auth/login_employee_step2.html', form=form) # ✅ بازگشت Response # ذخیره موقعیت فعلی کاربر session['employee_login_lat'] = lat session['employee_login_lon'] = lon session['employee_login_gps_accuracy'] = gps_accuracy session['employee_login_distance_to_project'] = distance_to_project session['employee_login_step'] = 2 session.modified = True flash( f'✅ موقعیت GPS تایید شد. ' f'فاصله تا پروژه: {distance_to_project:.1f} متر', 'success' ) return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response return render_template('auth/login_employee_step2.html', form=form) # ✅ بازگشت Response # ============================================ # مرحله 3: تشخیص چهره و ورود نهایی # ============================================ elif session.get('employee_login_step') == 2: form = EmployeeLoginStep3Form() if request.args.get('action') == 'back': session['employee_login_step'] = 1 session.modified = True flash('به مرحله قبل بازگشتید.', 'info') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response if form.validate_on_submit(): user_id = session.get('employee_login_user_id') if not user_id: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response user = User.query.get(user_id) if not user: flash('کاربر یافت نشد.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response # بررسی مجدد رمز موقت if user.temp_password and user.temp_password_expires: if user.temp_password_expires < datetime.now(): flash('⏰ رمز موقت منقضی شده است. لطفاً دوباره تلاش کنید.', 'danger') session.clear() return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response face_data = request.form.get('face_data') if not face_data: flash('📸 لطفاً تصویر چهره خود را ثبت کنید.', 'danger') return render_template('auth/login_employee_step3.html', form=form) # ✅ بازگشت Response # تایید چهره if not verify_face(face_data, user.id): flash('❌ تشخیص چهره ناموفق بود. لطفاً دوباره تلاش کنید.', 'danger') return render_template('auth/login_employee_step3.html', form=form) # ✅ بازگشت Response # ============================================ # ورود نهایی - ذخیره در دیتابیس # ============================================ try: # به‌روزرسانی موقعیت GPS کاربر user.gps_lat = session.get('employee_login_lat') user.gps_lon = session.get('employee_login_lon') user.gps_accuracy = session.get('employee_login_gps_accuracy') user.gps_updated_at = datetime.utcnow() user.last_login = datetime.utcnow() # به‌روزرسانی ProjectEmployee project_employee_id = session.get('employee_login_project_employee_id') if project_employee_id: project_employee = ProjectEmployee.query.get(project_employee_id) if project_employee: project_employee.gps_verified = True project_employee.face_verified = True project_employee.last_login = datetime.utcnow() project_employee.last_gps_lat = session.get('employee_login_lat') project_employee.last_gps_lon = session.get('employee_login_lon') project_employee.last_gps_accuracy = session.get('employee_login_gps_accuracy') project_employee.last_gps_distance = session.get('employee_login_distance_to_project') db.session.commit() except SQLAlchemyError as e: handle_db_error(e) return render_template('auth/login_employee_step3.html', form=form) # ✅ بازگشت Response # ورود کاربر success, role = login_user_handler(user, True) if success: # پاک کردن session موقت session.pop('employee_login_step', None) session.pop('employee_login_user_id', None) session.pop('employee_login_phone', None) session.pop('employee_login_project_id', None) session.pop('employee_login_project_employee_id', None) session.pop('employee_login_lat', None) session.pop('employee_login_lon', None) session.pop('employee_login_gps_accuracy', None) session.pop('employee_login_distance_to_project', None) session.pop('project_gps_lat', None) session.pop('project_gps_lon', None) session.pop('project_gps_radius', None) session.modified = True flash(f'✅ خوش آمدید {user.fullname}', 'success') return redirect_to_dashboard(user) # ✅ بازگشت Response else: flash('❌ خطا در ورود به سیستم.', 'danger') return render_template('auth/login_employee_step3.html', form=form) # ✅ بازگشت Response return render_template('auth/login_employee_step3.html', form=form) # ✅ بازگشت Response # ============================================ # اگر هیچ مرحله‌ای مطابقت نداشت # ============================================ # پاک کردن session و بازگشت به مرحله اول session.clear() flash('لطفاً دوباره از ابتدا شروع کنید.', 'info') return redirect(url_for('auth.login_employee')) # ✅ بازگشت Response # ============================================ # مسیرهای ثبت نام مدیر پروژه # ============================================ @auth_bp.route('/register/manager', methods=['GET', 'POST']) def register_manager() -> Response: """ثبت نام مدیر پروژه""" form = ManagerRegisterForm() if form.validate_on_submit(): # بررسی وجود کاربر با شماره موبایل existing_user: Optional[User] = User.query.filter_by(phone=form.phone.data).first() if existing_user: if existing_user.manager_request_status == 'pending': flash('شما قبلاً درخواست ثبت نام داده‌اید و در انتظار تایید هستید.', 'warning') elif existing_user.manager_request_status == 'approved': flash('شما قبلاً تایید شده‌اید. لطفاً وارد شوید.', 'info') else: flash('این شماره تماس قبلاً ثبت شده است.', 'danger') return render_template('auth/register_manager.html', form=form) # بررسی وجود کاربر با نام کاربری existing_username: Optional[User] = User.query.filter_by(username=form.username.data).first() if existing_username: flash('این نام کاربری قبلاً ثبت شده است.', 'danger') return render_template('auth/register_manager.html', form=form) # ایجاد کاربر جدید user = User( username=form.username.data, fullname=form.fullname.data, phone=form.phone.data, email=form.email.data, role='manager', manager_request_status='pending', is_active=False, is_verified=False ) user.set_password(form.password.data) try: db.session.add(user) db.session.commit() flash('✅ درخواست ثبت نام شما با موفقیت ثبت شد. در انتظار تایید مدیر سیستم.', 'success') return redirect(url_for('auth.login_manager')) except Exception as e: db.session.rollback() flash(f'❌ خطا در ثبت نام: {str(e)}', 'danger') return render_template('auth/register_manager.html', form=form) return render_template('auth/register_manager.html', form=form) # ============================================ # API endpoints # ============================================ @auth_bp.route('/api/check-project-code', methods=['POST']) def check_project_code(): """بررسی اعتبار کد پروژه (AJAX)""" try: data = request.get_json() if not data: return jsonify({'valid': False, 'message': 'داده‌های ارسالی نامعتبر است'}) project_code = data.get('project_code') role = data.get('role') phone = data.get('phone') if not project_code: return jsonify({'valid': False, 'message': 'کد پروژه وارد نشده است'}) if len(project_code) != 16: return jsonify({'valid': False, 'message': 'کد پروژه باید ۱۶ رقم باشد'}) project = Project.query.filter_by(project_code=project_code).first() if not project: return jsonify({'valid': False, 'message': 'کد پروژه نامعتبر است'}) if role == 'employer': if project.employer_id: if phone: user = get_user_by_phone(phone) if user and user.id == project.employer_id: return jsonify({ 'valid': True, 'project_name': project.name, 'message': f'پروژه {project.name} معتبر است' }) else: return jsonify({ 'valid': False, 'message': 'این پروژه به کارفرمای دیگری اختصاص دارد' }) else: return jsonify({ 'valid': False, 'message': 'این پروژه به کارفرما اختصاص دارد' }) else: return jsonify({ 'valid': True, 'project_name': project.name, 'message': f'پروژه {project.name} معتبر است' }) elif role == 'employee': if phone: user = get_user_by_phone(phone) if user and user.role == 'employee': project_employee = ProjectEmployee.query.filter_by( project_id=project.id, employee_id=user.id ).first() if project_employee: return jsonify({ 'valid': True, 'project_name': project.name, 'message': f'پروژه {project.name} معتبر است', 'employee_id': user.id, 'project_id': project.id, 'project_employee_id': project_employee.id }) else: return jsonify({ 'valid': False, 'message': 'شما به این پروژه متصل نیستید' }) else: return jsonify({ 'valid': False, 'message': 'شماره همراه ثبت نشده یا نقش نامعتبر است' }) else: return jsonify({ 'valid': True, 'project_name': project.name, 'message': f'پروژه {project.name} معتبر است' }) return jsonify({ 'valid': True, 'project_name': project.name, 'message': f'پروژه {project.name} معتبر است' }) except Exception as e: logger.error(f"خطا در بررسی کد پروژه: {e}") return jsonify({'valid': False, 'message': 'خطا در بررسی کد پروژه'}) @auth_bp.route('/api/check-manager-request-status', methods=['POST']) def check_manager_request_status(): """بررسی وضعیت درخواست مدیر پروژه""" try: data = request.get_json() if not data: return jsonify({'status': 'error', 'message': 'داده‌های ارسالی نامعتبر است'}) phone = data.get('phone') if not phone: return jsonify({'status': 'error', 'message': 'شماره تماس وارد نشده است'}) user = User.query.filter_by(phone=phone, role='manager').first() if not user: return jsonify({'status': 'not_found', 'message': 'درخواستی با این شماره یافت نشد'}) status_text = { 'pending': 'در انتظار تایید', 'approved': 'تایید شده ✅', 'rejected': 'رد شده ❌' } return jsonify({ 'status': user.manager_request_status, 'status_text': status_text.get(user.manager_request_status, 'نامشخص'), 'fullname': user.fullname, 'message': f'درخواست شما {status_text.get(user.manager_request_status, "نامشخص")} است' }) except Exception as e: logger.error(f"خطا در بررسی وضعیت مدیر: {e}") return jsonify({'status': 'error', 'message': 'خطا در بررسی وضعیت'}) @auth_bp.route('/api/resend-temp-password', methods=['POST']) def resend_temp_password(): """ارسال مجدد رمز موقت (API)""" try: data = request.get_json() if not data: return jsonify({'success': False, 'message': 'داده‌های ارسالی نامعتبر است'}) role = data.get('role') if role == 'employer': user_id = session.get('employer_login_user_id') if not user_id: return jsonify({'success': False, 'message': 'اطلاعات کاربر یافت نشد'}) user = User.query.get(user_id) if not user: return jsonify({'success': False, 'message': 'کاربر یافت نشد'}) temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() send_temp_password_via_sms(user.phone, temp_password) return jsonify({'success': True, 'message': 'رمز موقت جدید ارسال شد'}) elif role == 'employee': user_id = session.get('employee_login_user_id') if not user_id: return jsonify({'success': False, 'message': 'اطلاعات کاربر یافت نشد'}) user = User.query.get(user_id) if not user: return jsonify({'success': False, 'message': 'کاربر یافت نشد'}) temp_password = generate_temp_password() user.temp_password = temp_password user.temp_password_expires = datetime.now() + timedelta(minutes=TEMP_PASSWORD_EXPIRY_MINUTES) user.is_temp_password = True db.session.commit() send_temp_password_via_sms(user.phone, temp_password) return jsonify({'success': True, 'message': 'رمز موقت جدید ارسال شد'}) return jsonify({'success': False, 'message': 'نقش کاربر نامعتبر است'}) except SQLAlchemyError as e: db.session.rollback() logger.error(f"خطای دیتابیس در ارسال مجدد رمز: {e}") return jsonify({'success': False, 'message': 'خطا در دیتابیس'}) except Exception as e: logger.error(f"خطا در ارسال مجدد رمز: {e}") return jsonify({'success': False, 'message': 'خطا در ارسال رمز موقت'}) @auth_bp.route('/api/get-project-gps', methods=['POST']) def get_project_gps(): """دریافت محدوده GPS پروژه برای تطبیق با موقعیت کاربر""" try: data = request.get_json() if not data: return jsonify({ 'success': False, 'message': 'داده‌های ارسالی نامعتبر است' }) project_code = data.get('project_code') if not project_code: return jsonify({ 'success': False, 'message': 'کد پروژه وارد نشده است' }) project = Project.query.filter_by(project_code=project_code).first() if not project: return jsonify({ 'success': False, 'message': 'پروژه یافت نشد' }) if not project.gps_bounds: return jsonify({ 'success': True, 'has_bounds': False, 'message': 'محدوده GPS برای این پروژه تعریف نشده است', 'project_name': project.name, 'gps_lat': project.gps_lat, 'gps_lon': project.gps_lon, 'gps_radius': project.gps_radius or 100 }) return jsonify({ 'success': True, 'has_bounds': True, 'gps_bounds': project.gps_bounds, 'project_name': project.name, 'gps_lat': project.gps_lat, 'gps_lon': project.gps_lon, 'gps_radius': project.gps_radius or 100, 'message': f'محدوده GPS پروژه {project.name} دریافت شد' }) except Exception as e: logger.error(f"خطا در دریافت GPS پروژه: {e}") return jsonify({ 'success': False, 'message': 'خطا در دریافت اطلاعات GPS' }) @auth_bp.route('/api/verify-face', methods=['POST']) def verify_face_api(): """API تشخیص چهره""" try: data = request.get_json() if not data: return jsonify({'success': False, 'message': 'داده‌های ارسالی نامعتبر است'}) face_data = data.get('face_data') user_id = data.get('user_id') if not face_data: return jsonify({'success': False, 'message': 'داده چهره دریافت نشد'}) # اعتبارسنجی چهره is_valid = verify_face(face_data, user_id) if is_valid: return jsonify({'success': True, 'message': 'چهره تایید شد'}) else: return jsonify({'success': False, 'message': 'تشخیص چهره ناموفق بود'}) except Exception as e: logger.error(f"خطا در تشخیص چهره: {e}") return jsonify({'success': False, 'message': 'خطا در تشخیص چهره'}) # ============================================ # مسیرهای عمومی # ============================================ @auth_bp.route('/dashboard') @login_required def dashboard(): """داشبورد بر اساس نقش کاربر""" if not current_user.is_authenticated: flash('لطفاً وارد شوید.', 'warning') return redirect(url_for('auth.login')) return redirect_to_dashboard(current_user) @auth_bp.route('/logout') @login_required def logout(): """خروج از سیستم""" try: clear_temp_password(current_user) db.session.commit() except Exception as e: logger.error(f"خطا در خروج از سیستم: {e}") logout_user() session.clear() flash('شما با موفقیت از سیستم خارج شدید.', 'info') return redirect(url_for('public.index')) @auth_bp.route('/change-password', methods=['GET', 'POST']) @login_required def change_password(): """تغییر رمز عبور""" form = ChangePasswordForm() if form.validate_on_submit(): if not current_user.check_password(form.current_password.data): flash('رمز عبور فعلی اشتباه است', 'danger') return render_template('auth/change_password.html', form=form) try: current_user.set_password(form.new_password.data) clear_temp_password(current_user) db.session.commit() logger.info(f"تغییر رمز عبور توسط کاربر {current_user.username}") flash('✅ رمز عبور با موفقیت تغییر کرد.', 'success') return redirect_to_dashboard(current_user) except SQLAlchemyError as e: db.session.rollback() logger.error(f"خطا در تغییر رمز عبور: {e}") flash(f'❌ خطا در تغییر رمز عبور: لطفاً مجدداً تلاش کنید.', 'danger') return render_template('auth/change_password.html', form=form)