Add multi-user automation features and per-user error tracking. - Database migrations: add workflow_configs/workflow_runs (004), app_errors (005), pipeline_configs/pipeline_runs (006), and add user_token_hash to app_errors (007). - Backend: introduce per-request token handling (X-API-Token) via app.api.deps and update many API routes (auth, automations, bank, characters, dashboard, events, exchange, logs) to use user-scoped Artifacts client and character scoping. Auth endpoints no longer store tokens server-side (validate-only); clear is a no-op on server. - New Errors API and services: endpoint to list, filter, resolve, and report errors scoped to the requesting user; add error models, schemas, middleware/error handler and error_service for recording/hashing tokens. - Pipelines & Workflows: add API routers, models, schemas and engine modules (pipeline/worker/coordinator, workflow runner/conditions) and action_executor updates to support workflow/pipeline execution. - Logs: logs endpoint now prefers fetching recent action logs from the game API (with fallback to local DB), supports paging and filtering, and scopes results to the user. - Frontend: add pipeline/workflow builders, lists, progress components and hooks (use-errors, use-pipelines, use-workflows), sentry client config, and updates to API client/constants/types. - Misc: add middleware error handler, various engine strategy tweaks, tests adjusted. Overall this change enables per-user API tokens, scopes DB queries to each user, introduces pipelines/workflows runtime support, and centralizes application error tracking.
146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING
|
|
|
|
from app.engine.pathfinder import Pathfinder
|
|
from app.schemas.game import CharacterSchema
|
|
|
|
if TYPE_CHECKING:
|
|
from app.engine.decision.equipment_optimizer import EquipmentOptimizer
|
|
from app.schemas.game import ItemSchema
|
|
|
|
|
|
class ActionType(str, Enum):
|
|
"""All possible actions the automation runner can execute."""
|
|
|
|
MOVE = "move"
|
|
FIGHT = "fight"
|
|
GATHER = "gather"
|
|
REST = "rest"
|
|
EQUIP = "equip"
|
|
UNEQUIP = "unequip"
|
|
USE_ITEM = "use_item"
|
|
DEPOSIT_ITEM = "deposit_item"
|
|
WITHDRAW_ITEM = "withdraw_item"
|
|
CRAFT = "craft"
|
|
RECYCLE = "recycle"
|
|
GE_BUY = "ge_buy"
|
|
GE_CREATE_BUY = "ge_create_buy"
|
|
GE_SELL = "ge_sell"
|
|
GE_FILL = "ge_fill"
|
|
GE_CANCEL = "ge_cancel"
|
|
TASK_NEW = "task_new"
|
|
TASK_TRADE = "task_trade"
|
|
TASK_COMPLETE = "task_complete"
|
|
TASK_EXCHANGE = "task_exchange"
|
|
TASK_CANCEL = "task_cancel"
|
|
DEPOSIT_GOLD = "deposit_gold"
|
|
WITHDRAW_GOLD = "withdraw_gold"
|
|
NPC_BUY = "npc_buy"
|
|
NPC_SELL = "npc_sell"
|
|
IDLE = "idle"
|
|
COMPLETE = "complete"
|
|
|
|
|
|
@dataclass
|
|
class ActionPlan:
|
|
"""A single action to be executed by the runner."""
|
|
|
|
action_type: ActionType
|
|
params: dict = field(default_factory=dict)
|
|
reason: str = ""
|
|
|
|
|
|
class BaseStrategy(ABC):
|
|
"""Abstract base class for all automation strategies.
|
|
|
|
A strategy inspects the current character state and returns an
|
|
:class:`ActionPlan` describing the next action the runner should execute.
|
|
|
|
Subclasses must implement :meth:`next_action` and :meth:`get_state`.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
config: dict,
|
|
pathfinder: Pathfinder,
|
|
equipment_optimizer: EquipmentOptimizer | None = None,
|
|
available_items: list[ItemSchema] | None = None,
|
|
) -> None:
|
|
self.config = config
|
|
self.pathfinder = pathfinder
|
|
self._equipment_optimizer = equipment_optimizer
|
|
self._available_items = available_items or []
|
|
self._auto_equip_checked = False
|
|
|
|
@abstractmethod
|
|
async def next_action(self, character: CharacterSchema) -> ActionPlan:
|
|
"""Determine the next action based on the current character state.
|
|
|
|
Returns an :class:`ActionPlan` for the runner to execute. Returning
|
|
``ActionType.COMPLETE`` signals the runner to stop the automation
|
|
loop gracefully. ``ActionType.IDLE`` causes the runner to skip
|
|
execution and re-evaluate after a short delay.
|
|
"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def get_state(self) -> str:
|
|
"""Return a human-readable label describing the current strategy state.
|
|
|
|
Used for logging and status reporting.
|
|
"""
|
|
...
|
|
|
|
# ------------------------------------------------------------------
|
|
# Shared helpers available to all strategies
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _inventory_used_slots(character: CharacterSchema) -> int:
|
|
"""Count how many inventory slots are currently occupied."""
|
|
return len(character.inventory)
|
|
|
|
@staticmethod
|
|
def _inventory_free_slots(character: CharacterSchema) -> int:
|
|
"""Count how many inventory slots are free."""
|
|
return character.inventory_max_items - len(character.inventory)
|
|
|
|
@staticmethod
|
|
def _hp_percent(character: CharacterSchema) -> float:
|
|
"""Return the character's HP as a percentage of max HP."""
|
|
if character.max_hp == 0:
|
|
return 100.0
|
|
return (character.hp / character.max_hp) * 100.0
|
|
|
|
@staticmethod
|
|
def _is_at(character: CharacterSchema, x: int, y: int) -> bool:
|
|
"""Check whether the character is standing at the given tile."""
|
|
return character.x == x and character.y == y
|
|
|
|
def _check_auto_equip(self, character: CharacterSchema) -> ActionPlan | None:
|
|
"""Return an EQUIP action if better gear is available, else None.
|
|
|
|
Only runs once per strategy lifetime to avoid re-checking every tick.
|
|
"""
|
|
if self._auto_equip_checked:
|
|
return None
|
|
self._auto_equip_checked = True
|
|
|
|
if self._equipment_optimizer is None or not self._available_items:
|
|
return None
|
|
|
|
analysis = self._equipment_optimizer.suggest_equipment(
|
|
character, self._available_items
|
|
)
|
|
if analysis.suggestions:
|
|
best = analysis.suggestions[0]
|
|
return ActionPlan(
|
|
ActionType.EQUIP,
|
|
params={"code": best.suggested_item_code, "slot": best.slot},
|
|
reason=f"Auto-equip: {best.reason}",
|
|
)
|
|
return None
|