import logging from datetime import datetime, timedelta, timezone import jwt import orjson from jwt.exceptions import InvalidTokenError from common.core import security from common.core.config import settings logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) def generate_password_reset_token(email: str) -> str: delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS) now = datetime.now(timezone.utc) expires = now + delta exp = expires.timestamp() encoded_jwt = jwt.encode( {"exp": exp, "nbf": now, "sub": email}, settings.SECRET_KEY, algorithm=security.ALGORITHM, ) return encoded_jwt def verify_password_reset_token(token: str) -> str | None: try: decoded_token = jwt.decode( token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] ) return str(decoded_token["sub"]) except InvalidTokenError: return None def deepcopy_ignore_extra(src, dest): import copy for attr in vars(src): if hasattr(dest, attr): src_value = getattr(src, attr) dest_value = copy.deepcopy(src_value) # deep copy setattr(dest, attr, dest_value) return dest def extract_nested_json(text): stack = [] start_index = -1 results = [] for i, char in enumerate(text): if char in '{[': if not stack: # 记录起始位置 start_index = i stack.append(char) elif char in '}]': if stack and ((char == '}' and stack[-1] == '{') or (char == ']' and stack[-1] == '[')): stack.pop() if not stack: # 栈空时截取完整JSON json_str = text[start_index:i + 1] try: orjson.loads(json_str) # 验证有效性 results.append(json_str) except: pass else: stack = [] # 括号不匹配则重置 if len(results) > 0 and results[0]: return results[0] return None