k1nq
023612ce0a
- Updated timestamps and AST hashes for multiple files in the manifest.json. - Added new test file backend/tests/test_expected_income.py. - Added documentation files: перево́д_между_своими.md and 0002-fixes.md. - Introduced pytest.ini to configure pytest settings, including test paths and warning filters.
494 lines
16 KiB
Python
494 lines
16 KiB
Python
import uuid
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
from typing import Annotated, Literal
|
|
|
|
from fastapi import APIRouter, Query, Response
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from app.budgeting import service
|
|
from app.categorization import service as categorization
|
|
from app.categorization.models import CategoryMcc
|
|
from app.dependencies import Context, Csrf, Db
|
|
|
|
router = APIRouter(prefix="/budget", tags=["budget"])
|
|
|
|
|
|
class CategoryBody(BaseModel):
|
|
name: str = Field(min_length=1, max_length=200)
|
|
kind: Literal["monthly", "savings_envelope"] = "monthly"
|
|
mcc_codes: list[int] = Field(default_factory=list, alias="mccCodes")
|
|
|
|
|
|
class AssignmentItem(BaseModel):
|
|
category_id: uuid.UUID
|
|
planned_kopecks: int = Field(ge=0)
|
|
|
|
|
|
class AssignmentsBody(BaseModel):
|
|
items: list[AssignmentItem]
|
|
|
|
|
|
class FundingBody(BaseModel):
|
|
category_id: uuid.UUID
|
|
amount_kopecks: int = Field(gt=0)
|
|
kind: Literal["fund", "unfund"] = "fund"
|
|
note: str | None = Field(default=None, max_length=500)
|
|
|
|
|
|
class ExpectedIncomeBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
expected_on: date = Field(alias="expectedOn")
|
|
amount_kopecks: int = Field(gt=0, alias="amountKopecks")
|
|
source_name: str | None = Field(default=None, max_length=200, alias="sourceName")
|
|
note: str | None = Field(default=None, max_length=500, alias="note")
|
|
|
|
|
|
class ExpectedIncomePatchBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
expected_on: date | None = Field(default=None, alias="expectedOn")
|
|
amount_kopecks: int | None = Field(default=None, gt=0, alias="amountKopecks")
|
|
source_name: str | None = Field(default=None, max_length=200, alias="sourceName")
|
|
note: str | None = Field(default=None, max_length=500, alias="note")
|
|
|
|
|
|
class ExpectedIncomeMatchBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
operation_id: uuid.UUID = Field(alias="operationId")
|
|
amount_kopecks: int | None = Field(default=None, gt=0, alias="amountKopecks")
|
|
|
|
|
|
def period_for(db: Db, context: Context, year: int, month: int):
|
|
return service.get_or_create_period(db, space_id=context.space.id, year=year, month=month)
|
|
|
|
|
|
class MonthBody(BaseModel):
|
|
month: str
|
|
|
|
|
|
class PlanBody(MonthBody):
|
|
plannedKopecks: int = Field(ge=0)
|
|
|
|
|
|
class CategoryFundingBody(MonthBody):
|
|
amountKopecks: int = Field(gt=0)
|
|
|
|
|
|
def parse_month(value: str) -> tuple[int, int]:
|
|
try:
|
|
year_text, month_text = value.split("-", 1)
|
|
year, month = int(year_text), int(month_text)
|
|
except (ValueError, AttributeError) as exc:
|
|
from app.errors import bad_request
|
|
|
|
raise bad_request("BUDGET_PERIOD_INVALID", "Month must use YYYY-MM") from exc
|
|
service.validate_year_month(year, month)
|
|
return year, month
|
|
|
|
|
|
def expected_income_json(item: dict) -> dict:
|
|
return {
|
|
"id": item["id"],
|
|
"expectedOn": item["expected_on"],
|
|
"amountKopecks": item["amount_kopecks"],
|
|
"sourceName": item["source_name"],
|
|
"note": item["note"],
|
|
"matchedKopecks": item["matched_kopecks"],
|
|
"status": "matched"
|
|
if item["matched_kopecks"] >= item["amount_kopecks"]
|
|
else "partially_matched"
|
|
if item["matched_kopecks"]
|
|
else "expected",
|
|
}
|
|
|
|
|
|
def frontend_summary(state: dict) -> dict:
|
|
monthly = state["monthly_categories"]
|
|
envelopes = state["envelopes"]
|
|
funded = sum(row["funded"] for row in monthly) + sum(row["funded_total"] for row in envelopes)
|
|
spent = sum(row["spent"] for row in monthly)
|
|
available = sum(row["available"] for row in monthly + envelopes)
|
|
return {
|
|
"month": f"{state['period']['year']:04d}-{state['period']['month']:02d}",
|
|
"status": state["period"]["status"],
|
|
"projectedPoolKopecks": state["carry_pool"]
|
|
+ state["expected_income_total"]
|
|
- state["planned_total"],
|
|
"actualPoolKopecks": state["pool"],
|
|
"plannedTotalKopecks": state["planned_total"],
|
|
"fundedTotalKopecks": funded,
|
|
"spentTotalKopecks": spent,
|
|
"availableKopecks": available,
|
|
"warnings": [{"code": code} for code in state["warnings"]],
|
|
"blockers": state["close_blockers"],
|
|
"expectedIncomes": [expected_income_json(item) for item in state["expected_incomes"]],
|
|
}
|
|
|
|
|
|
@router.get("/summary")
|
|
def summary(month: str, db: Db, context: Context) -> dict:
|
|
year, month_number = parse_month(month)
|
|
period = period_for(db, context, year, month_number)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.get("/categories")
|
|
def categories(month: str, db: Db, context: Context) -> list[dict]:
|
|
year, month_number = parse_month(month)
|
|
period = period_for(db, context, year, month_number)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
rows = state["monthly_categories"] + state["envelopes"]
|
|
category_by_id = {
|
|
str(item.id): item
|
|
for item in service.list_categories(db, space_id=context.space.id, include_archived=True)
|
|
}
|
|
result = []
|
|
for row in rows:
|
|
category = category_by_id[row["category_id"]]
|
|
result.append(
|
|
{
|
|
"id": row["category_id"],
|
|
"name": row["name"],
|
|
"nameKey": row["name_key"],
|
|
"kind": category.kind,
|
|
"system": row["system"],
|
|
"archived": category.archived,
|
|
"creditAccountId": category.linked_account_id,
|
|
"plannedKopecks": row["planned"],
|
|
"fundedKopecks": row["funded"],
|
|
"spentKopecks": row["spent"],
|
|
"availableKopecks": row["available"],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def category_definition_json(category, mcc_codes: list[int]) -> dict:
|
|
return {
|
|
"id": category.id,
|
|
"name": categorization.resolve_category_name(category),
|
|
"nameKey": categorization.resolve_category_name_key(category),
|
|
"kind": category.kind,
|
|
"systemKey": category.system_key,
|
|
"linkedAccountId": category.linked_account_id,
|
|
"archived": category.archived,
|
|
"mccCodes": mcc_codes,
|
|
}
|
|
|
|
|
|
@router.get("/category-definitions")
|
|
def category_definitions(
|
|
db: Db,
|
|
context: Context,
|
|
include_archived: Annotated[bool, Query(alias="includeArchived")] = False,
|
|
) -> list[dict]:
|
|
items = service.list_categories(
|
|
db, space_id=context.space.id, include_archived=include_archived
|
|
)
|
|
codes_by_category: dict[uuid.UUID, list[int]] = defaultdict(list)
|
|
for category_id, code in (
|
|
db.query(CategoryMcc.category_id, CategoryMcc.mcc_code)
|
|
.filter_by(space_id=context.space.id)
|
|
.order_by(CategoryMcc.mcc_code)
|
|
):
|
|
codes_by_category[category_id].append(code)
|
|
return [category_definition_json(item, codes_by_category[item.id]) for item in items]
|
|
|
|
|
|
class CategoryDefinitionPatchBody(BaseModel):
|
|
name: str | None = Field(default=None, min_length=1, max_length=200)
|
|
archived: bool | None = None
|
|
mcc_codes: list[int] | None = Field(default=None, alias="mccCodes")
|
|
|
|
|
|
@router.patch("/category-definitions/{category_id}")
|
|
def patch_category_definition(
|
|
category_id: uuid.UUID,
|
|
body: CategoryDefinitionPatchBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
category = service.get_category(db, space_id=context.space.id, category_id=category_id)
|
|
if body.name is not None or body.archived is not None:
|
|
category = service.update_category_definition(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
category_id=category_id,
|
|
name=body.name,
|
|
archived=body.archived,
|
|
)
|
|
if body.mcc_codes is not None:
|
|
categorization.set_category_mcc_codes(
|
|
db,
|
|
space_id=context.space.id,
|
|
category_id=category_id,
|
|
codes=body.mcc_codes,
|
|
)
|
|
return category_definition_json(
|
|
category,
|
|
categorization.get_category_mcc_codes(
|
|
db, space_id=context.space.id, category_id=category.id
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/categories", status_code=201)
|
|
def create_category(body: CategoryBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
category = service.create_category(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
kind=body.kind,
|
|
name=body.name,
|
|
)
|
|
if body.mcc_codes:
|
|
categorization.set_category_mcc_codes(
|
|
db,
|
|
space_id=context.space.id,
|
|
category_id=category.id,
|
|
codes=body.mcc_codes,
|
|
)
|
|
return category_definition_json(
|
|
category,
|
|
categorization.get_category_mcc_codes(
|
|
db, space_id=context.space.id, category_id=category.id
|
|
),
|
|
)
|
|
|
|
|
|
@router.patch("/categories/{category_id}")
|
|
def update_plan(
|
|
category_id: uuid.UUID, body: PlanBody, db: Db, context: Context, _csrf: Csrf
|
|
) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.set_assignments(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
items=[{"category_id": category_id, "planned_kopecks": body.plannedKopecks}],
|
|
)
|
|
return next(item for item in categories(body.month, db, context) if item["id"] == category_id)
|
|
|
|
|
|
@router.post("/categories/{category_id}/fund")
|
|
def fund_category(
|
|
category_id: uuid.UUID,
|
|
body: CategoryFundingBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.fund(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
category_id=category_id,
|
|
amount_kopecks=body.amountKopecks,
|
|
kind="fund",
|
|
)
|
|
return next(item for item in categories(body.month, db, context) if item["id"] == category_id)
|
|
|
|
|
|
@router.put("/{year}/{month}/assignments")
|
|
def assignments(
|
|
year: int, month: int, body: AssignmentsBody, db: Db, context: Context, _csrf: Csrf
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
rows = service.set_assignments(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
items=[item.model_dump() for item in body.items],
|
|
)
|
|
return {"count": len(rows)}
|
|
|
|
|
|
@router.post("/{year}/{month}/funding", status_code=201)
|
|
def fund(year: int, month: int, body: FundingBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
movement = service.fund(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
**body.model_dump(),
|
|
)
|
|
return {"id": movement.id}
|
|
|
|
|
|
def expected_income_item(db: Db, context: Context, period, income_id: uuid.UUID) -> dict:
|
|
items = service.expected_income_items(db, space_id=context.space.id, period=period)
|
|
return next(item for item in items if item["id"] == str(income_id))
|
|
|
|
|
|
@router.get("/{year}/{month}/expected-incomes")
|
|
def expected_incomes(year: int, month: int, db: Db, context: Context) -> list[dict]:
|
|
period = period_for(db, context, year, month)
|
|
items = service.expected_income_items(db, space_id=context.space.id, period=period)
|
|
return [expected_income_json(item) for item in items]
|
|
|
|
|
|
@router.post("/{year}/{month}/expected-incomes", status_code=201)
|
|
def create_expected_income(
|
|
year: int,
|
|
month: int,
|
|
body: ExpectedIncomeBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
income = service.add_expected_income(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
**body.model_dump(),
|
|
)
|
|
return expected_income_json(expected_income_item(db, context, period, income.id))
|
|
|
|
|
|
@router.patch("/{year}/{month}/expected-incomes/{income_id}")
|
|
def patch_expected_income(
|
|
year: int,
|
|
month: int,
|
|
income_id: uuid.UUID,
|
|
body: ExpectedIncomePatchBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
service.update_expected_income(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
income_id=income_id,
|
|
changes=body.model_dump(exclude_unset=True),
|
|
)
|
|
return expected_income_json(expected_income_item(db, context, period, income_id))
|
|
|
|
|
|
@router.delete("/{year}/{month}/expected-incomes/{income_id}", status_code=204)
|
|
def delete_expected_income(
|
|
year: int,
|
|
month: int,
|
|
income_id: uuid.UUID,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> Response:
|
|
period = period_for(db, context, year, month)
|
|
service.delete_expected_income(
|
|
db, space_id=context.space.id, user=context.user, period=period, income_id=income_id
|
|
)
|
|
return Response(status_code=204)
|
|
|
|
|
|
@router.post("/{year}/{month}/expected-incomes/{income_id}/match", status_code=201)
|
|
def match_expected_income(
|
|
year: int,
|
|
month: int,
|
|
income_id: uuid.UUID,
|
|
body: ExpectedIncomeMatchBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
service.match_expected_income(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
expected_income_id=income_id,
|
|
operation_id=body.operation_id,
|
|
amount_kopecks=body.amount_kopecks,
|
|
)
|
|
return expected_income_json(expected_income_item(db, context, period, income_id))
|
|
|
|
|
|
@router.delete("/{year}/{month}/expected-incomes/{income_id}/match/{operation_id}", status_code=204)
|
|
def unmatch_expected_income(
|
|
year: int,
|
|
month: int,
|
|
income_id: uuid.UUID,
|
|
operation_id: uuid.UUID,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> Response:
|
|
period = period_for(db, context, year, month)
|
|
service.unmatch_expected_income(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
expected_income_id=income_id,
|
|
operation_id=operation_id,
|
|
)
|
|
return Response(status_code=204)
|
|
|
|
|
|
@router.post("/{year}/{month}/close")
|
|
def close(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
return service.close_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
|
|
|
|
@router.post("/{year}/{month}/reopen")
|
|
def reopen(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
affected = service.reopen_period(
|
|
db, space_id=context.space.id, user=context.user, period=period
|
|
)
|
|
return {"affected_period_ids": [item.id for item in affected]}
|
|
|
|
|
|
@router.post("/{year}/{month}/copy-previous")
|
|
def copy_previous(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
return service.copy_previous_month(
|
|
db, space_id=context.space.id, user=context.user, period=period
|
|
)
|
|
|
|
|
|
@router.post("/close")
|
|
def close_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.close_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.post("/reopen")
|
|
def reopen_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.reopen_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.post("/copy-previous")
|
|
def copy_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.copy_previous_month(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|