init medical training project

apps/user/utils/__init__.py

@@ -0,0 +1 @@
+# utils

apps/user/utils/jwt_redis.py

@@ -0,0 +1,31 @@
+import time
+
+from django.core.cache import cache
+
+_BLACKLIST_PREFIX = 'jwt_blacklist:'
+_INVALID_BEFORE_PREFIX = 'jwt_user_invalid_before:'
+_USER_TOKEN_TTL = 7 * 24 * 3600  # 覆盖 refresh 最长生命周期
+
+
+def revoke_token(jti: str, exp_ts: float) -> None:
+    """将单个 token 加入黑名单。U7 退出 / U8 旋转旧 refresh 时调用。"""
+    ttl = max(int(exp_ts - time.time()), 1)
+    cache.set(f'{_BLACKLIST_PREFIX}{jti}', '1', timeout=ttl)
+
+
+def invalidate_user_tokens(user_id: int) -> None:
+    """用户级失效截止：写入当前时间戳+1，此前所有 token 立即失效。U5/U6 改密后调用。
+
+    +1 确保同一秒内签发的旧 token 被拒绝，而改密后新登录签发的 token（iat >= now+1）被放行。
+    """
+    cache.set(f'{_INVALID_BEFORE_PREFIX}{user_id}', int(time.time()) + 1, timeout=_USER_TOKEN_TTL)
+
+
+def is_token_revoked(jti: str) -> bool:
+    return bool(cache.get(f'{_BLACKLIST_PREFIX}{jti}'))
+
+
+def get_user_invalid_before(user_id: int):
+    """返回用户级失效截止时间戳（unix seconds），不存在则返回 None。"""
+    val = cache.get(f'{_INVALID_BEFORE_PREFIX}{user_id}')
+    return int(val) if val is not None else None

apps/user/utils/password.py

@@ -0,0 +1,36 @@
+import re
+from typing import Callable, Optional
+
+
+def validate_password_strength(
+    password: str,
+    phone: Optional[str] = None,
+    real_name: Optional[str] = None,
+    old_password_check: Optional[Callable[[str], bool]] = None,
+) -> list:
+    """
+    校验密码强度，返回错误信息列表。列表为空表示通过。
+
+    old_password_check: 传入 password 返回 True 表示与旧密码相同。
+    """
+    errors = []
+
+    if len(password) < 8 or len(password) > 32:
+        errors.append('密码长度必须在 8-32 位之间')
+
+    if not re.search(r'[a-zA-Z]', password):
+        errors.append('密码必须包含字母')
+
+    if not re.search(r'\d', password):
+        errors.append('密码必须包含数字')
+
+    if phone and password == phone:
+        errors.append('密码不能与手机号相同')
+
+    if real_name and password == real_name:
+        errors.append('密码不能与真实姓名相同')
+
+    if old_password_check is not None and old_password_check(password):
+        errors.append('新密码不能与旧密码相同')
+
+    return errors

apps/user/utils/sms.py

@@ -0,0 +1,33 @@
+import random
+import string
+import logging
+from abc import ABC, abstractmethod
+
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+def generate_sms_code(length=6) -> str:
+    return ''.join(random.choices(string.digits, k=length))
+
+
+# ── 策略接口 ──────────────────────────────────────────────────────────────────
+
+class SmsService(ABC):
+    @abstractmethod
+    def send_code(self, phone: str, scene: str, code: str) -> None:
+        """发送验证码短信。失败时抛出 SmsError。"""
+
+
+class SmsError(Exception):
+    pass
+
+
+def get_sms_service() -> SmsService:
+    provider = getattr(settings, 'SMS_PROVIDER', 'mock')
+    if provider == 'aliyun':
+        from apps.user.utils.sms_aliyun import AliyunSmsService
+        return AliyunSmsService()
+    from apps.user.utils.sms_mock import MockSmsService
+    return MockSmsService()

apps/user/utils/sms_aliyun.py

@@ -0,0 +1,57 @@
+import logging
+
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+_SCENE_TEMPLATE = {
+    'register': 'ALIYUN_SMS_TEMPLATE_REGISTER',
+    'login': 'ALIYUN_SMS_TEMPLATE_LOGIN',
+    'reset': 'ALIYUN_SMS_TEMPLATE_RESET',
+}
+
+
+class AliyunSmsService:
+    """阿里云短信实现。SDK 按需 import，避免未安装时影响 mock 模式启动。"""
+
+    def send_code(self, phone: str, scene: str, code: str) -> None:
+        from apps.user.utils.sms import SmsError
+
+        try:
+            from alibabacloud_dysmsapi20170525.client import Client
+            from alibabacloud_dysmsapi20170525 import models as sms_models
+            from alibabacloud_tea_openapi import models as open_api_models
+        except ImportError as e:
+            logger.error('Aliyun SMS SDK not installed: %s', e)
+            raise SmsError('SMS_PROVIDER_ERROR') from e
+
+        config = open_api_models.Config(
+            access_key_id=settings.ALIYUN_SMS_ACCESS_KEY_ID,
+            access_key_secret=settings.ALIYUN_SMS_ACCESS_KEY_SECRET,
+        )
+        config.endpoint = 'dysmsapi.aliyuncs.com'
+        client = Client(config)
+
+        template_attr = _SCENE_TEMPLATE.get(scene, '')
+        template_code = getattr(settings, template_attr, '')
+
+        req = sms_models.SendSmsRequest(
+            phone_numbers=phone,
+            sign_name=settings.ALIYUN_SMS_SIGN_NAME,
+            template_code=template_code,
+            template_param=f'{{"code":"{code}"}}',
+        )
+
+        try:
+            resp = client.send_sms(req)
+            if resp.body.code != 'OK':
+                logger.error(
+                    'Aliyun SMS biz error: code=%s msg=%s request_id=%s',
+                    resp.body.code, resp.body.message, resp.body.request_id,
+                )
+                raise SmsError('SMS_BIZ_ERROR')
+        except SmsError:
+            raise
+        except Exception as e:
+            logger.error('Aliyun SMS provider error: %s', e)
+            raise SmsError('SMS_PROVIDER_ERROR') from e

apps/user/utils/sms_mock.py

@@ -0,0 +1,11 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class MockSmsService:
+    """开发环境短信实现：打印到控制台 + 写日志。不在响应里回填 code。"""
+
+    def send_code(self, phone: str, scene: str, code: str) -> None:
+        print(f'[SMS-MOCK] phone={phone} scene={scene} code={code}')
+        logger.info('[SMS-MOCK] phone=%s scene=%s code=%s', phone, scene, code)