from fastapi import Depends, APIRouter, Query, Path from internal.models import * from internal.database import ( fetch_one, fetch_all, execute_query, response_success, raise_if_exists, raise_if_not_found, ) from dependencies import get_current_active_user import json router = APIRouter(prefix="/expenses", tags=["收支管理"]) # 获取列表 @router.get("/list") async def expense_list(page: int = Query(None), page_size: int = Query(None)): limit_clause = "" if page is not None and page_size is not None: offset = (page - 1) * page_size limit_clause = f"LIMIT {page_size} OFFSET {offset}" select_query = f""" SELECT expenses.id,expenses.pay_price, expenses.live_price, expenses.create_at, expensetypes.typename,expenses.descr FROM expenses LEFT JOIN `expensetypes` ON expenses.typeid = expensetypes.id ORDER BY create_at DESC {limit_clause}; """ expense_list = fetch_all(select_query) return response_success(expense_list, "expense get list success") @router.get("/list/{id}") async def expense_one(id: int): # 列表参数:收支名称、收支内容、创建时间、收支图片、收支查看时间、收支阅读次数、收支字数、类型名称、标签名列表 select_query = """ SELECT expenses.id,expenses.pay_price, expenses.live_price, expenses.create_at, expensetypes.typename,expenses.descr FROM expenses LEFT JOIN `expensetypes` ON expenses.typeid = expensetypes.id WHERE id = %s ORDER BY create_at DESC; """ expense_one = fetch_one(select_query, (id,)) return response_success(expense_one, "expense get expense_one success") # 收支新增 @router.post("/add") async def expense_add(expense: expense, _: User = Depends(get_current_active_user)): select_query = "SELECT * FROM expenses WHERE expensetitle = %s" existing_expense = fetch_one(select_query, (expense.expensetitle,)) raise_if_exists(existing_expense, "expense already exists") insert_query = ( "INSERT INTO expenses (expensetitle, expensecontent,imglink, typeid, descr) VALUES (%s, %s, %s, %s,%s)" ) insert_value=(expense.expensetitle,expense.expensecontent,expense.imglink,expense.typeid,expense.descr) execute_query(insert_query,insert_value) return {"message": "expense created successfully"} # 收支删除 @router.delete("/delete/{id}") async def expense_delete(id: str = Path(description="收支id")): select_query = "SELECT * FROM expenses WHERE id = %s" existing_expense = fetch_one(select_query, (id,)) raise_if_not_found(existing_expense, "expense not found") delete_query = "DELETE FROM expenses WHERE id = %s" execute_query(delete_query, (id,)) return response_success(message="expense delete success") @router.put("/update/{id}") async def expense_update(id: int, expense: expense, _: User = Depends(get_current_active_user)): # 检查要编辑的收支是否存在 select_query = "SELECT * FROM expenses WHERE id = %s" existing_expense = fetch_one(select_query, (id,)) raise_if_not_found(existing_expense, "expense not found") # 更新收支信息 update_query = ( "UPDATE expenses SET expensetitle = %s, expensecontent = %s, imglink = %s, typeid = %s, descr = %s WHERE id = %s" ) update_data = (expense.expensetitle, expense.expensecontent, expense.imglink, expense.typeid, expense.descr, id) execute_query(update_query, update_data) return response_success("expense update sucess") # 收支模糊查询 @router.get("/list/search") async def expense_list_search( expensetitle: str = Query(None, description="收支标题"), typename: str = Query(None, description="收支类型"), start_date: str = Query(None, description="开始时间"), end_date: str = Query(None, description="结束时间"), ): select_query = """ SELECT expenses.id, expensetitle, expensecontent,wordcount, typename, create_at, update_at, expenses.descr FROM expenses LEFT JOIN `expensetypes` ON expenses.typeid = expensetypes.id WHERE 1=1 """ params = [] if expensetitle: select_query += " AND expensetitle LIKE %s" params.append(f"%{expensetitle}%") if typename: select_query += " AND typename LIKE %s" params.append(f"%{typename}%") if start_date: select_query += " AND create_at >= %s" params.append(start_date) if end_date: select_query += " AND create_at <= %s" params.append(end_date) select_query += "ORDER BY create_at DESC" expense_list = fetch_all(select_query, params=params, fetchall=True) return response_success(data=expense_list, message="expense serach succuessfully!") # 根据id查询收支 @router.get("/list/search/{id}") async def get_id_expense(id: str = Path(description="收支id")): select_query = """SELECT expenses.id, expensetitle, expensecontent,wordcount, expenses.typeid, expenses.descr,imglink FROM expenses LEFT JOIN `expensetypes` ON expenses.typeid = expensetypes.id WHERE expenses.id = %s """ expense_list = execute_query(select_query, (id,)) return response_success(data=expense_list, message="expense search success") # 我就测试一下