49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
from datetime import timedelta, datetime
|
|
from typing import Callable, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session as AlchemySession
|
|
|
|
from .models import Resource, Recipe, ResourceFlow
|
|
|
|
|
|
def resource_needs_update(resource: Resource | None, recipe_info_timeout: Optional[timedelta] = None) -> bool:
|
|
if recipe_info_timeout is None:
|
|
recipe_info_timeout = timedelta(days=30)
|
|
return (
|
|
resource is None
|
|
or resource.recipes_populated_at is None
|
|
or datetime.utcnow() - resource.recipes_populated_at > recipe_info_timeout
|
|
)
|
|
|
|
|
|
def chose_resource(session: AlchemySession, resource_label: str, prompt: Callable) -> Resource | None:
|
|
matching_resources = session.scalars(Resource.by_label(resource_label)).all()
|
|
if len(matching_resources) == 0:
|
|
print("Could not find existing resources matching the search string.. starting wiki search")
|
|
else:
|
|
options = {(idx + 1): str(matching_resources[idx].label) for idx in range(len(matching_resources))}
|
|
options[0] = ""
|
|
selected_res = prompt(
|
|
options=options, text="Chose a resource to continue or 0 to continue with a wiki search", default=1
|
|
)
|
|
if selected_res is not None and selected_res != 0:
|
|
return matching_resources[selected_res - 1]
|
|
return None
|
|
|
|
|
|
def chose_recipe(session: AlchemySession, resource: Resource, prompt: Callable) -> Recipe | None:
|
|
stmt = select(Recipe).join(Recipe.results).filter(ResourceFlow.resource_id == resource.id)
|
|
recipes = session.scalars(stmt).all()
|
|
if len(recipes) == 0:
|
|
print("No recipes found for resource")
|
|
return None
|
|
elif len(recipes) > 1:
|
|
options = {(idx + 1): recipes[idx].describe() for idx in range(len(recipes))}
|
|
user_choice = prompt(options=options, text="Select recipe", default=1)
|
|
if user_choice is None:
|
|
return None
|
|
return recipes[user_choice - 1]
|
|
else:
|
|
return recipes[0]
|