main.py

from fastapi import FastAPI, Header, Query, Path, Body
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
import uvicorn
import json
from pathlib import Path
from zt import (
    PasswordTreatment as PT, 
    FilePath as FP,
    APIFunction as AF
)
from model import LoginModel, ResponseModel, AddAccountModel
import uuid

app = FastAPI()

# init router
@app.get("")
async def root():
    return RedirectResponse(url="/")

# Back-end
## login
@app.post("/api/login")
async def login(req: LoginModel):
    # load password hash from file
    res = ResponseModel()
    login_info_path = "./_db/login_info.json"
    login_info_abs_path = FP.get_abs_path(login_info_path)
    with open(login_info_abs_path, "r", encoding="utf-8") as f:
        password_hash = json.load(f).get("password", "")
    is_valid = PT.verify_password(req.password, password_hash)
    if is_valid:
        session_token = str(uuid.uuid4())
        res.data = dict(sessionToken = session_token)
        session_info_path = "./_db/session_token"
        session_info_abs_path = FP.get_abs_path(session_info_path)
        with open(session_info_abs_path, "w", encoding="utf-8") as f:
            f.write(session_token)
        return JSONResponse(
            status_code = 200,
            content = res.model_dump()
        )
    else:
        res.success = False
        res.message = "invalid_password"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )

## Check token
@app.get("/api/check-token")
async def check_token(x_session_token: str | None = Header(default=None)):
    res = ResponseModel()
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if is_valid:
        return JSONResponse(
            status_code = 200,
            content = res.model_dump()
        )
    else:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )

## Get account list
@app.get("/api/get/account-list")
async def get_account_list(
    x_session_token: str | None = Header(default=None),
    atype: str | None = Query(default=None),
    name: str | None = Query(default=None)
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # load account index
    account_index_path = "./_db/account_index.json"
    account_index_abs_path = FP.get_abs_path(account_index_path)
    with open(account_index_abs_path, "r", encoding="utf-8") as f:
        account_index: dict = json.load(f)
    
    # filter the account index by atype and name
    filtered_list: list = []
    if (atype is None or atype == "") and (name is None or name == ""):
        # 顯示全部帳戶
        for data_list in account_index.values():
            for data in data_list:
                filtered_list.append(data)
    elif (atype is not None and atype != "") and (name is None or name == ""):
        # 只按類型過濾
        data_list = account_index.get(atype, [])
        for data in data_list:
            filtered_list.append(data)
    elif (atype is None or atype == "") and (name is not None and name != ""):
        # 只按名稱過濾
        for data_list in account_index.values():
            for data in data_list:
                if name.lower() in data.get("name", "").lower():
                    filtered_list.append(data)
    else:
        # 按類型和名稱過濾
        data_list = account_index.get(atype, [])
        for data in data_list:
            if name.lower() in data.get("name", "").lower():
                filtered_list.append(data)
    
    # return the filtered list
    res.data = filtered_list
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

## Get account detail
@app.get("/api/get/account-detail/{gid}")
async def get_account_detail(
    gid: str = Path(),
    x_session_token: str | None = Header(default=None)
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # load account index
    account_index_path: str = "./_db/account_row.json"
    account_index_abs_path: str = FP.get_abs_path(account_index_path)
    with open(account_index_abs_path, "r", encoding="utf-8") as f:
        account_row: dict = json.load(f)
    
    # check if gid exists
    if gid not in account_row:
        res.success = False
        res.message = "gid_not_found"
        return JSONResponse(
            status_code = 404,
            content = res.model_dump()
        )
    
    # return the account detail
    res.data = account_row[gid]
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

## Get account types
@app.get("/api/get/account-types")
async def get_account_types(
    x_session_token: str | None = Header(default=None)
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # load account types
    account_types_path: str = "./_db/account_index.json"
    account_types_abs_path: str = FP.get_abs_path(account_types_path)
    with open(account_types_abs_path, "r", encoding="utf-8") as f:
        account_types: list = list(json.load(f))
    
    # return the account types
    res.data = account_types
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

## Create account
@app.post("/api/create/account")
async def create_account(
    x_session_token: str | None = Header(default=None),
    req: AddAccountModel = Body()
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # validate required fields
    if req.type.strip() == "":
        res.success = False
        res.message = "type_is_required"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    
    if req.type.strip() not in ["技術平台", "社交平台", "金融相關", "個人資訊", "娛樂相關", "消費平台"]:
        res.success = False
        res.message = "invalid_type"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )

    if req.name.strip() == "":
        res.success = False
        res.message = "name_is_required"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    
    # load account index and row
    account_index_path: str = "./_db/account_index.json"
    account_index_abs_path: str = FP.get_abs_path(account_index_path)
    with open(account_index_abs_path, "r", encoding="utf-8") as f:
        account_index: dict = json.load(f)
    
    account_row_path: str = "./_db/account_row.json"
    account_row_abs_path: str = FP.get_abs_path(account_row_path)
    with open(account_row_abs_path, "r", encoding="utf-8") as f:
        account_row: dict = json.load(f)
    
    # generate gid
    gid: str = AF.get_random_gid()
    while gid in account_row:
        gid = AF.get_random_gid()
    
    # create new account data
    new_account_data: dict = {
        "gid": gid,
        "type": req.type,
        "name": req.name,
        "loginType": req.loginType or "",
        "account": req.account or "",
        "userName": req.userName or "",
        "password": req.password or "",
        "tradePassword": req.tradePassword or "",
        "cardPassword": req.cardPassword or "",
        "verifyPassword": req.verifyPassword or "",
        "note": req.note or ""
    }

    # add to account row
    account_row[gid] = new_account_data
    with open(account_row_abs_path, "w", encoding="utf-8") as f:
        json.dump(account_row, f, ensure_ascii=False, indent=4)

    # add to account index
    if req.type not in account_index:
        account_index[req.type] = []
    account_index[req.type].append({
        "gid": gid,
        "type": req.type,
        "name": req.name
    })
    with open(account_index_abs_path, "w", encoding="utf-8") as f:
        json.dump(account_index, f, ensure_ascii=False, indent=4)

    # return success with gid
    res.data = dict(gid=gid)
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

## Update account
@app.put("/api/update/account-detail/{gid}")
async def update_account(
    gid: str = Path(),
    x_session_token: str | None = Header(default=None),
    req: AddAccountModel = Body()
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # validate required fields
    if req.type.strip() == "":
        res.success = False
        res.message = "type_is_required"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    
    if req.type.strip() not in ["技術平台", "社交平台", "金融相關", "個人資訊", "娛樂相關", "消費平台"]:
        res.success = False
        res.message = "invalid_type"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )

    if req.name.strip() == "":
        res.success = False
        res.message = "name_is_required"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    
    # load account index and row
    account_index_path: str = "./_db/account_index.json"
    account_index_abs_path: str = FP.get_abs_path(account_index_path)
    with open(account_index_abs_path, "r", encoding="utf-8") as f:
        account_index: dict = json.load(f)
    
    account_row_path: str = "./_db/account_row.json"
    account_row_abs_path: str = FP.get_abs_path(account_row_path)
    with open(account_row_abs_path, "r", encoding="utf-8") as f:
        account_row: dict = json.load(f)
    
    # check if gid exists
    if gid not in account_row:
        res.success = False
        res.message = "gid_not_found"
        return JSONResponse(
            status_code = 404,
            content = res.model_dump()
        )
    
    # get old type and name
    old_type: str = account_row[gid].get("type", "")
    old_name: str = account_row[gid].get("name", "")

    # update account data
    account_row[gid]["type"] = req.type
    account_row[gid]["name"] = req.name
    account_row[gid]["loginType"] = req.loginType or ""
    account_row[gid]["account"] = req.account or ""
    account_row[gid]["userName"] = req.userName or ""
    account_row[gid]["password"] = req.password or ""
    account_row[gid]["tradePassword"] = req.tradePassword or ""
    account_row[gid]["cardPassword"] = req.cardPassword or ""
    account_row[gid]["verifyPassword"] = req.verifyPassword or ""
    account_row[gid]["note"] = req.note or ""
    with open(account_row_abs_path, "w", encoding="utf-8") as f:
        json.dump(account_row, f, ensure_ascii=False, indent=4)
    
    # update account index if type or name changed
    if old_type != req.type or old_name != req.name:
        # remove from old type list
        if old_type in account_index:
            account_index[old_type] = [item for item in account_index[old_type] if item.get("gid") != gid]
            if len(account_index[old_type]) == 0:
                del account_index[old_type]
        # add to new type list
        if req.type not in account_index:
            account_index[req.type] = []
        account_index[req.type].append({
            "gid": gid,
            "type": req.type,
            "name": req.name
        })
        with open(account_index_abs_path, "w", encoding="utf-8") as f:
            json.dump(account_index, f, ensure_ascii=False, indent=4)

    # return success
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

## Delete account
@app.delete("/api/delete/account/{gid}")
async def delete_account(
    gid: str = Path(),
    x_session_token: str | None = Header(default=None)
):
    # build response model
    res = ResponseModel()

    # check session token
    if x_session_token is None:
        res.success = False
        res.message = "no_session_token_provided"
        return JSONResponse(
            status_code = 400,
            content = res.model_dump()
        )
    is_valid = AF.check_session_token(x_session_token)
    if not is_valid:
        res.success = False
        res.message = "invalid_session_token"
        return JSONResponse(
            status_code = 401,
            content = res.model_dump()
        )
    
    # load account index and row
    account_index_path: str = "./_db/account_index.json"
    account_index_abs_path: str = FP.get_abs_path(account_index_path)
    with open(account_index_abs_path, "r", encoding="utf-8") as f:
        account_index: dict = json.load(f)
    
    account_row_path: str = "./_db/account_row.json"
    account_row_abs_path: str = FP.get_abs_path(account_row_path)
    with open(account_row_abs_path, "r", encoding="utf-8") as f:
        account_row: dict = json.load(f)
    
    # check if gid exists
    if gid not in account_row:
        res.success = False
        res.message = "gid_not_found"
        return JSONResponse(
            status_code = 404,
            content = res.model_dump()
        )
    
    # get old type
    old_type: str = account_row[gid].get("type", "")

    # delete from account row
    del account_row[gid]
    with open(account_row_abs_path, "w", encoding="utf-8") as f:
        json.dump(account_row, f, ensure_ascii=False, indent=4)
    
    # delete from account index
    if old_type in account_index:
        account_index[old_type] = [item for item in account_index[old_type] if item.get("gid") != gid]
        if len(account_index[old_type]) == 0:
            del account_index[old_type]
        with open(account_index_abs_path, "w", encoding="utf-8") as f:
            json.dump(account_index, f, ensure_ascii=False, indent=4)

    # return success
    return JSONResponse(
        status_code = 200,
        content = res.model_dump()
    )

# Front-end
@app.get("{path:path}")
async def router(path):
    router_path = "./_set/router.json"
    router_abs_path = FP.get_abs_path(router_path)
    with open(router_abs_path, "r", encoding="utf-8") as f:
        router_dict = json.load(f)
    if path in router_dict:
        file_path = router_dict[path]
        code = 200
    else:
        file_path = "./web/404.html"
        code = 404
    abs_file_path = FP.get_abs_path(file_path)
    if not Path(abs_file_path).is_file():
        abs_file_path = FP.get_abs_path("./web/404.html")
    with open(abs_file_path, "r", encoding="utf-8") as f:
        content = f.read()
    return HTMLResponse(content=content, status_code=code)

# Run the app
if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload = True)