import re from django.core.exceptions import ValidationError from pingcap_sso.validators.username import validate_username from pingcap_sso.models import User from pingcap_sso.github.exceptions import AuthUserUnregisteredError def check_if_user_exists(strategy, details, backend, user=None, *args, **kwargs): if user: return {'is_new': False} raise AuthUserUnregisteredError(backend) USER_FIELDS = ['username', 'email'] USERNAME_SLUGIFY_PATTERN = re.compile(r'[a-zA-Z\d\u4e00-\u9fa5]+') def get_username(strategy, details, backend, user=None, *args, **kwargs): if 'username' not in backend.setting('USER_FIELDS', USER_FIELDS): return storage = strategy.storage if not user: final_username = details.get('username', User.generate_random_username()) try: validate_username(final_username) except ValidationError: final_username = '_'.join(USERNAME_SLUGIFY_PATTERN.findall(final_username)) try: validate_username(final_username) except ValidationError: final_username = User.generate_random_username() if storage.user.user_exists(username=final_username): final_username = User.generate_random_username() else: final_username = storage.user.get_username(user) return {'username': final_username} def track_signup_channel(strategy, details, backend, user=None, is_new=False, *args, **kwargs): """新用户记录注册来源""" if not user or not is_new: return from django.urls import reverse from pingcap_sso.lib.sso import parse_signup_channel_from_url original_redirect = strategy.session_get('next', '') or reverse('index') user.signup_source_url = original_redirect[:500] user.signup_channel = parse_signup_channel_from_url(original_redirect) user.save(update_fields=['signup_channel', 'signup_source_url'])