217 lines
7.2 KiB
Python
217 lines
7.2 KiB
Python
"""Router exposing crafting recipe endpoints.
|
|
|
|
This is separated from `router.py` to keep modules focused.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Optional, Tuple, Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from pydantic import BaseModel
|
|
|
|
from .database import get_session
|
|
|
|
# Map craft names to their corresponding recipe tables in Postgres.
|
|
ALLOWED_CRAFTS: dict[str, str] = {
|
|
"woodworking": "recipes_woodworking",
|
|
"smithing": "recipes_smithing",
|
|
"alchemy": "recipes_alchemy",
|
|
"bonecraft": "recipes_bonecraft",
|
|
"goldsmithing": "recipes_goldsmithing",
|
|
"clothcraft": "recipes_clothcraft",
|
|
"leathercraft": "recipes_leathercraft",
|
|
"cooking": "recipes_cooking",
|
|
}
|
|
|
|
router = APIRouter(tags=["recipes"])
|
|
|
|
class RecipeUsageSummary(BaseModel):
|
|
craft: str
|
|
id: int
|
|
name: str
|
|
level: int
|
|
|
|
class DesynthRecipe(BaseModel):
|
|
id: int
|
|
craft: str
|
|
cap: Optional[int] | None = None
|
|
item: str
|
|
crystal: str
|
|
ingredients: str # raw text, quantity assumed 1 each
|
|
hq1: Optional[str] | None = None
|
|
hq2: Optional[str] | None = None
|
|
hq3: Optional[str] | None = None
|
|
|
|
class ItemRecipeUsage(BaseModel):
|
|
crafted: list[RecipeUsageSummary] = []
|
|
ingredient: list[RecipeUsageSummary] = []
|
|
|
|
|
|
class RecipeDetail(BaseModel):
|
|
id: int
|
|
name: str
|
|
level: int
|
|
category: str
|
|
crystal: str
|
|
key_item: Optional[str] | None = None
|
|
ingredients: List[Tuple[str, int]]
|
|
hq_yields: List[Optional[Tuple[str, int]]]
|
|
subcrafts: List[Tuple[str, int]]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
def _craft_table(craft: str) -> str:
|
|
craft_lower = craft.lower()
|
|
if craft_lower not in ALLOWED_CRAFTS:
|
|
raise HTTPException(status_code=404, detail="Unknown craft")
|
|
return ALLOWED_CRAFTS[craft_lower]
|
|
|
|
|
|
@router.get("/recipes/desynthesis", response_model=List[DesynthRecipe])
|
|
async def list_desynth_recipes(session: AsyncSession = Depends(get_session)):
|
|
q = text("SELECT * FROM recipes_desynthesis ORDER BY item")
|
|
result = await session.execute(q)
|
|
rows = result.fetchall()
|
|
out: list[DesynthRecipe] = []
|
|
for r in rows:
|
|
out.append(
|
|
DesynthRecipe(
|
|
id=r.id,
|
|
craft=r.craft,
|
|
cap=r.cap,
|
|
item=r.item,
|
|
crystal=r.crystal,
|
|
ingredients=r.ingredients,
|
|
hq1=r.hq1,
|
|
hq2=r.hq2,
|
|
hq3=r.hq3,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
@router.get("/recipes/{craft}", response_model=List[RecipeDetail])
|
|
async def list_recipes(
|
|
craft: str = Path(..., description="Craft name, e.g. woodworking"),
|
|
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
"""Return full list of recipes for a craft."""
|
|
table = _craft_table(craft)
|
|
q = text(f"SELECT * FROM {table} ORDER BY level, name")
|
|
result = await session.execute(q)
|
|
rows = result.fetchall()
|
|
print("[DEBUG] list_recipes", table, len(rows))
|
|
details: List[RecipeDetail] = []
|
|
for r in rows:
|
|
details.append(
|
|
RecipeDetail(
|
|
id=r.id,
|
|
name=r.name,
|
|
level=r.level,
|
|
category=r.category,
|
|
crystal=r.crystal,
|
|
key_item=r.key_item,
|
|
ingredients=_to_list_tuples(r.ingredients),
|
|
hq_yields=[tuple(h) if h else None for h in r.hq_yields] if r.hq_yields else [],
|
|
subcrafts=_to_list_tuples(r.subcrafts),
|
|
)
|
|
)
|
|
return details
|
|
|
|
|
|
def _to_list_tuples(value: Any) -> List[Tuple[str, int]]:
|
|
"""Convert JSON/array from Postgres to List[Tuple[str, int]]."""
|
|
if not value:
|
|
return []
|
|
# asyncpg already converts jsonb to Python lists/dicts
|
|
formatted: List[Tuple[str, int]] = []
|
|
for item in value:
|
|
# Accept [name, qty] or {"name": n, "qty": q}
|
|
if isinstance(item, (list, tuple)) and len(item) == 2:
|
|
name, qty = item
|
|
elif isinstance(item, dict):
|
|
name = item.get("0") or item.get("name") or item.get("item")
|
|
qty = item.get("1") or item.get("qty") or item.get("quantity")
|
|
else:
|
|
# Fallback: treat as string name, qty=1
|
|
name, qty = str(item), 1
|
|
if name is None or qty is None:
|
|
continue
|
|
try:
|
|
qty_int = int(qty)
|
|
except Exception:
|
|
qty_int = 1
|
|
formatted.append((str(name), qty_int))
|
|
return formatted
|
|
|
|
|
|
@router.get("/recipes/usage/{item_name}", response_model=ItemRecipeUsage)
|
|
async def item_recipe_usage(item_name: str, session: AsyncSession = Depends(get_session)):
|
|
"""Return lists of recipes that craft `item_name` or use it as an ingredient (excluding crystals)."""
|
|
if item_name.lower().endswith(" crystal"):
|
|
return ItemRecipeUsage()
|
|
|
|
crafted: list[RecipeUsageSummary] = []
|
|
ingredient: list[RecipeUsageSummary] = []
|
|
|
|
for craft, table in ALLOWED_CRAFTS.items():
|
|
# Crafted results
|
|
q1 = text(f"SELECT id, name, level FROM {table} WHERE name = :n LIMIT 50")
|
|
res1 = await session.execute(q1, {"n": item_name})
|
|
crafted.extend(
|
|
[RecipeUsageSummary(craft=craft, id=r.id, name=r.name, level=r.level) for r in res1.fetchall()]
|
|
)
|
|
|
|
# As ingredient (simple text match in JSON/array column)
|
|
# Match exact ingredient name by looking for the item quoted in the JSON text.
|
|
# Using the surrounding double quotes prevents partial matches, e.g. "Chestnut" will not
|
|
# match the ingredient string "Chestnut Lumber".
|
|
quoted = f'"{item_name}"'
|
|
q2 = text(
|
|
f"SELECT id, name, level FROM {table} "
|
|
f"WHERE ingredients::text ILIKE :pat LIMIT 50"
|
|
)
|
|
res2 = await session.execute(q2, {"pat": f"%{quoted}%"})
|
|
for r in res2.fetchall():
|
|
if not any(c.id == r.id and c.craft == craft for c in crafted) and not any(
|
|
i.id == r.id and i.craft == craft for i in ingredient
|
|
):
|
|
ingredient.append(
|
|
RecipeUsageSummary(craft=craft, id=r.id, name=r.name, level=r.level)
|
|
)
|
|
|
|
return ItemRecipeUsage(crafted=crafted, ingredient=ingredient)
|
|
|
|
|
|
@router.get("/recipes/{craft}/{recipe_id}", response_model=RecipeDetail)
|
|
async def recipe_detail(
|
|
craft: str = Path(..., description="Craft name"),
|
|
recipe_id: int = Path(..., description="Recipe ID"),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
"""Return full recipe record."""
|
|
table = _craft_table(craft)
|
|
q = text(f"SELECT * FROM {table} WHERE id = :id LIMIT 1")
|
|
result = await session.execute(q, {"id": recipe_id})
|
|
row = result.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Recipe not found")
|
|
print("[DEBUG] recipe_detail", craft, recipe_id)
|
|
|
|
return RecipeDetail(
|
|
id=row.id,
|
|
name=row.name,
|
|
level=row.level,
|
|
category=row.category,
|
|
crystal=row.crystal,
|
|
key_item=row.key_item,
|
|
ingredients=_to_list_tuples(row.ingredients),
|
|
hq_yields=[tuple(h) if h else None for h in row.hq_yields] if row.hq_yields else [],
|
|
subcrafts=_to_list_tuples(row.subcrafts),
|
|
)
|