Ty
732310fdbe
- backend/pricing.py: /admin/pricing CRUD (one flat $/bdft per species leaf, SETTINGS-35).
- backend/review.py: POST /slabs/{id}/review -> deterministic title + short description
(7 leaf patterns, length ft/in split, thickness 1/8-in precision, METADATA-80/85/90),
price suggestion (bdft x species rate, REVIEW-25), server-derived ready + missing set
(REVIEW-5).
- frontend: create/review/Review.svelte (slab id, primary category, price override,
ready state + generated copy) + dev toggle.
- tests: 3 pricing + 4 review tests (title no-dims, ready set, price suggestion,
knife-blank dims).
Verified: pytest 43 passed; gates exit 0; npm build ok.
100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
"""Species pricing rules (SETTINGS-35): one flat $/bdft per synced species leaf, no tiers.
|
|
|
|
Cites SETTINGS-35, REVIEW-25, API.md §9.
|
|
"""
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from .db import get_connection
|
|
|
|
router = APIRouter(prefix="/api/v1/admin/pricing")
|
|
|
|
|
|
def _error(code: str, message: str, detail: Any = None) -> dict:
|
|
return {"error": {"code": code, "message": message, "detail": detail}}
|
|
|
|
|
|
def _row_to_rule(row) -> dict:
|
|
d = dict(row)
|
|
d["species_id"] = row["species_id"]
|
|
return d
|
|
|
|
|
|
@router.get("")
|
|
async def list_rules() -> list[dict]:
|
|
conn = get_connection()
|
|
rows = conn.execute(
|
|
"SELECT id, species_id, price_per_bdft, created_at, updated_at FROM pricing_rules ORDER BY species_id"
|
|
).fetchall()
|
|
return [_row_to_rule(r) for r in rows]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_rule(payload: dict[str, Any]) -> dict:
|
|
species_id = payload.get("species_id")
|
|
rate = payload.get("price_per_bdft")
|
|
if species_id is None or rate is None:
|
|
return JSONResponse(status_code=400, content=_error("validation", "species_id and price_per_bdft required"))
|
|
if not isinstance(rate, (int, float)) or rate < 0:
|
|
return JSONResponse(status_code=400, content=_error("validation", "price_per_bdft must be a non-negative number"))
|
|
conn = get_connection()
|
|
if conn.execute("SELECT 1 FROM pricing_rules WHERE species_id=?", (species_id,)).fetchone():
|
|
return JSONResponse(status_code=409, content=_error("duplicate", "rule exists for species", {"species_id": species_id}))
|
|
now = _now()
|
|
conn.execute(
|
|
"INSERT INTO pricing_rules(species_id, price_per_bdft, created_at, updated_at) VALUES (?,?,?,?)",
|
|
(species_id, round(float(rate), 2), now, now),
|
|
)
|
|
conn.commit()
|
|
row = conn.execute("SELECT * FROM pricing_rules WHERE species_id=?", (species_id,)).fetchone()
|
|
return _row_to_rule(row)
|
|
|
|
|
|
@router.put("/{rule_id}")
|
|
async def update_rule(rule_id: int, payload: dict[str, Any]) -> dict:
|
|
conn = get_connection()
|
|
cur = conn.execute("SELECT * FROM pricing_rules WHERE id=?", (rule_id,)).fetchone()
|
|
if cur is None:
|
|
return JSONResponse(status_code=404, content=_error("not_found", "rule not found"))
|
|
rate = payload.get("price_per_bdft")
|
|
if rate is None or not isinstance(rate, (int, float)) or rate < 0:
|
|
return JSONResponse(status_code=400, content=_error("validation", "price_per_bdft must be a non-negative number"))
|
|
species_id = payload.get("species_id", cur["species_id"])
|
|
if species_id != cur["species_id"]:
|
|
if conn.execute("SELECT 1 FROM pricing_rules WHERE species_id=? AND id<>?", (species_id, rule_id)).fetchone():
|
|
return JSONResponse(status_code=409, content=_error("duplicate", "rule exists for species", {"species_id": species_id}))
|
|
conn.execute(
|
|
"UPDATE pricing_rules SET species_id=?, price_per_bdft=?, updated_at=? WHERE id=?",
|
|
(species_id, round(float(rate), 2), _now(), rule_id),
|
|
)
|
|
conn.commit()
|
|
row = conn.execute("SELECT * FROM pricing_rules WHERE id=?", (rule_id,)).fetchone()
|
|
return _row_to_rule(row)
|
|
|
|
|
|
@router.delete("/{rule_id}", status_code=204)
|
|
async def delete_rule(rule_id: int):
|
|
conn = get_connection()
|
|
cur = conn.execute("DELETE FROM pricing_rules WHERE id=?", (rule_id,))
|
|
conn.commit()
|
|
if cur.rowcount == 0:
|
|
return JSONResponse(status_code=404, content=_error("not_found", "rule not found"))
|
|
from fastapi import Response
|
|
|
|
return Response(status_code=204)
|
|
|
|
|
|
def rate_for_species(conn, species_id: int) -> float | None:
|
|
row = conn.execute("SELECT price_per_bdft FROM pricing_rules WHERE species_id=?", (species_id,)).fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
def _now() -> str:
|
|
from datetime import datetime, timezone
|
|
|
|
return datetime.now(timezone.utc).replace(microsecond=0, tzinfo=None).isoformat()
|