小登口算（进阶篇）

dynamodb = boto3.resource('dynamodb')

AUTH_TABLE = 'Oral-Arithmetic-Auth'
SESSION_TABLE = 'Oral-Arithmetic-Session'
USER_TABLE = 'Oral-Arithmetic-USER'
QUIZ_TABLE = 'Oral-Arithmetic-Quiz'

event_type = event["queryStringParameters"]["type"]

try:
    event_type = event["queryStringParameters"]["type"]
except KeyError:
    return {
        "statusCode": 400,
        "body": json.dumps({"message": "Bad Request: Missing parameter"}),
    }

body = event["body"]

body = (
    (
        json.loads(base64.b64decode(event["body"]).decode("utf-8"))
        if event.get("isBase64Encoded")
        else event["body"]
    )
    if "body" in event
    else None
)

email = body.get("email", None) if body else None
nickname = body.get("nickname", None) if body else None
password = body.get("password", None) if body else None

def register(email: str, nickname: str, password: str) -> None:
    """
    注册

    :param email: 邮箱
    :param nickname: 昵称
    :param password: 密码
    :raise ValueError: 参数为空或邮箱已存在
    """
    # 检查参数是否为空
    if not email or not nickname or not password:
        raise ValueError("Missing parameter")

    # 定义数据表
    auth_table = dynamodb.Table(AUTH_TABLE)
    user_table = dynamodb.Table(USER_TABLE)

    # 检查邮箱是否存在
    if auth_table.get_item(Key={"email": email}).get("Item"):
        raise ValueError("Email already exists")

    # 对密码进行加密
    hashed_password = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())

    # 生成 UID
    while True:
        uid = int(str(uuid.uuid4().int)[:8])
        print(uid)
        data = user_table.get_item(Key={"uid": uid})
        if "Item" in data:
            continue
        else:
            break

    # 存入 DynamoDB
    auth_item = {
        "email": email,
        "password": hashed_password.decode("utf-8"),
        "uid": uid,
    }
    auth_table.put_item(Item=auth_item)
    user_item = {"uid": uid, "email": email, "nickname": nickname}
    user_table.put_item(Item=user_item)

def login(email: str, password: str) -> [str, int]:
    """
    登录

    :param email: 邮箱
    :param password: 密码
    :return: 用于 Cookie 的 session 值和有效期
    :raise ValueError: 参数为空或邮箱密码错误。
    """
    # 检查参数是否为空
    if not email or not password:
        raise ValueError("Missing parameter")

    # 定义数据表
    auth_table = dynamodb.Table(AUTH_TABLE)
    session_table = dynamodb.Table(SESSION_TABLE)

    # 获取用户数据
    data = auth_table.get_item(Key={"email": email}).get("Item")

    # 验证用户名和密码
    if not data or not bcrypt.checkpw(
        password.encode("utf-8"), data["password"].encode("utf-8")
    ):
        raise ValueError("Invalid Email or Password")

    uid = data.get("uid")

    # 通过 UID 与时间戳和随机数生成初始 session
    timestamp = int(time.time())
    random_number = str(random.randint(1000, 9999))
    session_raw = f"{uid}{str(timestamp)}{random_number}".encode("utf-8")

    # 生成 session
    session = hashlib.sha256(session_raw).hexdigest()
    expiration = 604800  # 有效期一周

    # 将 token 存储在 DynamoDB
    session_table.put_item(
        Item={"session": session, "uid": uid, "expiration": timestamp + expiration}
    )

    return session, expiration

def lambda_handler(event, context):
    # 获取事件类型
    try:
        event_type = event["queryStringParameters"]["type"]
    except KeyError:
        return {
            "statusCode": 400,
            "body": json.dumps({"message": "Missing parameter"}),
        }

    # 解析请求体
    body = (
        (
            json.loads(base64.b64decode(event["body"]).decode("utf-8"))
            if event.get("isBase64Encoded")
            else event["body"]
        )
        if "body" in event
        else None
    )
    email = body.get("email", None) if body else None
    nickname = body.get("nickname", None) if body else None
    password = body.get("password", None) if body else None

    # 注册
    if event_type == "register":
        try:
            register(email, nickname, password)
            return {"statusCode": 201, "body": "Success"}
        except ValueError as e:
            return {"statusCode": 400, "body": json.dumps({"message": str(e)})}

    # 登录
    if event_type == "login":
        try:
            session, expiration = login(email, password)
            return {
                "statusCode": 201,
                "headers": {
                    "Set-Cookie": f"session={session}; Path=/; Max-Age={expiration}; Secure"
                },
                "body": "Cookie Set",
            }
        except ValueError as e:
            return {"statusCode": 400, "body": json.dumps({"message": str(e)})}

    return {"statusCode": 400, "body": ({"message": "Invalid parameter"})}

def lambda_handler(event, context):
    # 获取 HTTP 请求方法
    http_method = event["requestContext"]["http"]["method"]

    # 处理 OPTIONS 请求
    if http_method == "OPTIONS":
        return {
            "statusCode": 200,
            "headers": {
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
                "Access-Control-Allow-Headers": "*",
                "Access-Control-Allow-Credentials": True
            },
            "body": "",
        }

user_item = {
    "uid": uid,
    "email": email,
    "nickname": nickname,
    "avatar": "",
    "total": 0,
    "competition_total": 0,
    "competition_win": 0,
    "qid": [],
    "mistake": [],
}
user_table.put_item(Item=user_item)

"body": json.dumps(
    {
        "message": "Cookie Set",
        "session": session,
        "expiration": expiration,
        "nickname": nickname,
    }
),

def get_uid_from_cookie(cookie: dict) -> int:
    """
    通过 Cookie 获取 UID

    :param cookie: Cookie
    :return: UID
    """
    # 检查参数是否为空
    if not cookie:
        raise ValueError("Missing parameter")

    # 解析 Cookie
    cookie_dict = {i.split("=")[0].strip(): i.split("=")[1].strip() for i in cookie}
    session = cookie_dict.get("session")

    # 定义数据表
    session_table = dynamodb.Table(SESSION_TABLE)

    # 获取 UID
    data = session_table.get_item(Key={"session": session})
    if "Item" in data:
        expiration = data["Item"].get("expiration")
        if expiration < int(time.time()):
            raise ValueError("Session expired")
        return data["Item"].get("uid")
    else:
        raise ValueError("Missing parameter")

def get(uid: int) -> dict:
    """
    获取用户数据

    :param uid: UID
    """
    # 检查参数是否为空
    if not uid:
        raise ValueError("Missing parameter")

    # 定义数据表
    user_table = dynamodb.Table(USER_TABLE)

    # 读取 DynamoDB
    data = user_table.get_item(Key={"uid": uid})

    if "Item" in data:
        userdata = data["Item"]
        userdata = json.loads(json.dumps(userdata, default=str))
        return userdata
    else:
        raise ValueError("Missing parameter")

while True:
    qid = str(uuid.uuid4())
    data = quiz_table.get_item(Key={"qid": qid})
    if "Item" in data:
        continue
    else:
        break

# 存入 DynamoDB
quiz_item = {
    "qid": qid,
    "mode": mode,
    "quiz_time": quiz_time,
    "questions": questions,
    "question_count": question_count,
    "correct_count": correct_count,
    "used_time": used_time,
    "is_competition": is_competition,
    "allow_competition": allow_competition,
    "p1_uid": uid,
    "p2_uid": [],
}
quiz_table.put_item(Item=quiz_item)

# 更新用户数据，将 qid 添加到 qid 列表中，并使总场数 +1
user_table.update_item(
    Key={"uid": uid},
    UpdateExpression="SET qid = list_append(if_not_exists(qid, :empty_list), :qid), #total = #total + :increment",
    ExpressionAttributeNames={"#total": "total"},
    ExpressionAttributeValues={":qid": [qid], ":empty_list": [], ":increment": 1},
)

def save_mistake(
    uid: int, question: str, user_answer: int, correct_answer: int
) -> None:
    """
    保存错题

    :param uid: 用户 ID
    :param question: 题目
    :param user_answer: 用户答案
    :param correct_answer: 正确答案
    """
    # 检查参数是否为空
    if uid is None or not question or user_answer is None or correct_answer is None:
        raise ValueError("Missing parameter")

    # 定义数据表
    user_table = dynamodb.Table(USER_TABLE)

    # 错题记录
    mistake = {
        "question": question,
        "userAnswer": user_answer,
        "correctAnswer": correct_answer,
    }

    # 更新用户数据，将错题添加到 mistake 列表中
    user_table.update_item(
        Key={"uid": uid},
        UpdateExpression="SET mistake = list_append(if_not_exists(mistake, :empty_list), :mistake)",
        ExpressionAttributeValues={":mistake": [mistake], ":empty_list": []},
    )


def remove_mistake(uid: int, question: str) -> None:
    """
    移除错题

    :param uid: 用户 ID
    :param question: 题目
    """
    # 检查参数是否为空
    if uid is None or not question:
        raise ValueError("Missing parameter")

    # 定义数据表
    user_table = dynamodb.Table(USER_TABLE)

    # 获取用户数据
    response = user_table.get_item(Key={"uid": uid})
    if "Item" in response:
        mistakes = response["Item"].get("mistake", [])
        # 移除指定错题
        mistakes = [mistake for mistake in mistakes if mistake["question"] != question]
        # 更新用户数据
        user_table.update_item(
            Key={"uid": uid},
            UpdateExpression="SET mistake = :mistakes",
            ExpressionAttributeValues={":mistakes": mistakes},
        )
    else:
        raise ValueError("错题不存在")


def get_mistakes(uid: int) -> list:
    """
    获取错题

    :param uid: 用户 ID
    :return: 错题列表
    """
    # 检查参数是否为空
    if uid is None:
        raise ValueError("Missing parameter")

    # 定义数据表
    user_table = dynamodb.Table(USER_TABLE)

    # 获取用户数据
    response = user_table.get_item(Key={"uid": uid})
    if "Item" in response:
        return response["Item"].get("mistake", [])
    else:
        raise ValueError("Missing parameter")